test_core_infrastructure.py 文件源码

def test_signals_all_listeners_reports_first_failure(self, collector_cls):
        collector = collector_cls()

        class MyException(Exception):
            pass

        def get_mock_listener():
            m = Mock()
            m.return_value = None
            return m

        first_attached_handler = get_mock_listener()

        def second_handler_that_will_fail(*a, **kw):
            raise MyException('foo')
        last_attached_handler = get_mock_listener()

        listeners = [
            first_attached_handler, second_handler_that_will_fail,
            last_attached_handler]
        for l in listeners:
            results_collected.connect(l)
        try:
            with pytest.raises(MyException) as excinfo:
                with collector:
                    pass
            assert str(excinfo.value).endswith('foo')
            method_name_in_stacktrace = 'second_handler_that_will_fail'
            assert method_name_in_stacktrace in str(excinfo.value)
            assert first_attached_handler.called
            assert last_attached_handler.called
        finally:
            for l in listeners:
                results_collected.disconnect(l)