Compare commits

...

2 Commits

Author SHA1 Message Date
Devin AI
7d35af3c62 fix: use isolated module loading in Windows compatibility tests
The previous approach using importlib.reload() on the canonical module
caused test interference - the reloaded classes were different objects
from the ones imported by other tests, breaking event bus registration.

This fix uses importlib.util to load an isolated copy of the module
under a different name, avoiding pollution of the canonical module.

Co-Authored-By: João <joao@crewai.com>
2025-12-11 04:57:11 +00:00
Devin AI
7d9d96d0db fix: handle missing signals on Windows (SIGHUP, SIGTSTP, SIGCONT)
This fixes GitHub issue #4062 where crewai crashes on Windows with
'module signal has no attribute SIGHUP' error.

Changes:
- Use getattr with fallback values for signals not available on Windows
- Make telemetry signal registration conditional on signal availability
- Add regression tests for Windows compatibility

The fix maintains backward compatibility on Unix systems while allowing
crewai to work on Windows where SIGHUP, SIGTSTP, and SIGCONT are not
available.

Co-Authored-By: João <joao@crewai.com>
2025-12-11 04:50:51 +00:00
4 changed files with 181 additions and 12 deletions

View File

@@ -14,14 +14,25 @@ from pydantic import Field, TypeAdapter
from crewai.events.base_events import BaseEvent
# Fallback values for signals not available on Windows.
# These are negative to avoid conflicts with real signal numbers.
_FALLBACK_SIGHUP = -1
_FALLBACK_SIGTSTP = -2
_FALLBACK_SIGCONT = -3
class SignalType(IntEnum):
"""Enumeration of supported system signals."""
"""Enumeration of supported system signals.
Note: SIGHUP, SIGTSTP, and SIGCONT are not available on Windows.
On Windows, these will use fallback negative values.
"""
SIGTERM = signal.SIGTERM
SIGINT = signal.SIGINT
SIGHUP = signal.SIGHUP
SIGTSTP = signal.SIGTSTP
SIGCONT = signal.SIGCONT
SIGHUP = getattr(signal, "SIGHUP", _FALLBACK_SIGHUP)
SIGTSTP = getattr(signal, "SIGTSTP", _FALLBACK_SIGTSTP)
SIGCONT = getattr(signal, "SIGCONT", _FALLBACK_SIGCONT)
class SigTermEvent(BaseEvent):

View File

