fix: patch event-bus singleton in LLM emit unit tests

Class-level CrewAIEventsBus.emit patches are unreliable under
pytest --import-mode=importlib / xdist. Patch the singleton instance
instead so streaming finish-reason and usage-event tests observe emits.

Co-authored-by: Rip&Tear <theCyberTech@users.noreply.github.com>
This commit is contained in:
Cursor Agent
2026-08-05 11:15:20 +00:00
parent 4b4195e8da
commit fa33af4595
2 changed files with 19 additions and 6 deletions

View File

@@ -4,7 +4,7 @@ from unittest.mock import patch
import pytest
from pydantic import BaseModel
from crewai.events.event_bus import CrewAIEventsBus
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.llm_events import LLMCallCompletedEvent, LLMCallType
from crewai.llm import LLM
from crewai.llms.base_llm import BaseLLM
@@ -203,7 +203,9 @@ class _StubLLM(BaseLLM):
class TestEmitCallCompletedEventPassesUsage:
@pytest.fixture
def mock_emit(self):
with patch.object(CrewAIEventsBus, "emit") as mock:
# Patch the singleton instance; class-level patches are unreliable
# under pytest ``--import-mode=importlib`` / xdist.
with patch.object(crewai_event_bus, "emit") as mock:
yield mock
@pytest.fixture

View File

@@ -11,22 +11,33 @@ from unittest.mock import patch
import pytest
from crewai.events.event_bus import CrewAIEventsBus
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.llm_events import LLMCallCompletedEvent
from crewai.llm import LLM
@pytest.fixture
def mock_emit():
with patch.object(CrewAIEventsBus, "emit") as mock:
# Patch the singleton instance (not the class). Class-level patches are
# unreliable under pytest ``--import-mode=importlib`` / xdist because the
# test and ``crewai.llm`` can observe different class objects.
with patch.object(crewai_event_bus, "emit") as mock:
yield mock
def _event_from_call(call) -> object | None:
if "event" in call.kwargs:
return call.kwargs["event"]
if len(call.args) >= 2:
return call.args[1]
return None
def _completed_event(mock_emit) -> LLMCallCompletedEvent:
matches = [
call.kwargs["event"]
event
for call in mock_emit.call_args_list
if isinstance(call.kwargs.get("event"), LLMCallCompletedEvent)
if isinstance((event := _event_from_call(call)), LLMCallCompletedEvent)
]
assert matches, "expected an LLMCallCompletedEvent to be emitted"
assert len(matches) == 1, f"expected one completed event, got {len(matches)}"