import py, sys, os, signal, cStringIO, tempfile

import runner
import pypy

pytest_script = py.path.local(pypy.__file__).dirpath('test_all.py')


def test_busywait():
    class FakeProcess:
        def poll(self):
            if timers[0] >= timers[1]:
                return 42
            return None
    class FakeTime:
        def sleep(self, delay):
            timers[0] += delay
        def time(self):
            timers[2] += 1
            return 12345678.9 + timers[0]
    p = FakeProcess()
    prevtime = runner.time
    try:
        runner.time = FakeTime()
        #
        timers = [0.0, 0.0, 0]
        returncode = runner.busywait(p, 10)
        assert returncode == 42 and 0.0 <= timers[0] <= 1.0
        #
        timers = [0.0, 3.0, 0]
        returncode = runner.busywait(p, 10)
        assert returncode == 42 and 3.0 <= timers[0] <= 5.0 and timers[2] <= 10
        #
        timers = [0.0, 500.0, 0]
        returncode = runner.busywait(p, 1000)
        assert returncode == 42 and 500.0<=timers[0]<=510.0 and timers[2]<=100
        #
        timers = [0.0, 500.0, 0]
        returncode = runner.busywait(p, 100)    # get a timeout
        assert returncode == None and 100.0 <= timers[0] <= 110.0
        #
    finally:
        runner.time = prevtime

def test_should_report_failure():
    should_report_failure = runner.should_report_failure
    assert should_report_failure("")
    assert should_report_failure(". Abc\n. Def\n")
    assert should_report_failure("s Ghi\n")
    assert not should_report_failure(". Abc\nF Def\n")
    assert not should_report_failure(". Abc\nE Def\n")
    assert not should_report_failure(". Abc\nP Def\n")
    assert not should_report_failure("F Def\n. Ghi\n. Jkl\n")



class TestRunHelper(object):
    def pytest_funcarg__out(self, request):
        tmpdir = request.getfuncargvalue('tmpdir')
        return tmpdir.ensure('out')

    def test_run(self, out):
        res = runner.run([sys.executable, "-c", "print 42"], '.', out)
        assert res == 0
        assert out.read() == "42\n"

    def test_error(self, out):
        res = runner.run([sys.executable, "-c", "import sys; sys.exit(3)"], '.', out)
        assert res == 3

    def test_signal(self, out):
        if sys.platform == 'win32':
            py.test.skip("no death by signal on windows")
        res = runner.run([sys.executable, "-c", "import os; os.kill(os.getpid(), 9)"], '.', out)
        assert res == -9

    def test_timeout(self, out):
        res = runner.run([sys.executable, "-c", "while True: pass"], '.', out, timeout=3)
        assert res == -999

    def test_timeout_lock(self, out):
        res = runner.run([sys.executable, "-c", "import threading; l=threading.Lock(); l.acquire(); l.acquire()"], '.', out, timeout=3)
        assert res == -999

    def test_timeout_syscall(self, out):
        res = runner.run([sys.executable, "-c", "import socket; s=s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM); s.bind(('', 0)); s.recv(1000)"], '.', out, timeout=3)
        assert res == -999        

    def test_timeout_success(self, out):
        res = runner.run([sys.executable, "-c", "print 42"], '.',
                         out, timeout=2)
        assert res == 0
        out = out.read()
        assert out == "42\n"        


