import py
import pytest
from xdist.slavemanage import SlaveController, unserialize_report
from xdist.remote import serialize_report
import execnet
import marshal

queue = py.builtin._tryimport("queue", "Queue")

WAIT_TIMEOUT = 10.0


def check_marshallable(d):
    try:
        marshal.dumps(d)
    except ValueError:
        py.std.pprint.pprint(d)
        raise ValueError("not marshallable")


class EventCall:
    def __init__(self, eventcall):
        self.name, self.kwargs = eventcall

    def __str__(self):
        return "<EventCall %s(**%s)>" % (self.name, self.kwargs)


class SlaveSetup:
    use_callback = False

    def __init__(self, request, testdir):
        self.request = request
        self.testdir = testdir
        self.events = queue.Queue()

    def setup(self, ):
        self.testdir.chdir()
        # import os ; os.environ['EXECNET_DEBUG'] = "2"
        self.gateway = execnet.makegateway()
        self.config = config = self.testdir.parseconfigure()
        putevent = self.use_callback and self.events.put or None

        class DummyMananger:
            specs = [0, 1]

        self.slp = SlaveController(DummyMananger, self.gateway, config,
                                   putevent)
        self.request.addfinalizer(self.slp.ensure_teardown)
        self.slp.setup()

    def popevent(self, name=None):
        while 1:
            if self.use_callback:
                data = self.events.get(timeout=WAIT_TIMEOUT)
            else:
                data = self.slp.channel.receive(timeout=WAIT_TIMEOUT)
            ev = EventCall(data)
            if name is None or ev.name == name:
                return ev
            print("skipping %s" % (ev, ))

    def sendcommand(self, name, **kwargs):
        self.slp.sendcommand(name, **kwargs)


@pytest.fixture
def slave(request, testdir):
    return SlaveSetup(request, testdir)


@pytest.mark.xfail(reason='#59')
def test_remoteinitconfig(testdir):
    from xdist.remote import remote_initconfig
    config1 = testdir.parseconfig()
    config2 = remote_initconfig(config1.option.__dict__, config1.args)
    assert config2.option.__dict__ == config1.option.__dict__
    assert config2.pluginmanager.getplugin("terminal") in (-1, None)


