"""B1 completion diagnostics: no provider, network or production writes in these tests.""" import json from types import SimpleNamespace from unittest.mock import Mock import pytest from scripts import diagnose_discovery_admission_13b as diagnostic @pytest.mark.asyncio async def test_comparison_uses_real_gate_and_keeps_fresh_and_reused_lifecycles(monkeypatch): clients, calls = [], [] class FakeRedis: def __init__(self): self.connection_pool = SimpleNamespace(connection_class=object) self.closed = False clients.append(self) async def ping(self): return True async def info(self, section): assert section == "errorstats" return {} async def eval(self, script, count, key): assert "'INCR'" in script and "'EXPIRE',KEYS[1],120" in script assert count == 1 and key.startswith("project-intake:budget:") assert not self.closed calls.append(self) return 1 async def aclose(self): self.closed = True settings = SimpleNamespace(redis_url="redis://test", project_intake_calls_per_minute=60) monkeypatch.setattr(diagnostic, "get_settings", lambda: settings) monkeypatch.setattr(diagnostic.Redis, "from_url", lambda *a, **kw: FakeRedis()) output = Mock() output.exists.return_value = False await diagnostic.diagnose(2, output) receipt = json.loads(output.write_text.call_args.args[0]) assert len(clients) == 3 # one persistent, two fresh assert all(client.closed for client in clients) assert len(calls) == 4 and calls.count(clients[0]) == 2 assert receipt["environment"] == "local" assert receipt["provider_calls"] == 0 assert receipt["policy_changed"] is False assert len(receipt["rows"]) == 4 assert all(row["error_type"] is None for row in receipt["rows"]) @pytest.mark.asyncio async def test_diagnostic_refuses_overwrite(monkeypatch): monkeypatch.setattr(diagnostic, "preflight", lambda: {}) output = Mock() output.exists.return_value = True with pytest.raises(ValueError, match="new diagnostic"): await diagnostic.diagnose(1, output) output.write_text.assert_not_called() def test_preflight_fails_if_admission_policy_differs(monkeypatch): monkeypatch.setattr( diagnostic, "get_settings", lambda: SimpleNamespace(redis_url="redis://test", project_intake_calls_per_minute=30), ) with pytest.raises(ValueError, match="60/min"): diagnostic.preflight() def test_strict_extraction_arrays_are_required_not_nullable(): from openai.lib._pydantic import to_strict_json_schema from src.domain.projects.discovery_schemas import DiscoveryTurnResult schema = to_strict_json_schema(DiscoveryTurnResult) spec = schema["$defs"]["DiscoverySpec"] for name in ( "entities", "workflows", "actors", "archetypes", "forms", "explicit_exclusions", "requires_configuration", ): assert name in spec["required"] assert spec["properties"][name]["type"] == "array" assert "anyOf" not in spec["properties"][name] assert "requested_capabilities" in schema["required"] def test_discovery_schema_avoids_unsupported_object_max_properties(): from openai.lib._pydantic import to_strict_json_schema from src.domain.projects.discovery_schemas import DiscoveryTurnResult schema = to_strict_json_schema(DiscoveryTurnResult) auth = schema["$defs"]["AuthenticationSpec"] assert "maxProperties" not in auth["properties"]["audience_methods"] audience_methods = auth["properties"]["audience_methods"] assert audience_methods["required"] == ["staff", "customer"] assert audience_methods["additionalProperties"] is False assert set(audience_methods["properties"]) == {"staff", "customer"} def test_core_preserves_explicit_nouns_number_and_distinct_entities(): from src.domain.projects.discovery_fidelity import apply_fidelity_hygiene from src.domain.projects.discovery_schemas import DiscoverySpec, Entity names = ["inspection report", "inspection reports", "job requests", "contact submissions"] spec = DiscoverySpec( entities=[Entity(id=str(i), name=name, description=name) for i, name in enumerate(names)] ) result = apply_fidelity_hygiene( spec, {"messages": [{"role": "user", "content": ", ".join(names)}]} ) assert [e.name for e in result.entities] == names def test_extraction_discipline_is_in_the_model_facing_prompt(): from src.integrations.project_intake_client import DISCOVERY_PROMPT assert "EXTRACTION DISCIPLINE" in DISCOVERY_PROMPT assert "preserve the user's usable" in DISCOVERY_PROMPT assert "noun phrase" in DISCOVERY_PROMPT assert "Leave workflows empty when no" in DISCOVERY_PROMPT assert "meaningful process is" in DISCOVERY_PROMPT assert "INTENT FIDELITY" in DISCOVERY_PROMPT assert "Never invent App Blocks" in DISCOVERY_PROMPT assert "FORBIDDEN QUESTIONS" in DISCOVERY_PROMPT assert "Naming a form" in DISCOVERY_PROMPT assert "Get a Quote, Contact, lead intake" in DISCOVERY_PROMPT assert "navigation through sections" in DISCOVERY_PROMPT assert "browse→contact/quote" in DISCOVERY_PROMPT or "browse→convert" in DISCOVERY_PROMPT assert "would a wrong answer materially change" in DISCOVERY_PROMPT assert "MARKETING SITE / LANDING PAGE / PORTFOLIO" in DISCOVERY_PROMPT assert "often in one or two turns" in DISCOVERY_PROMPT def test_extraction_schema_descriptions_bind_noun_and_workflow_discipline(): from src.domain.projects.discovery_schemas import DiscoverySpec, Entity entity_name = Entity.model_fields["name"].description or "" spec = DiscoverySpec.model_fields assert "grammatical number" in entity_name assert "explicitly named" in (spec["entities"].description or "") assert "including when decision is clarify" in (spec["workflows"].description or "") assert "customer's own product phrasing" in (spec["original_intent"].description or "") def test_core_does_not_extract_implicit_entities_or_workflows(): from src.domain.projects.discovery_fidelity import apply_fidelity_hygiene from src.domain.projects.discovery_schemas import DiscoverySpec result = apply_fidelity_hygiene( DiscoverySpec(), { "messages": [ { "role": "user", "content": "Let visitors write to our team and let the team read it.", } ] }, ) assert result.entities == [] and result.workflows == [] and result.archetypes == []