0

我在测试中积极使用 record_xml_property 夹具，但不幸的是它不适用于 xdsit 插件。我试图实施一种解决方法。我没有直接向记者添加属性，而是将它们放到从输出中：

@pytest.fixture
def record_xml_property(request):
    xml = getattr(request.config, "_xml", None)

    if xml is not None:
        node_reporter = xml.node_reporter(request.node.nodeid)
        return node_reporter.add_property

    else:
        def add_property_xdist(name, value):
            request.node.config.slaveoutput.update({'properties': {name: value}})

        return add_property_xdist

之后，我想将此属性添加到pytest_runtest_logreport钩子中的报告中：

@pytest.hookimpl(tryfirst=True)
def pytest_runtest_logreport(report):
    if report.when != 'teardown':
        return

    node = getattr(report, 'node', None)

    if not node:
        return

    xml = getattr(node.config, '_xml', None)

    if not xml:
        return

    node_reporter = xml.node_reporter(report)
    slaveoutput = getattr(node, 'slaveoutput', None)

    if not slaveoutput:
        return

    node_properties = slaveoutput.get('properties', {})

    for key, value in node_properties.items():
        node_reporter.add_property(key, value)

但问题是在执行钩子SlaveController时，slaveoutput 通常在对象中不可用。pytest_runtest_logreport它在 中可用pytest_testnodedown，SlaveInteractor与"slavefinished"事件一起发送，但此时报告已经完成。有什么方法可以更早地从从节点获取从节点输出？

4

1 回答 1

1

最后我决定避免使用slaveoutput. 我正在考虑一些涉及使用共享目录的解决方案，但最后我决定在主节点和从节点之间使用 execnet 通道。不幸的是，xdist 的设计似乎不允许轻松扩展主节点和从节点之间的通信。我不得不做一些猴子补丁、代码重复和其他一些我不喜欢的讨厌的事情。但至少它有效。这是我的解决方案：

@pytest.fixture
def record_xml_property(request):
    """
    Overrides original record_xml_property fixture from junitxml plugin
    """
    xml = getattr(request.config, "_xml", None)

    if xml is not None:
        node_reporter = xml.node_reporter(request.node.nodeid)
        return node_reporter.add_property
    else:
        pluginmanager = request.node.config.pluginmanager
        plugins = pluginmanager.get_plugins()
        slave_interactor = None

        for plugin in plugins:
            if plugin.__class__.__name__ == 'SlaveInteractor':
                slave_interactor = plugin

        if slave_interactor:
            # if test is running in slave node
            def add_property_xdist(name, value):
                slave_interactor.sendevent("record_xml_property", property=(name, value), nodeid=request.node.nodeid)

            return add_property_xdist

    def add_property_noop(name, value):
        pass

    return add_property_noop


def pytest_configure_node(node):
    xml = getattr(node.config, '_xml', None)

    if xml:
        # monkeypatching SlaveController right after it is created
        # process_from_remote is a callback function which is executed on channel event
        original_process_from_remote = node.process_from_remote

        def process_from_remote(self, eventcall):
            try:
                if eventcall != self.ENDMARK:
                    eventname, kwargs = eventcall

                    # In case of record_xml_property event, process it here. Otherwise process it in original
                    # process_from_remote function
                    if eventname == "record_xml_property":
                        self.log("recording xml property %s(%s)" % (eventname, kwargs))

                        # Report is not yet exists. I should use fake one to get proper reporter
                        class Fakereport:
                            def __init__(self, node, nodeid):
                                self.node = node
                                self.nodeid = nodeid

                        node_reporter = xml.node_reporter(Fakereport(self, kwargs['nodeid']))
                        name, value = kwargs['property']
                        node_reporter.add_property(name, value)
                        return
            except KeyboardInterrupt:
                # should not land in receiver-thread
                raise
            except:
                # the same except as in original process_from_remote method
                excinfo = py.code.ExceptionInfo()
                py.builtin.print_("!" * 20, excinfo)
                self.config.pluginmanager.notify_exception(excinfo)

            return original_process_from_remote(eventcall)

        node.process_from_remote = types.MethodType(process_from_remote, node)
于 2017-02-27T14:05:18.190 回答