@@ -172,11 +172,21 @@ class Telemetry:
self._original_handlers: dict[int, Any] = {}
# Always-supported signals on all platforms
self._register_signal_handler(signal.SIGTERM, SigTermEvent, shutdown=True)
self._register_signal_handler(signal.SIGINT, SigIntEvent, shutdown=True)
self._register_signal_handler(signal.SIGHUP, SigHupEvent, shutdown=False)
self._register_signal_handler(signal.SIGTSTP, SigTStpEvent, shutdown=False)
self._register_signal_handler(signal.SIGCONT, SigContEvent, shutdown=False)
# Optional signals (not available on Windows)
optional_signals: list[tuple[str, type, bool]] = [
("SIGHUP", SigHupEvent, False),
("SIGTSTP", SigTStpEvent, False),
("SIGCONT", SigContEvent, False),
]
for sig_name, event_class, shutdown in optional_signals:
sig = getattr(signal, sig_name, None)
if sig is not None:
self._register_signal_handler(sig, event_class, shutdown=shutdown)
def _register_signal_handler(
self,

View File

@@ -24,12 +24,25 @@ class TestSignalType:
"""Tests for SignalType enum."""
def test_signal_type_values(self) -> None:
"""Verify SignalType maps to correct signal numbers."""
"""Verify SignalType maps to correct signal numbers when available."""
assert SignalType.SIGTERM == signal.SIGTERM
assert SignalType.SIGINT == signal.SIGINT
assert SignalType.SIGHUP == signal.SIGHUP
assert SignalType.SIGTSTP == signal.SIGTSTP
assert SignalType.SIGCONT == signal.SIGCONT
# These signals are not available on Windows, so only check if present
if hasattr(signal, "SIGHUP"):
assert SignalType.SIGHUP == signal.SIGHUP
if hasattr(signal, "SIGTSTP"):
assert SignalType.SIGTSTP == signal.SIGTSTP
if hasattr(signal, "SIGCONT"):
assert SignalType.SIGCONT == signal.SIGCONT
def test_signal_type_enum_members_always_exist(self) -> None:
"""Verify all SignalType enum members exist regardless of platform."""
assert hasattr(SignalType, "SIGTERM")
assert hasattr(SignalType, "SIGINT")
assert hasattr(SignalType, "SIGHUP")
assert hasattr(SignalType, "SIGTSTP")
assert hasattr(SignalType, "SIGCONT")
class TestSignalEvents:
@@ -194,4 +207,87 @@ class TestSignalEventSerialization:
restored = signal_event_adapter.validate_python(serialized)
assert isinstance(restored, SigTermEvent)
assert restored.reason == original.reason
assert restored.type == original.type
assert restored.type == original.type
def _load_isolated_system_events_module(monkeypatch: pytest.MonkeyPatch):
"""Load an isolated copy of system_events module with missing signals.
This avoids reloading the canonical module which would break other tests
that depend on the original class references.
"""
import importlib.util
import pathlib
import sys
import crewai.events.types.system_events as orig_module
# Simulate Windows by removing optional signals
monkeypatch.delattr(signal, "SIGHUP", raising=False)
monkeypatch.delattr(signal, "SIGTSTP", raising=False)
monkeypatch.delattr(signal, "SIGCONT", raising=False)
# Load an isolated copy of the module under a different name
path = pathlib.Path(orig_module.__file__)
spec = importlib.util.spec_from_file_location(
"crewai.events.types.system_events_isolated", path
)
assert spec is not None
assert spec.loader is not None
isolated_module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = isolated_module
try:
spec.loader.exec_module(isolated_module)
finally:
# Clean up to avoid polluting sys.modules
del sys.modules[spec.name]
return isolated_module
class TestWindowsCompatibility:
"""Tests for Windows compatibility (signals not available on Windows)."""
def test_system_events_imports_when_optional_signals_missing(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""system_events should import even if some signals are missing (Windows-like).
This is a regression test for GitHub issue #4062.
"""
isolated = _load_isolated_system_events_module(monkeypatch)
# Import should succeed and enum members should exist
assert hasattr(isolated.SignalType, "SIGHUP")
assert hasattr(isolated.SignalType, "SIGTSTP")
assert hasattr(isolated.SignalType, "SIGCONT")
# Event classes should still be importable
assert hasattr(isolated, "SigHupEvent")
assert hasattr(isolated, "SigTStpEvent")
assert hasattr(isolated, "SigContEvent")
# Fallback values should be negative (to avoid conflicts with real signals)
assert isolated.SignalType.SIGHUP < 0
assert isolated.SignalType.SIGTSTP < 0
assert isolated.SignalType.SIGCONT < 0
def test_signal_events_can_be_created_when_signals_missing(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Signal events should be creatable even when signals are missing.
This is a regression test for GitHub issue #4062.
"""
isolated = _load_isolated_system_events_module(monkeypatch)
# Events should be creatable
hup_event = isolated.SigHupEvent()
assert hup_event.type == "SIGHUP"
tstp_event = isolated.SigTStpEvent()
assert tstp_event.type == "SIGTSTP"
cont_event = isolated.SigContEvent()
assert cont_event.type == "SIGCONT"

View File

@@ -1,4 +1,5 @@
import os
import signal
import threading
from unittest.mock import patch
@@ -121,3 +122,54 @@ def test_telemetry_singleton_pattern():
thread.join()
assert all(instance is telemetry1 for instance in instances)
def test_telemetry_register_shutdown_handlers_with_missing_optional_signals(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Telemetry shouldn't fail when optional signals are missing (Windows-like).
This is a regression test for GitHub issue #4062.
Note: This test uses an isolated module loading approach to avoid
polluting the canonical telemetry module which other tests depend on.
"""
import importlib.util
import pathlib
import sys
from crewai.telemetry import telemetry as orig_telemetry_module
# Disable telemetry to avoid real OTLP setup
monkeypatch.setenv("CREWAI_DISABLE_TELEMETRY", "true")
# Simulate a Windows-like signal module by removing optional signals
monkeypatch.delattr(signal, "SIGHUP", raising=False)
monkeypatch.delattr(signal, "SIGTSTP", raising=False)
monkeypatch.delattr(signal, "SIGCONT", raising=False)
# Load an isolated copy of the telemetry module under a different name
path = pathlib.Path(orig_telemetry_module.__file__)
spec = importlib.util.spec_from_file_location(
"crewai.telemetry.telemetry_isolated", path
)
assert spec is not None
assert spec.loader is not None
isolated_module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = isolated_module
try:
spec.loader.exec_module(isolated_module)
# Reset the singleton to allow a new instance
isolated_module.Telemetry._instance = None
isolated_module.Telemetry._lock = threading.Lock()
# This should not raise an error even with missing signals
telemetry = isolated_module.Telemetry()
# Telemetry should be disabled (due to env var), but import should succeed
assert telemetry.ready is False
finally:
# Clean up to avoid polluting sys.modules
del sys.modules[spec.name]