class TestReportSerialization:
    def test_xdist_longrepr_to_str_issue_241(self, testdir):
        testdir.makepyfile("""
            import os
            def test_a(): assert False
            def test_b(): pass
        """)
        testdir.makeconftest("""
            def pytest_runtest_logreport(report):
                print(report.longrepr)
        """)
        res = testdir.runpytest('-n1', '-s')
        res.stdout.fnmatch_lines([
            '*1 failed, 1 passed *'
        ])

    def test_xdist_report_longrepr_reprcrash_130(self, testdir):
        reprec = testdir.inline_runsource("""
                    import py
                    def test_fail(): assert False, 'Expected Message'
                """)
        reports = reprec.getreports("pytest_runtest_logreport")
        assert len(reports) == 3
        rep = reports[1]
        added_section = ('Failure Metadata', str("metadata metadata"), "*")
        rep.longrepr.sections.append(added_section)
        d = serialize_report(rep)
        check_marshallable(d)
        a = unserialize_report("testreport", d)
        # Check assembled == rep
        assert a.__dict__.keys() == rep.__dict__.keys()
        for key in rep.__dict__.keys():
            if key != 'longrepr':
                assert getattr(a, key) == getattr(rep, key)
        assert rep.longrepr.reprcrash.lineno == a.longrepr.reprcrash.lineno
        assert rep.longrepr.reprcrash.message == a.longrepr.reprcrash.message
        assert rep.longrepr.reprcrash.path == a.longrepr.reprcrash.path
        assert rep.longrepr.reprtraceback.entrysep \
            == a.longrepr.reprtraceback.entrysep
        assert rep.longrepr.reprtraceback.extraline \
            == a.longrepr.reprtraceback.extraline
        assert rep.longrepr.reprtraceback.style \
            == a.longrepr.reprtraceback.style
        assert rep.longrepr.sections == a.longrepr.sections
        # Missing section attribute PR171
        assert added_section in a.longrepr.sections

    def test_reprentries_serialization_170(self, testdir):
        from _pytest._code.code import ReprEntry
        reprec = testdir.inline_runsource("""
                            def test_repr_entry():
                                x = 0
                                assert x
                        """, '--showlocals')
        reports = reprec.getreports("pytest_runtest_logreport")
        assert len(reports) == 3
        rep = reports[1]
        d = serialize_report(rep)
        a = unserialize_report("testreport", d)

        rep_entries = rep.longrepr.reprtraceback.reprentries
        a_entries = a.longrepr.reprtraceback.reprentries
        for i in range(len(a_entries)):
            assert isinstance(rep_entries[i], ReprEntry)
            assert rep_entries[i].lines == a_entries[i].lines
            assert rep_entries[i].localssep == a_entries[i].localssep
            assert rep_entries[i].reprfileloc.lineno == a_entries[i].reprfileloc.lineno
            assert rep_entries[i].reprfileloc.message == a_entries[i].reprfileloc.message
            assert rep_entries[i].reprfileloc.path == a_entries[i].reprfileloc.path
            assert rep_entries[i].reprfuncargs.args == a_entries[i].reprfuncargs.args
            assert rep_entries[i].reprlocals.lines == a_entries[i].reprlocals.lines
            assert rep_entries[i].style == a_entries[i].style

    def test_reprentries_serialization_196(self, testdir):
        from _pytest._code.code import ReprEntryNative
        reprec = testdir.inline_runsource("""
                            def test_repr_entry_native():
                                x = 0
                                assert x
                        """, '--tb=native')
        reports = reprec.getreports("pytest_runtest_logreport")
        assert len(reports) == 3
        rep = reports[1]
        d = serialize_report(rep)
        a = unserialize_report("testreport", d)

        rep_entries = rep.longrepr.reprtraceback.reprentries
        a_entries = a.longrepr.reprtraceback.reprentries
        for i in range(len(a_entries)):
            assert isinstance(rep_entries[i], ReprEntryNative)
            assert rep_entries[i].lines == a_entries[i].lines

    def test_itemreport_outcomes(self, testdir):
        reprec = testdir.inline_runsource("""
            import py
            def test_pass(): pass
            def test_fail(): 0/0
            @py.test.mark.skipif("True")
            def test_skip(): pass
            def test_skip_imperative():
                py.test.skip("hello")
            @py.test.mark.xfail("True")
            def test_xfail(): 0/0
            def test_xfail_imperative():
                py.test.xfail("hello")
        """)
        reports = reprec.getreports("pytest_runtest_logreport")
        assert len(reports) == 17  # with setup/teardown "passed" reports
        for rep in reports:
            d = serialize_report(rep)
            check_marshallable(d)
            newrep = unserialize_report("testreport", d)
            assert newrep.passed == rep.passed
            assert newrep.failed == rep.failed
            assert newrep.skipped == rep.skipped
            if newrep.skipped and not hasattr(newrep, "wasxfail"):
                assert len(newrep.longrepr) == 3
            assert newrep.outcome == rep.outcome
            assert newrep.when == rep.when
            assert newrep.keywords == rep.keywords
            if rep.failed:
                assert newrep.longreprtext == rep.longreprtext

    def test_collectreport_passed(self, testdir):
        reprec = testdir.inline_runsource("def test_func(): pass")
        reports = reprec.getreports("pytest_collectreport")
        for rep in reports:
            d = serialize_report(rep)
            check_marshallable(d)
            newrep = unserialize_report("collectreport", d)
            assert newrep.passed == rep.passed
            assert newrep.failed == rep.failed
            assert newrep.skipped == rep.skipped

    def test_collectreport_fail(self, testdir):
        reprec = testdir.inline_runsource("qwe abc")
        reports = reprec.getreports("pytest_collectreport")
        assert reports
        for rep in reports:
            d = serialize_report(rep)
            check_marshallable(d)
            newrep = unserialize_report("collectreport", d)
            assert newrep.passed == rep.passed
            assert newrep.failed == rep.failed
            assert newrep.skipped == rep.skipped
            if rep.failed:
                assert newrep.longrepr == str(rep.longrepr)

    def test_extended_report_deserialization(self, testdir):
        reprec = testdir.inline_runsource("qwe abc")
        reports = reprec.getreports("pytest_collectreport")
        assert reports
        for rep in reports:
            rep.extra = True
            d = serialize_report(rep)
            check_marshallable(d)
            newrep = unserialize_report("collectreport", d)
            assert newrep.extra
            assert newrep.passed == rep.passed
            assert newrep.failed == rep.failed
            assert newrep.skipped == rep.skipped
            if rep.failed:
                assert newrep.longrepr == str(rep.longrepr)


