""" tests/track_a/test_story_5_07.py Story 5.07 — LeadQualificationWorker: Qualification Script Runner 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. No real Gemini calls. No real Redis. No network. No SQLite. """ import sys sys.path.insert(0, "/mnt/e/genesis-system") import asyncio import json import pytest from datetime import datetime from unittest.mock import AsyncMock, MagicMock, patch, call from core.intent.intent_signal import IntentType, IntentSignal from core.workers.lead_qualification_worker import ( LeadQualificationWorker, FALLBACK_QUESTIONS, QUALIFICATION_PROMPT, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_signal( service: str = "plumbing", location: str = "Cairns", session_id: str = "test-session-507", ) -> IntentSignal: """Factory for an IntentSignal with QUALIFY_LEAD intent and given entities.""" return IntentSignal( session_id=session_id, utterance=f"I need {service} in {location}", intent_type=IntentType.QUALIFY_LEAD, confidence=0.9, extracted_entities={"service": service, "location": location}, requires_swarm=True, created_at=datetime(2026, 2, 25, 12, 0, 0), raw_gemini_response=None, ) def _make_gemini_client(questions: list | None = None, raise_exc: Exception | None = None): """ Build a mock Gemini client. - If raise_exc is provided, generate_content() raises that exception. - Otherwise, generate_content().text returns a JSON string with given questions (default: 3 plumbing-specific questions). """ if questions is None: questions = [ "How soon do you need this fixed?", "What is your approximate budget for this job?", "Do you have any existing issues we should know about?" ] mock_client = MagicMock(name="GeminiClient") if raise_exc is not None: mock_client.generate_content.side_effect = raise_exc else: mock_response = MagicMock() mock_response.text = json.dumps({"questions": questions}) mock_client.generate_content.return_value = mock_response return mock_client def _make_redis_client(): """Build a mock Redis client.""" mock_redis = MagicMock(name="RedisClient") mock_redis.hset = MagicMock(return_value=1) return mock_redis # --------------------------------------------------------------------------- # BB1: "plumbing" + "Cairns" → 3 relevant questions returned # --------------------------------------------------------------------------- class TestBB1_PlumbingCairnsQuestions: """BB1 — Gemini returns 3 relevant questions for plumbing in Cairns.""" @pytest.mark.asyncio async def test_returns_dict_with_questions_key(self): worker = LeadQualificationWorker( gemini_client=_make_gemini_client(), redis_client=_make_redis_client(), ) result = await worker.execute(_make_signal("plumbing", "Cairns")) assert isinstance(result, dict) assert "questions" in result @pytest.mark.asyncio async def test_returns_exactly_3_questions(self): worker = LeadQualificationWorker( gemini_client=_make_gemini_client(), redis_client=_make_redis_client(), ) result = await worker.execute(_make_signal("plumbing", "Cairns")) assert len(result["questions"]) == 3 @pytest.mark.asyncio async def test_questions_are_strings(self): worker = LeadQualificationWorker( gemini_client=_make_gemini_client(), redis_client=_make_redis_client(), ) result = await worker.execute(_make_signal("plumbing", "Cairns")) for q in result["questions"]: assert isinstance(q, str) assert len(q) > 0 @pytest.mark.asyncio async def test_returns_gemini_questions_not_fallback(self): """When Gemini succeeds, returned questions must be from Gemini, not fallback.""" gemini_questions = [ "How soon do you need this fixed?", "What is your approximate budget for this job?", "Any prior plumbing issues in Cairns?" ] worker = LeadQualificationWorker( gemini_client=_make_gemini_client(questions=gemini_questions), redis_client=_make_redis_client(), ) result = await worker.execute(_make_signal("plumbing", "Cairns")) assert result["questions"] == gemini_questions # --------------------------------------------------------------------------- # BB2: Gemini failure → 3 generic fallback questions returned (no crash) # --------------------------------------------------------------------------- class TestBB2_GeminiFailureFallback: """BB2 — Gemini failure triggers fallback without crash.""" @pytest.mark.asyncio async def test_gemini_exception_does_not_raise(self): worker = LeadQualificationWorker( gemini_client=_make_gemini_client(raise_exc=RuntimeError("API timeout")), redis_client=_make_redis_client(), ) # Must not raise result = await worker.execute(_make_signal()) assert result is not None @pytest.mark.asyncio async def test_gemini_exception_returns_3_questions(self): worker = LeadQualificationWorker( gemini_client=_make_gemini_client(raise_exc=RuntimeError("API timeout")), redis_client=_make_redis_client(), ) result = await worker.execute(_make_signal()) assert len(result["questions"]) == 3 @pytest.mark.asyncio async def test_gemini_exception_returns_fallback_questions(self): worker = LeadQualificationWorker( gemini_client=_make_gemini_client(raise_exc=ConnectionError("Network down")), redis_client=_make_redis_client(), ) result = await worker.execute(_make_signal()) assert result["questions"] == FALLBACK_QUESTIONS @pytest.mark.asyncio async def test_no_gemini_client_returns_fallback(self): """If gemini_client is None, fallback questions returned.""" worker = LeadQualificationWorker( gemini_client=None, redis_client=_make_redis_client(), ) result = await worker.execute(_make_signal()) assert len(result["questions"]) == 3 assert result["questions"] == FALLBACK_QUESTIONS @pytest.mark.asyncio async def test_invalid_gemini_json_returns_fallback(self): """If Gemini returns non-JSON, fallback questions returned.""" mock_client = MagicMock() mock_response = MagicMock() mock_response.text = "Sorry, I can't do that right now." mock_client.generate_content.return_value = mock_response worker = LeadQualificationWorker( gemini_client=mock_client, redis_client=_make_redis_client(), ) result = await worker.execute(_make_signal()) assert result["questions"] == FALLBACK_QUESTIONS @pytest.mark.asyncio async def test_gemini_returns_wrong_schema_uses_fallback(self): """If Gemini JSON has no 'questions' key, fallback used.""" mock_client = MagicMock() mock_response = MagicMock() mock_response.text = json.dumps({"result": "something else"}) mock_client.generate_content.return_value = mock_response worker = LeadQualificationWorker( gemini_client=mock_client, redis_client=_make_redis_client(), ) result = await worker.execute(_make_signal()) assert result["questions"] == FALLBACK_QUESTIONS # --------------------------------------------------------------------------- # BB3: Questions written to Redis aiva:state:{session_id} # --------------------------------------------------------------------------- class TestBB3_RedisWrite: """BB3 — Questions are written to Redis aiva:state:{session_id}.""" @pytest.mark.asyncio async def test_redis_hset_called(self): redis = _make_redis_client() worker = LeadQualificationWorker( gemini_client=_make_gemini_client(), redis_client=redis, ) await worker.execute(_make_signal(session_id="sess-abc")) redis.hset.assert_called_once() @pytest.mark.asyncio async def test_redis_key_matches_session_id(self): redis = _make_redis_client() worker = LeadQualificationWorker( gemini_client=_make_gemini_client(), redis_client=redis, ) await worker.execute(_make_signal(session_id="sess-abc")) call_args = redis.hset.call_args key_arg = call_args[0][0] assert key_arg == "aiva:state:sess-abc" @pytest.mark.asyncio async def test_redis_field_is_qualification_questions(self): redis = _make_redis_client() worker = LeadQualificationWorker( gemini_client=_make_gemini_client(), redis_client=redis, ) await worker.execute(_make_signal(session_id="sess-abc")) call_args = redis.hset.call_args field_arg = call_args[0][1] assert field_arg == "qualification_questions" @pytest.mark.asyncio async def test_redis_value_is_json_list(self): redis = _make_redis_client() questions = [ "When do you need this done?", "What is your budget?", "Any special requirements?" ] worker = LeadQualificationWorker( gemini_client=_make_gemini_client(questions=questions), redis_client=redis, ) await worker.execute(_make_signal(session_id="sess-xyz")) call_args = redis.hset.call_args value_arg = call_args[0][2] parsed = json.loads(value_arg) assert parsed == questions @pytest.mark.asyncio async def test_fallback_questions_also_written_to_redis(self): """Fallback questions must still be written to Redis.""" redis = _make_redis_client() worker = LeadQualificationWorker( gemini_client=_make_gemini_client(raise_exc=RuntimeError("fail")), redis_client=redis, ) await worker.execute(_make_signal(session_id="sess-fallback")) redis.hset.assert_called_once() call_args = redis.hset.call_args value_arg = call_args[0][2] parsed = json.loads(value_arg) assert parsed == FALLBACK_QUESTIONS @pytest.mark.asyncio async def test_no_redis_client_does_not_crash(self): """If redis_client is None, execute() must not crash.""" worker = LeadQualificationWorker( gemini_client=_make_gemini_client(), redis_client=None, ) result = await worker.execute(_make_signal()) assert len(result["questions"]) == 3 # --------------------------------------------------------------------------- # WB1: Exactly 3 questions in response list # --------------------------------------------------------------------------- class TestWB1_ExactlyThreeQuestions: """WB1 — Response always contains exactly 3 questions.""" @pytest.mark.asyncio async def test_gemini_returns_4_questions_only_3_used(self): """Even if Gemini returns 4 questions, only first 3 are returned.""" four_questions = [ "Q1?", "Q2?", "Q3?", "Q4?" ] worker = LeadQualificationWorker( gemini_client=_make_gemini_client(questions=four_questions), redis_client=_make_redis_client(), ) result = await worker.execute(_make_signal()) assert len(result["questions"]) == 3 assert result["questions"] == four_questions[:3] @pytest.mark.asyncio async def test_fallback_has_exactly_3_questions(self): """The module-level FALLBACK_QUESTIONS constant contains exactly 3 items.""" assert len(FALLBACK_QUESTIONS) == 3 @pytest.mark.asyncio async def test_gemini_fewer_than_3_questions_uses_fallback(self): """If Gemini returns fewer than 3 questions, fallback is used.""" two_questions = ["Q1?", "Q2?"] mock_client = MagicMock() mock_response = MagicMock() mock_response.text = json.dumps({"questions": two_questions}) mock_client.generate_content.return_value = mock_response worker = LeadQualificationWorker( gemini_client=mock_client, redis_client=_make_redis_client(), ) result = await worker.execute(_make_signal()) assert len(result["questions"]) == 3 assert result["questions"] == FALLBACK_QUESTIONS # --------------------------------------------------------------------------- # WB2: Fallback questions defined as class constant # --------------------------------------------------------------------------- class TestWB2_FallbackAsClassConstant: """WB2 — FALLBACK_QUESTIONS is accessible as a class-level constant.""" def test_class_has_fallback_questions_constant(self): assert hasattr(LeadQualificationWorker, "FALLBACK_QUESTIONS") def test_class_constant_matches_module_constant(self): assert LeadQualificationWorker.FALLBACK_QUESTIONS == FALLBACK_QUESTIONS def test_fallback_questions_are_non_empty_strings(self): for q in LeadQualificationWorker.FALLBACK_QUESTIONS: assert isinstance(q, str) assert len(q) > 0 def test_fallback_questions_count_is_3(self): assert len(LeadQualificationWorker.FALLBACK_QUESTIONS) == 3 # --------------------------------------------------------------------------- # WB3: Gemini prompt uses entity values from intent # --------------------------------------------------------------------------- class TestWB3_PromptUsesEntityValues: """WB3 — Gemini prompt includes the service and location from extracted_entities.""" @pytest.mark.asyncio async def test_prompt_includes_service_entity(self): """generate_content() receives a prompt containing the service value.""" mock_client = _make_gemini_client() worker = LeadQualificationWorker( gemini_client=mock_client, redis_client=_make_redis_client(), ) await worker.execute(_make_signal(service="electrical", location="Brisbane")) call_args = mock_client.generate_content.call_args prompt = call_args[0][0] assert "electrical" in prompt @pytest.mark.asyncio async def test_prompt_includes_location_entity(self): """generate_content() receives a prompt containing the location value.""" mock_client = _make_gemini_client() worker = LeadQualificationWorker( gemini_client=mock_client, redis_client=_make_redis_client(), ) await worker.execute(_make_signal(service="electrical", location="Brisbane")) call_args = mock_client.generate_content.call_args prompt = call_args[0][0] assert "Brisbane" in prompt @pytest.mark.asyncio async def test_prompt_uses_default_when_entities_missing(self): """If extracted_entities is empty, prompt uses default placeholder values.""" signal = IntentSignal( session_id="sess-empty", utterance="I need help", intent_type=IntentType.QUALIFY_LEAD, confidence=0.8, extracted_entities={}, requires_swarm=True, created_at=datetime(2026, 2, 25, 12, 0, 0), raw_gemini_response=None, ) mock_client = _make_gemini_client() worker = LeadQualificationWorker( gemini_client=mock_client, redis_client=_make_redis_client(), ) result = await worker.execute(signal) # Must return 3 questions (Gemini or fallback) without crashing assert len(result["questions"]) == 3 @pytest.mark.asyncio async def test_different_services_produce_different_prompts(self): """Two different service values must produce different prompt strings.""" prompts_seen = [] def capture_prompt(prompt): prompts_seen.append(prompt) mock_response = MagicMock() mock_response.text = json.dumps({"questions": ["Q1?", "Q2?", "Q3?"]}) return mock_response mock_client = MagicMock() mock_client.generate_content.side_effect = capture_prompt worker = LeadQualificationWorker( gemini_client=mock_client, redis_client=_make_redis_client(), ) await worker.execute(_make_signal(service="plumbing", location="Cairns")) await worker.execute(_make_signal(service="roofing", location="Cairns")) assert prompts_seen[0] != prompts_seen[1]