mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-05-03 16:22:49 +00:00
fix: add additional stack checks
This commit is contained in:
99
lib/crewai/tests/events/test_event_context.py
Normal file
99
lib/crewai/tests/events/test_event_context.py
Normal file
@@ -0,0 +1,99 @@
|
||||
"""Tests for event context management."""
|
||||
|
||||
import pytest
|
||||
|
||||
from crewai.events.event_context import (
|
||||
SCOPE_ENDING_EVENTS,
|
||||
SCOPE_STARTING_EVENTS,
|
||||
VALID_EVENT_PAIRS,
|
||||
EmptyStackError,
|
||||
EventPairingError,
|
||||
MismatchBehavior,
|
||||
StackDepthExceededError,
|
||||
_event_context_config,
|
||||
EventContextConfig,
|
||||
get_current_parent_id,
|
||||
get_enclosing_parent_id,
|
||||
handle_empty_pop,
|
||||
handle_mismatch,
|
||||
pop_event_scope,
|
||||
push_event_scope,
|
||||
)
|
||||
|
||||
|
||||
class TestStackOperations:
|
||||
"""Tests for stack push/pop operations."""
|
||||
|
||||
def test_empty_stack_returns_none(self) -> None:
|
||||
assert get_current_parent_id() is None
|
||||
assert get_enclosing_parent_id() is None
|
||||
|
||||
def test_push_and_get_parent(self) -> None:
|
||||
push_event_scope("event-1", "task_started")
|
||||
assert get_current_parent_id() == "event-1"
|
||||
|
||||
def test_nested_push(self) -> None:
|
||||
push_event_scope("event-1", "crew_kickoff_started")
|
||||
push_event_scope("event-2", "task_started")
|
||||
assert get_current_parent_id() == "event-2"
|
||||
assert get_enclosing_parent_id() == "event-1"
|
||||
|
||||
def test_pop_restores_parent(self) -> None:
|
||||
push_event_scope("event-1", "crew_kickoff_started")
|
||||
push_event_scope("event-2", "task_started")
|
||||
popped = pop_event_scope()
|
||||
assert popped == ("event-2", "task_started")
|
||||
assert get_current_parent_id() == "event-1"
|
||||
|
||||
def test_pop_empty_stack_returns_none(self) -> None:
|
||||
assert pop_event_scope() is None
|
||||
|
||||
|
||||
class TestStackDepthLimit:
|
||||
"""Tests for stack depth limit."""
|
||||
|
||||
def test_depth_limit_exceeded_raises(self) -> None:
|
||||
_event_context_config.set(EventContextConfig(max_stack_depth=3))
|
||||
|
||||
push_event_scope("event-1", "type-1")
|
||||
push_event_scope("event-2", "type-2")
|
||||
push_event_scope("event-3", "type-3")
|
||||
|
||||
with pytest.raises(StackDepthExceededError):
|
||||
push_event_scope("event-4", "type-4")
|
||||
|
||||
|
||||
class TestMismatchHandling:
|
||||
"""Tests for mismatch behavior."""
|
||||
|
||||
def test_handle_mismatch_raises_when_configured(self) -> None:
|
||||
_event_context_config.set(
|
||||
EventContextConfig(mismatch_behavior=MismatchBehavior.RAISE)
|
||||
)
|
||||
|
||||
with pytest.raises(EventPairingError):
|
||||
handle_mismatch("task_completed", "llm_call_started", "task_started")
|
||||
|
||||
def test_handle_empty_pop_raises_when_configured(self) -> None:
|
||||
_event_context_config.set(
|
||||
EventContextConfig(empty_pop_behavior=MismatchBehavior.RAISE)
|
||||
)
|
||||
|
||||
with pytest.raises(EmptyStackError):
|
||||
handle_empty_pop("task_completed")
|
||||
|
||||
|
||||
class TestEventTypeSets:
|
||||
"""Tests for event type set completeness."""
|
||||
|
||||
def test_all_ending_events_have_pairs(self) -> None:
|
||||
for ending_event in SCOPE_ENDING_EVENTS:
|
||||
assert ending_event in VALID_EVENT_PAIRS
|
||||
|
||||
def test_all_pairs_reference_starting_events(self) -> None:
|
||||
for ending_event, starting_event in VALID_EVENT_PAIRS.items():
|
||||
assert starting_event in SCOPE_STARTING_EVENTS
|
||||
|
||||
def test_starting_and_ending_are_disjoint(self) -> None:
|
||||
overlap = SCOPE_STARTING_EVENTS & SCOPE_ENDING_EVENTS
|
||||
assert not overlap
|
||||
Reference in New Issue
Block a user