Function test_context_thread_isolation

tests/api/test_thread_context.py:101–146

Test context isolation between different threads

()

def test_context_thread_isolation():
    """Test context isolation between different threads"""
    # Set up main thread context
    main_context = RequestContext(trace_id="isolation-test-trace")
    main_context.test_data = "main thread data"
    set_request_context(main_context)

    results = []

    def thread_task(task_id: str, custom_data: str):
        # Get and maintain reference to context in child thread
        context = get_current_context()
        if context:
            # Modify context data
            context.test_data = custom_data
            # Re-set context to make modifications take effect
            set_request_context(context)

        # Get modified context data
        current_context = get_current_context()
        results.append(
            {
                "task_id": task_id,
                "test_data": current_context.test_data if current_context else None,
            }
        )

    # Create two threads with different data
    thread1 = ContextThread(target=thread_task, args=("thread1", "thread1 data"))
    thread2 = ContextThread(target=thread_task, args=("thread2", "thread2 data"))

    thread1.start()
    thread2.start()
    thread1.join()
    thread2.join()

    # Verify thread isolation
    thread1_result = next(r for r in results if r["task_id"] == "thread1")
    thread2_result = next(r for r in results if r["task_id"] == "thread2")

    assert thread1_result["test_data"] == "thread1 data"
    assert thread2_result["test_data"] == "thread2 data"

    # Verify main thread context wasn't modified by child threads
    main_context_after = get_current_context()
    assert main_context_after.test_data == "main thread data"


def test_context_thread_error_with_context():