opentelemetry-python
opentelemetry-python copied to clipboard
TestConcurrentMultiSpanProcessor.test_force_flush_late_by_timeout failure
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