opentelemetry-python icon indicating copy to clipboard operation
opentelemetry-python copied to clipboard

TestConcurrentMultiSpanProcessor.test_force_flush_late_by_timeout failure

Open xrmx opened this issue 8 months ago • 0 comments

With python 3.8 on windows on a branch on top of e5a93079d173c5385d37085db6339d3bccc54736:

______ TestConcurrentMultiSpanProcessor.test_force_flush_late_by_timeout ______

self = <tests.trace.test_span_processor.TestConcurrentMultiSpanProcessor testMethod=test_force_flush_late_by_timeout>

    @mark.skipif(
        python_implementation() == "PyPy" and system() == "Windows",
        reason="This test randomly fails in Windows with PyPy",
    )
    def test_force_flush_late_by_timeout(self):
        multi_processor = trace.ConcurrentMultiSpanProcessor(5)
        wait_event = Event()
    
        def delayed_flush(_):
            wait_event.wait()
    
        late_mock = mock.Mock(spec=trace.SpanProcessor)
        late_mock.force_flush = mock.Mock(side_effect=delayed_flush)
        mocks = [mock.Mock(spec=trace.SpanProcessor) for _ in range(0, 4)]
        mocks.insert(0, late_mock)
    
        for mock_processor in mocks:
            multi_processor.add_span_processor(mock_processor)
    
        flushed = multi_processor.force_flush(timeout_millis=10)
        # let the thread executing the late_mock continue
        wait_event.set()
    
        self.assertFalse(flushed)
        for mock_processor in mocks:
>           self.assertEqual(1, mock_processor.force_flush.call_count)
E           AssertionError: 1 != 0

xrmx avatar Apr 15 '25 09:04 xrmx