Test task function that returns task name and current context's trace_id
(task_name: str, delay: int)
| 14 | |
| 15 | |
| 16 | def task_with_context(task_name: str, delay: int) -> tuple[str, str | None]: |
| 17 | """Test task function that returns task name and current context's trace_id""" |
| 18 | context = get_current_context() |
| 19 | trace_id = context.trace_id if context else None |
| 20 | logger.info(f"Task {task_name} running with trace_id: {trace_id}") |
| 21 | time.sleep(delay) |
| 22 | return task_name, trace_id |
| 23 | |
| 24 | |
| 25 | def test_context_thread_propagation(): |
nothing calls this directly
no test coverage detected