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>
This commit is contained in:
Devin AI
2025-12-11 04:57:11 +00:00
parent 7d9d96d0db
commit 7d35af3c62
2 changed files with 80 additions and 46 deletions

View File

@@ -210,6 +210,42 @@ class TestSignalEventSerialization:
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)."""
@@ -220,32 +256,22 @@ class TestWindowsCompatibility:
This is a regression test for GitHub issue #4062.
"""
import importlib
import crewai.events.types.system_events as system_events_module
# 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)
# Reload after patching so class definitions see the modified signal module
reloaded = importlib.reload(system_events_module)
isolated = _load_isolated_system_events_module(monkeypatch)
# Import should succeed and enum members should exist
assert hasattr(reloaded.SignalType, "SIGHUP")
assert hasattr(reloaded.SignalType, "SIGTSTP")
assert hasattr(reloaded.SignalType, "SIGCONT")
assert hasattr(isolated.SignalType, "SIGHUP")
assert hasattr(isolated.SignalType, "SIGTSTP")
assert hasattr(isolated.SignalType, "SIGCONT")
# Event classes should still be importable
assert hasattr(reloaded, "SigHupEvent")
assert hasattr(reloaded, "SigTStpEvent")
assert hasattr(reloaded, "SigContEvent")
assert hasattr(isolated, "SigHupEvent")
assert hasattr(isolated, "SigTStpEvent")
assert hasattr(isolated, "SigContEvent")
# Fallback values should be negative (to avoid conflicts with real signals)
assert reloaded.SignalType.SIGHUP < 0
assert reloaded.SignalType.SIGTSTP < 0
assert reloaded.SignalType.SIGCONT < 0
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
@@ -254,24 +280,14 @@ class TestWindowsCompatibility:
This is a regression test for GitHub issue #4062.
"""
import importlib
import crewai.events.types.system_events as system_events_module
# Simulate a Windows-like signal module
monkeypatch.delattr(signal, "SIGHUP", raising=False)
monkeypatch.delattr(signal, "SIGTSTP", raising=False)
monkeypatch.delattr(signal, "SIGCONT", raising=False)
# Reload after patching
reloaded = importlib.reload(system_events_module)
isolated = _load_isolated_system_events_module(monkeypatch)
# Events should be creatable
hup_event = reloaded.SigHupEvent()
hup_event = isolated.SigHupEvent()
assert hup_event.type == "SIGHUP"
tstp_event = reloaded.SigTStpEvent()
tstp_event = isolated.SigTStpEvent()
assert tstp_event.type == "SIGTSTP"
cont_event = reloaded.SigContEvent()
cont_event = isolated.SigContEvent()
assert cont_event.type == "SIGCONT"

View File

@@ -130,10 +130,15 @@ def test_telemetry_register_shutdown_handlers_with_missing_optional_signals(
"""Telemetry shouldn't fail when optional signals are missing (Windows-like).
This is a regression test for GitHub issue #4062.
"""
import importlib
from crewai.telemetry import telemetry as telemetry_module
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")
@@ -143,15 +148,28 @@ def test_telemetry_register_shutdown_handlers_with_missing_optional_signals(
monkeypatch.delattr(signal, "SIGTSTP", raising=False)
monkeypatch.delattr(signal, "SIGCONT", raising=False)
# Reload after patching so the module sees the modified signal module
reloaded = importlib.reload(telemetry_module)
# 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
# Reset the singleton to allow a new instance
reloaded.Telemetry._instance = None
reloaded.Telemetry._lock = threading.Lock()
isolated_module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = isolated_module
try:
spec.loader.exec_module(isolated_module)
# This should not raise an error even with missing signals
telemetry = reloaded.Telemetry()
# Reset the singleton to allow a new instance
isolated_module.Telemetry._instance = None
isolated_module.Telemetry._lock = threading.Lock()
# Telemetry should be disabled (due to env var), but import should succeed
assert telemetry.ready is False
# 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]