""" tests/track_a/test_story_5_11.py Story 5.11 — EscalationWorker: Human Handoff Signal Black-box tests (BB1–BB3): validate external behaviour via public API White-box tests (WB1–WB3): validate internal implementation details All external dependencies are fully mocked. Zero real Redis. Zero real filesystem (tmp_path injected in every test). No SQLite. """ import sys sys.path.insert(0, "/mnt/e/genesis-system") import asyncio import json import pytest from datetime import datetime from pathlib import Path from unittest.mock import MagicMock, patch from core.intent.intent_signal import IntentType, IntentSignal from core.workers.escalation_worker import ( EscalationWorker, ESCALATION_TTL, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_intent( session_id: str = "test-session-511", intent_type: IntentType = IntentType.ESCALATE_HUMAN, utterance: str = "I want to speak to a person", ) -> IntentSignal: """Factory: creates a real IntentSignal for ESCALATE_HUMAN intent.""" return IntentSignal( session_id=session_id, utterance=utterance, intent_type=intent_type, confidence=0.95, extracted_entities={}, requires_swarm=True, created_at=datetime(2026, 2, 25, 12, 0, 0), raw_gemini_response=None, ) def _make_redis_client() -> MagicMock: """Factory: mock Redis client with setex() method.""" client = MagicMock(name="RedisClient") client.setex = MagicMock() return client def _make_worker( redis_client: MagicMock = None, tmp_path: Path = None, ) -> EscalationWorker: """Factory: EscalationWorker with mocked Redis and tmp_path for events.""" if redis_client is None: redis_client = _make_redis_client() events_path = (tmp_path / "events.jsonl") if tmp_path is not None else None return EscalationWorker(redis_client=redis_client, events_path=events_path) # --------------------------------------------------------------------------- # BB1: execute() → Redis key set to "requested" # --------------------------------------------------------------------------- class TestBB1_RedisKeySetToRequested: """BB1 — execute() writes 'requested' value to Redis at correct key.""" @pytest.mark.asyncio async def test_setex_called_once(self, tmp_path): redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) redis.setex.assert_called_once() @pytest.mark.asyncio async def test_redis_key_contains_session_id(self, tmp_path): redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent(session_id="sess-abc")) args, _ = redis.setex.call_args assert args[0] == "aiva:escalation:sess-abc" @pytest.mark.asyncio async def test_redis_value_is_requested_string(self, tmp_path): redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) args, _ = redis.setex.call_args assert args[2] == "requested" @pytest.mark.asyncio async def test_redis_value_is_string_not_dict(self, tmp_path): """WB3 requirement: value must be plain string 'requested', not a dict.""" redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) args, _ = redis.setex.call_args stored_value = args[2] assert isinstance(stored_value, str) assert stored_value == "requested" # --------------------------------------------------------------------------- # BB2: Redis key has TTL ≤ 300s # --------------------------------------------------------------------------- class TestBB2_RedisTTL: """BB2 — SETEX is called with TTL equal to ESCALATION_TTL (300).""" @pytest.mark.asyncio async def test_ttl_is_300(self, tmp_path): redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) args, _ = redis.setex.call_args assert args[1] == 300 @pytest.mark.asyncio async def test_ttl_matches_module_constant(self, tmp_path): redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) args, _ = redis.setex.call_args assert args[1] == ESCALATION_TTL @pytest.mark.asyncio async def test_ttl_is_not_zero(self, tmp_path): """TTL must be positive — a TTL of 0 would delete the key immediately.""" redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) args, _ = redis.setex.call_args assert args[1] > 0 @pytest.mark.asyncio async def test_ttl_is_at_most_300(self, tmp_path): """TTL must not exceed 300 seconds (per acceptance criteria).""" redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) args, _ = redis.setex.call_args assert args[1] <= 300 # --------------------------------------------------------------------------- # BB3: Return dict has status="escalation_requested" # --------------------------------------------------------------------------- class TestBB3_ReturnDict: """BB3 — execute() returns the correct status dict.""" @pytest.mark.asyncio async def test_returns_dict(self, tmp_path): worker = _make_worker(tmp_path=tmp_path) result = await worker.execute(_make_intent()) assert isinstance(result, dict) @pytest.mark.asyncio async def test_result_never_none(self, tmp_path): worker = _make_worker(tmp_path=tmp_path) result = await worker.execute(_make_intent()) assert result is not None @pytest.mark.asyncio async def test_status_is_escalation_requested(self, tmp_path): worker = _make_worker(tmp_path=tmp_path) result = await worker.execute(_make_intent()) assert result["status"] == "escalation_requested" @pytest.mark.asyncio async def test_session_id_in_result(self, tmp_path): worker = _make_worker(tmp_path=tmp_path) result = await worker.execute(_make_intent(session_id="my-session-511")) assert result["session_id"] == "my-session-511" @pytest.mark.asyncio async def test_result_has_status_key(self, tmp_path): worker = _make_worker(tmp_path=tmp_path) result = await worker.execute(_make_intent()) assert "status" in result @pytest.mark.asyncio async def test_result_has_session_id_key(self, tmp_path): worker = _make_worker(tmp_path=tmp_path) result = await worker.execute(_make_intent()) assert "session_id" in result # --------------------------------------------------------------------------- # WB1: SETEX used — not SET without TTL # --------------------------------------------------------------------------- class TestWB1_SetexUsedNotSet: """WB1 — Implementation calls setex(), not set() without expiry.""" @pytest.mark.asyncio async def test_setex_method_called(self, tmp_path): """setex() must be called — not set() without TTL.""" redis = _make_redis_client() redis.set = MagicMock() # Also add set() to detect if it's wrongly called worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) redis.setex.assert_called_once() @pytest.mark.asyncio async def test_set_without_ttl_not_called(self, tmp_path): """set() without TTL args must NOT be called — only setex().""" redis = _make_redis_client() redis.set = MagicMock() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) redis.set.assert_not_called() @pytest.mark.asyncio async def test_setex_args_are_positional(self, tmp_path): """setex(name, time, value) — all positional per the implementation rule.""" redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent(session_id="pos-args-session")) args, kwargs = redis.setex.call_args # All 3 args should be positional assert len(args) == 3 assert args[0] == "aiva:escalation:pos-args-session" assert args[1] == ESCALATION_TTL assert args[2] == "requested" @pytest.mark.asyncio async def test_no_redis_client_does_not_raise(self, tmp_path): """Missing redis_client must be handled silently — no exception raised.""" worker = EscalationWorker(redis_client=None, events_path=tmp_path / "events.jsonl") intent = _make_intent() result = await worker.execute(intent) assert result["status"] == "escalation_requested" # --------------------------------------------------------------------------- # WB2: Observability event has session_id and intent_type # --------------------------------------------------------------------------- class TestWB2_ObservabilityEvent: """WB2 — events.jsonl entry contains session_id and intent_type.""" @pytest.mark.asyncio async def test_event_file_created(self, tmp_path): events_path = tmp_path / "events.jsonl" worker = EscalationWorker( redis_client=_make_redis_client(), events_path=events_path, ) await worker.execute(_make_intent()) assert events_path.exists() @pytest.mark.asyncio async def test_event_line_is_valid_json(self, tmp_path): events_path = tmp_path / "events.jsonl" worker = EscalationWorker( redis_client=_make_redis_client(), events_path=events_path, ) await worker.execute(_make_intent()) lines = events_path.read_text().strip().splitlines() assert len(lines) == 1 event = json.loads(lines[0]) assert isinstance(event, dict) @pytest.mark.asyncio async def test_event_contains_session_id(self, tmp_path): events_path = tmp_path / "events.jsonl" worker = EscalationWorker( redis_client=_make_redis_client(), events_path=events_path, ) await worker.execute(_make_intent(session_id="evt-sess-123")) event = json.loads(events_path.read_text().strip()) assert event["session_id"] == "evt-sess-123" @pytest.mark.asyncio async def test_event_contains_intent_type(self, tmp_path): events_path = tmp_path / "events.jsonl" worker = EscalationWorker( redis_client=_make_redis_client(), events_path=events_path, ) await worker.execute(_make_intent(intent_type=IntentType.ESCALATE_HUMAN)) event = json.loads(events_path.read_text().strip()) assert event["intent_type"] == "escalate_human" @pytest.mark.asyncio async def test_event_type_is_escalation_requested(self, tmp_path): events_path = tmp_path / "events.jsonl" worker = EscalationWorker( redis_client=_make_redis_client(), events_path=events_path, ) await worker.execute(_make_intent()) event = json.loads(events_path.read_text().strip()) assert event["event_type"] == "escalation_requested" @pytest.mark.asyncio async def test_event_has_timestamp(self, tmp_path): events_path = tmp_path / "events.jsonl" worker = EscalationWorker( redis_client=_make_redis_client(), events_path=events_path, ) await worker.execute(_make_intent()) event = json.loads(events_path.read_text().strip()) assert "timestamp" in event assert "T" in event["timestamp"] # ISO 8601 check @pytest.mark.asyncio async def test_multiple_events_each_on_own_line(self, tmp_path): events_path = tmp_path / "events.jsonl" worker = EscalationWorker( redis_client=_make_redis_client(), events_path=events_path, ) await worker.execute(_make_intent(session_id="sess-1")) await worker.execute(_make_intent(session_id="sess-2")) lines = events_path.read_text().strip().splitlines() assert len(lines) == 2 e1 = json.loads(lines[0]) e2 = json.loads(lines[1]) assert e1["session_id"] == "sess-1" assert e2["session_id"] == "sess-2" # --------------------------------------------------------------------------- # WB3: Value is "requested" string (not dict) # --------------------------------------------------------------------------- class TestWB3_ValueIsString: """WB3 — Redis value is plain string 'requested', not a JSON dict or bytes.""" @pytest.mark.asyncio async def test_stored_value_is_str_type(self, tmp_path): redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) args, _ = redis.setex.call_args assert type(args[2]) is str @pytest.mark.asyncio async def test_stored_value_equals_requested(self, tmp_path): redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) args, _ = redis.setex.call_args assert args[2] == "requested" @pytest.mark.asyncio async def test_stored_value_is_not_json(self, tmp_path): """Value must not be a JSON-encoded string — just the plain word 'requested'.""" redis = _make_redis_client() worker = _make_worker(redis_client=redis, tmp_path=tmp_path) await worker.execute(_make_intent()) args, _ = redis.setex.call_args value = args[2] # "requested" is a valid JSON string (it parses to the Python str "requested") # but it must NOT be a dict representation like '{"status": "requested"}' assert not value.startswith("{") assert value == "requested" @pytest.mark.asyncio async def test_different_session_ids_produce_different_keys(self, tmp_path): """Distinct session_ids must produce distinct Redis keys.""" redis = MagicMock() redis.setex = MagicMock() worker = EscalationWorker(redis_client=redis, events_path=tmp_path / "events.jsonl") await worker.execute(_make_intent(session_id="session-A")) await worker.execute(_make_intent(session_id="session-B")) calls = redis.setex.call_args_list assert len(calls) == 2 key_a = calls[0][0][0] key_b = calls[1][0][0] assert key_a == "aiva:escalation:session-A" assert key_b == "aiva:escalation:session-B" assert key_a != key_b @pytest.mark.asyncio async def test_redis_error_does_not_propagate(self, tmp_path): """If Redis raises, execute() must not re-raise — result dict still returned.""" redis = MagicMock() redis.setex = MagicMock(side_effect=ConnectionError("Redis down")) worker = EscalationWorker(redis_client=redis, events_path=tmp_path / "events.jsonl") result = await worker.execute(_make_intent()) assert result["status"] == "escalation_requested"