Source

pytest-gui / pytest_gui / slave.py

from xdist.remote import serialize_report


class SubprocessStream(object):
    def __init__(self, channel, items=None):
        self.channel = channel
        if items is not None:
            self.items = set(items)
        else:
            self.items = None

    def send(self, *k):
        try:
            self.channel.send(k)
        except IOError:
            print self.channel._remoteerrors

    def pytest_collectstart(self, collector):
        self.send('context_start', collector.nodeid)

    def pytest_collectreport(self, report):
        self.send('context_report', serialize_report(report))

    def pytest_itemcollected(self, item):
        self.send('collect_item', item.nodeid, list(item.keywords))

    def pytest_runtest_logstart(self, nodeid):
        self.send('item_start', nodeid)

    def pytest_runtest_logreport(self, report):
        self.send('report', serialize_report(report))

    def pytest_collection_modifyitems(self, config, items):
        if self.items is None:
            return
        new_items = [item for item in items if item.nodeid in self.items]
        items[:] = new_items


def slave_session(channel, args, items, collectonly):
    # close to xdist looponfail
    import sys
    outchannel = channel.gateway.newchannel()
    sys.stdout = sys.stderr = outchannel.makefile('w')
    channel.send(outchannel)

    from pytest_gui.slave import SubprocessStream
    plugin = SubprocessStream(channel, items=items)

    if collectonly:
        #XXX: hack
        args.append('--collectonly')

    import pytest
    pytest.main(args=args, plugins=[plugin])