class TestExecuteTest(object):

    def setup_class(cls):
        cls.real_run = (runner.run,)
        cls.called = []
        cls.exitcode = [0]
        
        def fake_run(args, cwd, out, timeout):
            cls.called = (args, cwd, out, timeout)
            return cls.exitcode[0]
        runner.run = fake_run

    def teardown_class(cls):
        runner.run = cls.real_run[0]

    def test_explicit(self):
        res = runner.execute_test('/wd', 'test_one', 'out', 'LOGFILE',
                                  interp=['INTERP', 'IARG'],
                                  test_driver=['driver', 'darg'],
                                  timeout='secs')

        expected = ['INTERP', 'IARG',
                    'driver', 'darg',
                    '-p', 'resultlog',
                    '--resultlog=LOGFILE',
                    #'--junitxml=LOGFILE.junit',

                    'test_one']

        assert self.called == (expected, '/wd', 'out', 'secs')        
        assert res == 0

    def test_explicit_win32(self):
        res = runner.execute_test('/wd', 'test_one', 'out', 'LOGFILE',
                                  interp=['./INTERP', 'IARG'],
                                  test_driver=['driver', 'darg'],
                                  timeout='secs',
                                  _win32=True
                                  )

        expected = ['/wd' + os.sep + './INTERP', 'IARG',
                    'driver', 'darg',
                    '-p', 'resultlog',
                    '--resultlog=LOGFILE',
                    #'--junitxml=LOGFILE.junit',
                    'test_one']
        assert self.called[0] == expected
        assert self.called == (expected, '/wd', 'out', 'secs')        
        assert res == 0

    def test_error(self):
        self.exitcode[:] = [1]
        res = runner.execute_test('/wd', 'test_one', 'out', 'LOGFILE',
                                  interp=['INTERP', 'IARG'],
                                  test_driver=['driver', 'darg'])
        assert res == 1


        self.exitcode[:] = [-signal.SIGSEGV]
        res = runner.execute_test('/wd', 'test_one', 'out', 'LOGFILE',
                                  interp=['INTERP', 'IARG'],
                                  test_driver=['driver', 'darg'])
        assert res == -signal.SIGSEGV

    def test_interpret_exitcode(self):
        failure, extralog = runner.interpret_exitcode(0, "test_foo")
        assert not failure
        assert extralog == ""

        failure, extralog = runner.interpret_exitcode(1, "test_foo", "")
        assert failure
        assert extralog == """! test_foo
 Exit code 1.
"""

        failure, extralog = runner.interpret_exitcode(1, "test_foo", "F Foo\n")
        assert failure
        assert extralog == "  (somefailed=True in test_foo)\n"

        failure, extralog = runner.interpret_exitcode(2, "test_foo")
        assert failure
        assert extralog == """! test_foo
 Exit code 2.
"""

        failure, extralog = runner.interpret_exitcode(-signal.SIGSEGV,
                                                      "test_foo")
        assert failure
        assert extralog == """! test_foo
 Killed by SIGSEGV.
"""

