""" tests/track_a/test_story_5_04.py Story 5.04 — SwarmRouter: Base Class + Route Registry Black-box tests (BB1–BB5): validate external behaviour via public API White-box tests (WB1–WB3): validate internal structure / branching All external dependencies are fully mocked. No I/O, no network, no database. """ import sys sys.path.insert(0, "/mnt/e/genesis-system") import pytest from datetime import datetime from unittest.mock import MagicMock from core.intent.intent_signal import IntentType, IntentSignal from core.routing.swarm_router import ( SwarmRouter, ROUTE_REGISTRY, CONFIDENCE_THRESHOLD, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_signal( intent_type: IntentType, confidence: float = 0.9, session_id: str = "test-session", utterance: str = "test utterance", ) -> IntentSignal: """Factory that creates a minimal IntentSignal for testing.""" return IntentSignal( session_id=session_id, utterance=utterance, intent_type=intent_type, confidence=confidence, extracted_entities={}, requires_swarm=True, created_at=datetime(2026, 2, 25, 12, 0, 0), raw_gemini_response=None, ) def _make_router(extra_workers: dict | None = None) -> SwarmRouter: """Factory that creates a SwarmRouter with mocked worker instances.""" registry = { "BookingWorker": MagicMock(name="BookingWorker"), "LeadQualificationWorker": MagicMock(name="LeadQualificationWorker"), "FAQWorker": MagicMock(name="FAQWorker"), "EscalationWorker": MagicMock(name="EscalationWorker"), "MemoryCaptureWorker": MagicMock(name="MemoryCaptureWorker"), "GenesisTaskWorker": MagicMock(name="GenesisTaskWorker"), } if extra_workers: registry.update(extra_workers) return SwarmRouter(worker_registry=registry) # --------------------------------------------------------------------------- # BB1: BOOK_JOB signal → get_worker_name() returns "BookingWorker" # --------------------------------------------------------------------------- class TestBB1_BookJobWorkerName: """BB1 — get_worker_name returns 'BookingWorker' for BOOK_JOB intent.""" def test_book_job_returns_booking_worker(self): router = _make_router() signal = _make_signal(IntentType.BOOK_JOB) assert router.get_worker_name(signal) == "BookingWorker" def test_book_job_return_type_is_str(self): router = _make_router() signal = _make_signal(IntentType.BOOK_JOB) result = router.get_worker_name(signal) assert isinstance(result, str) def test_book_job_confidence_not_checked_by_get_worker_name(self): """get_worker_name does NOT gate on confidence — that is can_route's job.""" router = _make_router() signal = _make_signal(IntentType.BOOK_JOB, confidence=0.01) # Should still return the name regardless of confidence assert router.get_worker_name(signal) == "BookingWorker" # --------------------------------------------------------------------------- # BB2: UNKNOWN signal → get_worker_name() returns None # --------------------------------------------------------------------------- class TestBB2_UnknownReturnsNone: """BB2 — get_worker_name returns None for UNKNOWN intent.""" def test_unknown_returns_none(self): router = _make_router() signal = _make_signal(IntentType.UNKNOWN) assert router.get_worker_name(signal) is None def test_unknown_high_confidence_still_returns_none(self): router = _make_router() signal = _make_signal(IntentType.UNKNOWN, confidence=1.0) assert router.get_worker_name(signal) is None # --------------------------------------------------------------------------- # BB3: BOOK_JOB signal, confidence=0.4 → can_route() returns False # --------------------------------------------------------------------------- class TestBB3_LowConfidenceCannotRoute: """BB3 — can_route returns False when confidence is below threshold.""" def test_book_job_low_confidence_false(self): router = _make_router() signal = _make_signal(IntentType.BOOK_JOB, confidence=0.4) assert router.can_route(signal) is False def test_exactly_at_threshold_minus_epsilon_is_false(self): router = _make_router() signal = _make_signal(IntentType.BOOK_JOB, confidence=CONFIDENCE_THRESHOLD - 0.001) assert router.can_route(signal) is False def test_zero_confidence_is_false(self): router = _make_router() signal = _make_signal(IntentType.QUALIFY_LEAD, confidence=0.0) assert router.can_route(signal) is False # --------------------------------------------------------------------------- # BB4: BOOK_JOB signal, confidence=0.8 → can_route() returns True # --------------------------------------------------------------------------- class TestBB4_HighConfidenceCanRoute: """BB4 — can_route returns True when intent is known and confidence is sufficient.""" def test_book_job_high_confidence_true(self): router = _make_router() signal = _make_signal(IntentType.BOOK_JOB, confidence=0.8) assert router.can_route(signal) is True def test_exactly_at_threshold_is_true(self): router = _make_router() signal = _make_signal(IntentType.BOOK_JOB, confidence=CONFIDENCE_THRESHOLD) assert router.can_route(signal) is True def test_full_confidence_is_true(self): router = _make_router() signal = _make_signal(IntentType.ANSWER_FAQ, confidence=1.0) assert router.can_route(signal) is True # --------------------------------------------------------------------------- # BB5: All 6 intent types map to correct worker names # --------------------------------------------------------------------------- class TestBB5_AllIntentsMapped: """BB5 — Every non-UNKNOWN IntentType maps to the expected worker name.""" EXPECTED_MAPPING = { IntentType.BOOK_JOB: "BookingWorker", IntentType.QUALIFY_LEAD: "LeadQualificationWorker", IntentType.ANSWER_FAQ: "FAQWorker", IntentType.ESCALATE_HUMAN: "EscalationWorker", IntentType.CAPTURE_MEMORY: "MemoryCaptureWorker", IntentType.TASK_DISPATCH: "GenesisTaskWorker", } @pytest.mark.parametrize("intent_type,expected_worker", EXPECTED_MAPPING.items()) def test_intent_maps_to_correct_worker(self, intent_type, expected_worker): router = _make_router() signal = _make_signal(intent_type) assert router.get_worker_name(signal) == expected_worker def test_all_six_intents_can_route_at_high_confidence(self): router = _make_router() for intent_type in self.EXPECTED_MAPPING: signal = _make_signal(intent_type, confidence=0.9) assert router.can_route(signal) is True, ( f"Expected can_route to be True for {intent_type}" ) # --------------------------------------------------------------------------- # WB1: ROUTE_REGISTRY has exactly 6 keys # --------------------------------------------------------------------------- class TestWB1_RegistrySize: """WB1 — ROUTE_REGISTRY contains exactly 6 IntentType entries.""" def test_route_registry_has_six_keys(self): assert len(ROUTE_REGISTRY) == 6 def test_route_registry_does_not_contain_unknown(self): assert IntentType.UNKNOWN not in ROUTE_REGISTRY def test_route_registry_keys_are_intent_types(self): for key in ROUTE_REGISTRY: assert isinstance(key, IntentType) def test_route_registry_values_are_strings(self): for value in ROUTE_REGISTRY.values(): assert isinstance(value, str) def test_route_registry_covers_all_non_unknown_intent_types(self): """Every IntentType except UNKNOWN must appear in ROUTE_REGISTRY.""" non_unknown = {it for it in IntentType if it != IntentType.UNKNOWN} assert set(ROUTE_REGISTRY.keys()) == non_unknown # --------------------------------------------------------------------------- # WB2: can_route() checks BOTH intent_type AND confidence # --------------------------------------------------------------------------- class TestWB2_CanRouteDoubleGate: """WB2 — can_route gates on both intent_type and confidence independently.""" def test_unknown_with_high_confidence_is_false(self): """UNKNOWN blocks routing regardless of confidence.""" router = _make_router() signal = _make_signal(IntentType.UNKNOWN, confidence=1.0) assert router.can_route(signal) is False def test_known_intent_with_low_confidence_is_false(self): """Low confidence blocks routing even for known intents.""" router = _make_router() signal = _make_signal(IntentType.BOOK_JOB, confidence=0.59) assert router.can_route(signal) is False def test_known_intent_with_sufficient_confidence_is_true(self): """Both gates pass → can_route is True.""" router = _make_router() signal = _make_signal(IntentType.BOOK_JOB, confidence=0.60) assert router.can_route(signal) is True def test_confidence_threshold_constant_is_0_6(self): assert CONFIDENCE_THRESHOLD == 0.6 # --------------------------------------------------------------------------- # WB3: worker_registry parameter is used (not ROUTE_REGISTRY for dispatch) # --------------------------------------------------------------------------- class TestWB3_WorkerRegistryInjected: """WB3 — SwarmRouter uses the injected worker_registry for live dispatch.""" def test_route_returns_injected_worker_instance(self): """route() must return the exact instance injected, not a new one.""" mock_worker = MagicMock(name="BookingWorkerInstance") router = SwarmRouter(worker_registry={"BookingWorker": mock_worker}) signal = _make_signal(IntentType.BOOK_JOB, confidence=0.9) result = router.route(signal) assert result is mock_worker def test_route_returns_none_when_worker_not_in_registry(self): """Known intent with registered route but no matching worker instance → None.""" router = SwarmRouter(worker_registry={}) # empty — no workers injected signal = _make_signal(IntentType.BOOK_JOB, confidence=0.9) result = router.route(signal) assert result is None def test_route_returns_none_for_low_confidence(self): """route() delegates to can_route, so low confidence yields None.""" mock_worker = MagicMock() router = SwarmRouter(worker_registry={"BookingWorker": mock_worker}) signal = _make_signal(IntentType.BOOK_JOB, confidence=0.3) result = router.route(signal) assert result is None def test_route_returns_none_for_unknown_intent(self): """route() delegates to can_route, so UNKNOWN intent yields None.""" mock_worker = MagicMock() router = SwarmRouter(worker_registry={"UnknownHandler": mock_worker}) signal = _make_signal(IntentType.UNKNOWN, confidence=0.99) result = router.route(signal) assert result is None def test_worker_registry_is_stored_on_instance(self): """The injected registry must be accessible as router.workers.""" registry = {"BookingWorker": MagicMock()} router = SwarmRouter(worker_registry=registry) assert router.workers is registry def test_different_worker_registries_produce_different_instances(self): """Swarm routers with different registries return different workers.""" worker_a = MagicMock(name="WorkerA") worker_b = MagicMock(name="WorkerB") router_a = SwarmRouter(worker_registry={"BookingWorker": worker_a}) router_b = SwarmRouter(worker_registry={"BookingWorker": worker_b}) signal = _make_signal(IntentType.BOOK_JOB, confidence=0.9) assert router_a.route(signal) is worker_a assert router_b.route(signal) is worker_b assert router_a.route(signal) is not router_b.route(signal)