Repository: crewAIInc/crewAI
File: lib/crewai/tests/telemetry/test_telemetry.py (Python)

Code before:

    instances = []

    def create_instance():
        instances.append(Telemetry())

    threads = [threading.Thread(target=create_instance) for _ in range(5)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    assert all(instance is telemetry1 for instance in instances)


def test_no_signal_handler_traceback_in_non_main_thread():
    """Signal handler registration should be silently skipped in non-main threads.

    Regression test for https://github.com/crewAIInc/crewAI/issues/4289
    """
    errors: list[Exception] = []

    def init_in_thread():
        try:
            t = Telemetry()
            with patch("signal.signal") as mock_signal:
                t._register_shutdown_handlers()
                assert not mock_signal.called, "signal.signal must not be called from a non-main thread"
        except Exception as exc:
            errors.append(exc)

    thread = threading.Thread(target=init_in_thread)
    thread.start()
    thread.join()

    assert not errors, f"Unexpected error: {errors}"


Review comment (suggestion):
The `patch("signal.signal")` context is opened **after** `Telemetry()` is already constructed. Since `Telemetry.__init__` calls `_register_shutdown_handlers()` internally (which is where the fix lives), the mock is never active during the constructor call — so any accidental `signal.signal` call during construction would go undetected by `mock_signal.called`.

To actually verify that initialization from a non-main thread never calls `signal.signal`, the patch should wrap the `Telemetry()` construction as well, i.e. `t = Telemetry()` should be inside the `with patch("signal.signal") as mock_signal:` block. That way the assertion covers both the constructor-triggered call and the explicit call.
```suggestion
            with patch("signal.signal") as mock_signal:
                t = Telemetry()
```

Code after:

    instances = []

    def create_instance():
        instances.append(Telemetry())

    threads = [threading.Thread(target=create_instance) for _ in range(5)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    assert all(instance is telemetry1 for instance in instances)


def test_no_signal_handler_traceback_in_non_main_thread():
    """Signal handler registration should be silently skipped in non-main threads.

    Regression test for https://github.com/crewAIInc/crewAI/issues/4289
    """
    errors: list[Exception] = []
    mock_holder: dict = {}

    def init_in_thread():
        try:
            Telemetry._instance = None
            with (
                patch.dict(
                    os.environ,
                    {"CREWAI_DISABLE_TELEMETRY": "false", "OTEL_SDK_DISABLED": "false"},
                ),
                patch("crewai.telemetry.telemetry.TracerProvider"),
                patch("signal.signal") as mock_signal,
                patch("crewai.telemetry.telemetry.logger") as mock_logger,
            ):
                Telemetry()
                mock_holder["signal"] = mock_signal
                mock_holder["logger"] = mock_logger
        except Exception as exc:
            errors.append(exc)

    thread = threading.Thread(target=init_in_thread)
    thread.start()
    thread.join()

    assert not errors, f"Unexpected error: {errors}"
    assert mock_holder, "Thread did not execute"
    mock_holder["signal"].assert_not_called()
    mock_holder["logger"].debug.assert_any_call(
        "Skipping signal handler registration: not running in main thread"
    )