class RunnerTests(object):
    with_thread = True

    def setup_class(cls):
        cls.real_invoke_in_thread = (runner.invoke_in_thread,)
        if not cls.with_thread:
            runner.invoke_in_thread = lambda func, args: func(*args)
        
        cls.udir = py.path.local.make_numbered_dir(prefix='usession-runner-',
                                              keep=3)
        cls.manydir = cls.udir.join('many').ensure(dir=1)

        cls.udir.join("conftest.py").write("pytest_plugins = 'resultlog'\n")

        def fill_test_dir(test_dir, fromdir='normal'):       
            for p in py.path.local(__file__).dirpath(
                'examples', fromdir).listdir("*.py"):
                p.copy(test_dir.join('test_'+p.basename))


        test_normal_dir0 = cls.manydir.join('one', 'test_normal').ensure(dir=1)
        cls.one_test_dir = cls.manydir.join('one')

        fill_test_dir(test_normal_dir0)
        

        test_normal_dir1 = cls.manydir.join('two', 'test_normal1').ensure(dir=1)
        test_normal_dir2 = cls.manydir.join('two', 'pkg',
                                         'test_normal2').ensure(dir=1)
        cls.two_test_dir = cls.manydir.join('two')

        fill_test_dir(test_normal_dir1)
        fill_test_dir(test_normal_dir2)

        cls.test_stall_dir = cls.udir.join('stall').ensure(dir=1)
        test_stall_dir0 = cls.test_stall_dir.join('zero').ensure(dir=1)
        fill_test_dir(test_stall_dir0, 'stall')

    def teardown_class(cls):
        runner.invoke_in_thread = cls.real_invoke_in_thread[0]

    def test_one_dir(self):
        test_driver = [pytest_script]

        log = cStringIO.StringIO()
        out = cStringIO.StringIO()

        run_param = runner.RunParam(self.one_test_dir)
        run_param.test_driver = test_driver
        run_param.parallel_runs = 3        
        
        res = runner.execute_tests(run_param, ['test_normal'], log, out)

        assert res

        out = out.getvalue()
        assert out
        assert '\r\n' not in out
        assert '\n' in out

        log = log.getvalue()
        assert '\r\n' not in log
        assert '\n' in log        
        log_lines = log.splitlines()

        assert ". test_normal/test_example.py::test_one" in log_lines
        nfailures = 0
        noutcomes = 0
        for line in log_lines:
            if line[0] != ' ':
                noutcomes += 1
                if line[0] != '.':
                    nfailures += 1

        assert noutcomes == 107
        assert nfailures == 6

    def test_one_dir_dry_run(self):
        test_driver = [pytest_script]

        log = cStringIO.StringIO()
        out = cStringIO.StringIO()

        run_param = runner.RunParam(self.one_test_dir)
        run_param.test_driver = test_driver
        run_param.parallel_runs = 3
        run_param.dry_run = True
        
        res = runner.execute_tests(run_param, ['test_normal'], log, out)

        assert not res

        assert log.getvalue() == ""

        out_lines = out.getvalue().splitlines()

        assert len(out_lines) == 5

        assert out_lines[2].startswith("++ starting")
        assert out_lines[4].startswith("run [")
        for line in out_lines[2:]:
            assert "test_normal" in line

    def test_many_dirs(self):
        test_driver = [pytest_script]

        log = cStringIO.StringIO()
        out = cStringIO.StringIO()

        cleanedup = []
        def cleanup(testdir):
            cleanedup.append(testdir)

        run_param = runner.RunParam(self.manydir)
        run_param.test_driver = test_driver
        run_param.parallel_runs = 3
        run_param.cleanup = cleanup

        testdirs = []
        run_param.collect_testdirs(testdirs)
        alltestdirs = testdirs[:]
        
        res = runner.execute_tests(run_param, testdirs, log, out)

        assert res

        assert out.getvalue()

        log_lines = log.getvalue().splitlines()

        nfailures = 0
        noutcomes = 0
        for line in log_lines:
            if line[0] != ' ':
                noutcomes += 1
                if line[0] != '.':
                    nfailures += 1

        assert noutcomes == 3*107
        assert nfailures == 3*6

        assert set(cleanedup) == set(alltestdirs)

    def test_timeout(self):
        test_driver = [pytest_script]

        log = cStringIO.StringIO()
        out = cStringIO.StringIO()

        run_param = runner.RunParam(self.test_stall_dir)
        run_param.test_driver = test_driver
        run_param.parallel_runs = 3
        run_param.timeout = 3

        testdirs = []
        run_param.collect_testdirs(testdirs)
        res = runner.execute_tests(run_param, testdirs, log, out)
        assert res

        log_lines = log.getvalue().splitlines()
        assert log_lines[1] == ' TIMEOUT'

    def test_run_wrong_interp(self):
        log = cStringIO.StringIO()
        out = cStringIO.StringIO()

        run_param = runner.RunParam(self.one_test_dir)
        run_param.interp = ['wrong-interp']
        run_param.parallel_runs = 3

        testdirs = []
        run_param.collect_testdirs(testdirs)
        res = runner.execute_tests(run_param, testdirs, log, out)
        assert res

        log_lines = log.getvalue().splitlines()
        assert log_lines[1] == ' Failed to run interp'

    def test_run_bad_get_test_driver(self):
        test_driver = [pytest_script]
        
        log = cStringIO.StringIO()
        out = cStringIO.StringIO()

        run_param = runner.RunParam(self.one_test_dir)
        run_param.parallel_runs = 3
        def boom(testdir):
            raise RuntimeError("Boom")
        run_param.get_test_driver = boom

        testdirs = []
        run_param.collect_testdirs(testdirs)
        res = runner.execute_tests(run_param, testdirs, log, out)
        assert res

        log_lines = log.getvalue().splitlines()
        assert log_lines[1] == ' Failed with exception in execute-test'


class TestRunnerNoThreads(RunnerTests):
    with_thread = False

    def test_collect_testdirs(self):
        res = []
        seen = []
        run_param = runner.RunParam(self.one_test_dir)
        real_collect_one_testdir = run_param.collect_one_testdir

        def witness_collect_one_testdir(testdirs, reldir, tests):
            seen.append((reldir, sorted(map(str, tests))))
            real_collect_one_testdir(testdirs, reldir, tests)

        run_param.collect_one_testdir = witness_collect_one_testdir
        
        run_param.collect_testdirs(res)

        assert res == ['test_normal']
        assert len(seen) == 1
        reldir, tests = seen[0]
        assert reldir == 'test_normal'
        for test in tests:
            assert test.startswith('test_normal/')

        run_param.collect_one_testdir = real_collect_one_testdir
        res = []
        run_param = runner.RunParam(self.two_test_dir)
        
        run_param.collect_testdirs(res)

        assert sorted(res) == ['pkg/test_normal2', 'test_normal1']        


class TestRunner(RunnerTests):
    pass