class TestSlaveInteractor:
    def test_basic_collect_and_runtests(self, slave):
        slave.testdir.makepyfile("""
            def test_func():
                pass
        """)
        slave.setup()
        ev = slave.popevent()
        assert ev.name == "slaveready"
        ev = slave.popevent()
        assert ev.name == "collectionstart"
        assert not ev.kwargs
        ev = slave.popevent("collectionfinish")
        assert ev.kwargs['topdir'] == slave.testdir.tmpdir
        ids = ev.kwargs['ids']
        assert len(ids) == 1
        slave.sendcommand("runtests", indices=list(range(len(ids))))
        slave.sendcommand("shutdown")
        ev = slave.popevent("logstart")
        assert ev.kwargs["nodeid"].endswith("test_func")
        assert len(ev.kwargs["location"]) == 3
        ev = slave.popevent("testreport")  # setup
        ev = slave.popevent("testreport")
        assert ev.name == "testreport"
        rep = unserialize_report(ev.name, ev.kwargs['data'])
        assert rep.nodeid.endswith("::test_func")
        assert rep.passed
        assert rep.when == "call"
        ev = slave.popevent("slavefinished")
        assert 'slaveoutput' in ev.kwargs

    @pytest.mark.skipif(pytest.__version__ >= '3.0',
                        reason='skip at module level illegal in pytest 3.0')
    def test_remote_collect_skip(self, slave):
        slave.testdir.makepyfile("""
            import py
            py.test.skip("hello")
        """)
        slave.setup()
        ev = slave.popevent("collectionstart")
        assert not ev.kwargs
        ev = slave.popevent()
        assert ev.name == "collectreport"
        ev = slave.popevent()
        assert ev.name == "collectreport"
        rep = unserialize_report(ev.name, ev.kwargs['data'])
        assert rep.skipped
        ev = slave.popevent("collectionfinish")
        assert not ev.kwargs['ids']

    def test_remote_collect_fail(self, slave):
        slave.testdir.makepyfile("""aasd qwe""")
        slave.setup()
        ev = slave.popevent("collectionstart")
        assert not ev.kwargs
        ev = slave.popevent()
        assert ev.name == "collectreport"
        ev = slave.popevent()
        assert ev.name == "collectreport"
        rep = unserialize_report(ev.name, ev.kwargs['data'])
        assert rep.failed
        ev = slave.popevent("collectionfinish")
        assert not ev.kwargs['ids']

    def test_runtests_all(self, slave):
        slave.testdir.makepyfile("""
            def test_func(): pass
            def test_func2(): pass
        """)
        slave.setup()
        ev = slave.popevent()
        assert ev.name == "slaveready"
        ev = slave.popevent()
        assert ev.name == "collectionstart"
        assert not ev.kwargs
        ev = slave.popevent("collectionfinish")
        ids = ev.kwargs['ids']
        assert len(ids) == 2
        slave.sendcommand("runtests_all", )
        slave.sendcommand("shutdown", )
        for func in "::test_func", "::test_func2":
            for i in range(3):  # setup/call/teardown
                ev = slave.popevent("testreport")
                assert ev.name == "testreport"
                rep = unserialize_report(ev.name, ev.kwargs['data'])
                assert rep.nodeid.endswith(func)
        ev = slave.popevent("slavefinished")
        assert 'slaveoutput' in ev.kwargs

    def test_happy_run_events_converted(self, testdir, slave):
        py.test.xfail("implement a simple test for event production")
        assert not slave.use_callback
        slave.testdir.makepyfile("""
            def test_func():
                pass
        """)
        slave.setup()
        hookrec = testdir.getreportrecorder(slave.config)
        for data in slave.slp.channel:
            slave.slp.process_from_remote(data)
        slave.slp.process_from_remote(slave.slp.ENDMARK)
        py.std.pprint.pprint(hookrec.hookrecorder.calls)
        hookrec.hookrecorder.contains([
            ("pytest_collectstart", "collector.fspath == aaa"),
            ("pytest_pycollect_makeitem", "name == 'test_func'"),
            ("pytest_collectreport", "report.collector.fspath == aaa"),
            ("pytest_collectstart", "collector.fspath == bbb"),
            ("pytest_pycollect_makeitem", "name == 'test_func'"),
            ("pytest_collectreport", "report.collector.fspath == bbb"),
        ])

    def test_process_from_remote_error_handling(self, slave, capsys):
        slave.use_callback = True
        slave.setup()
        slave.slp.process_from_remote(('<nonono>', ()))
        out, err = capsys.readouterr()
        assert 'INTERNALERROR> ValueError: unknown event: <nonono>' in out
        ev = slave.popevent()
        assert ev.name == "errordown"


def test_remote_env_vars(testdir):
    testdir.makepyfile('''
        import os
        def test():
            assert os.environ['PYTEST_XDIST_WORKER'] in ('gw0', 'gw1')
            assert os.environ['PYTEST_XDIST_WORKER_COUNT'] == '2'
    ''')
    result = testdir.runpytest('-n2', '--max-worker-restart=0')
    assert result.ret == 0