Test if ContextThread correctly propagates context from main thread to child thread
()
| 23 | |
| 24 | |
| 25 | def test_context_thread_propagation(): |
| 26 | """Test if ContextThread correctly propagates context from main thread to child thread""" |
| 27 | # Set up main thread context |
| 28 | main_context = RequestContext(trace_id="main-thread-trace") |
| 29 | main_context.test_data = "test value" # Add extra context data |
| 30 | set_request_context(main_context) |
| 31 | |
| 32 | # Store child thread results |
| 33 | results = {} |
| 34 | |
| 35 | def thread_task(): |
| 36 | # Get context in child thread |
| 37 | child_context = get_current_context() |
| 38 | results["trace_id"] = child_context.trace_id if child_context else None |
| 39 | results["test_data"] = child_context.test_data if child_context else None |
| 40 | |
| 41 | # Create and run child thread |
| 42 | thread = ContextThread(target=thread_task) |
| 43 | thread.start() |
| 44 | thread.join() |
| 45 | |
| 46 | # Verify context propagation |
| 47 | assert results["trace_id"] == "main-thread-trace" |
| 48 | assert results["test_data"] == "test value" |
| 49 | |
| 50 | |
| 51 | def test_context_thread_pool_propagation(): |
nothing calls this directly
no test coverage detected