backend/app/test_assistant_fallback.py

from __future__ import annotations

import time

from app.assistant_fallback import (
    classify_concept_request,
    concept_answer_needs_web_refinement,
    deterministic_concept_payload,
    responses_create_with_deadline,
)


class _SlowResponses:
    def __init__(self, sleep_s: float) -> None:
        self.sleep_s = float(sleep_s)

    def create(self, **kwargs):  # noqa: ANN003
        time.sleep(self.sleep_s)
        return {"ok": True, "kwargs": kwargs}


class _Client:
    def __init__(self, sleep_s: float) -> None:
        self.responses = _SlowResponses(sleep_s)


def test_responses_create_with_deadline_returns_before_timeout() -> None:
    client = _Client(0.01)
    result = responses_create_with_deadline(client, timeout_s=0.25, model="gpt-5-mini")
    assert result["ok"] is True
    assert result["kwargs"]["model"] == "gpt-5-mini"
    assert float(result["kwargs"]["timeout"]) == 0.25


def test_responses_create_with_deadline_raises_timeout_for_slow_calls() -> None:
    client = _Client(0.3)
    started = time.perf_counter()
    try:
        responses_create_with_deadline(client, timeout_s=0.05, model="gpt-5-mini")
    except TimeoutError as exc:
        elapsed = time.perf_counter() - started
        assert "responses.create exceeded" in str(exc)
        assert elapsed < 0.2
    else:  # pragma: no cover
        raise AssertionError("Expected TimeoutError for slow responses.create call")


def test_concept_answer_does_not_force_web_for_generic_current_wording() -> None:
    assistant = (
        "**Result**\n\nCellular failover keeps a branch on a backup cellular path if the primary wired circuit fails."
        "\n\n**Why**\n\n- Concept explainer only.\n\n**Next action**\n\n- Confirm outage traffic requirements."
    )
    assert (
        concept_answer_needs_web_refinement(
            "What is the current difference between cellular failover and wired failover for a branch router?",
            assistant,
            {"llm_assisted": True},
        )
        is False
    )


def test_concept_answer_still_requests_web_for_latest_wording() -> None:
    assistant = (
        "**Result**\n\n5G generally provides more headroom than 4G."
        "\n\n**Why**\n\n- Concept explainer only.\n\n**Next action**\n\n- Validate public network state if needed."
    )
    assert (
        concept_answer_needs_web_refinement(
            "What are the latest public differences between 4G and 5G for branch backup?",
            assistant,
            {"llm_assisted": True},
        )
        is True
    )


def test_classify_concept_request_blocks_high_risk_code_adjudication() -> None:
    classified = classify_concept_request(
        "Will this pass AHJ inspection for a fire alarm pathway conversion?",
        domain="pots",
        enabled=True,
        scope_terms=("pots", "fire alarm", "elevator", "pathway"),
    )
    assert classified["in_scope"] is True
    assert classified["blocked"] is True
    assert classified["allow_concept"] is False


def test_classify_concept_request_keeps_generic_karis_law_explainer_allowed() -> None:
    classified = classify_concept_request(
        "What is Kari's Law in plain English for a POTS replacement discussion?",
        domain="pots",
        enabled=True,
        scope_terms=("pots", "kari's law", "ray baum", "e911"),
    )
    assert classified["in_scope"] is True
    assert classified["blocked"] is False
    assert classified["allow_concept"] is True


def test_classify_concept_request_does_not_false_positive_on_ul_substring() -> None:
    classified = classify_concept_request(
        "Explain POTS replacement in plain English for a multi-site retail customer.",
        domain="pots",
        enabled=True,
        scope_terms=("pots", "pots replacement", "multi-site", "retail"),
    )
    assert classified["in_scope"] is True
    assert classified["blocked"] is False
    assert classified["allow_concept"] is True


def test_classify_concept_request_still_blocks_real_ul_compliance_ask() -> None:
    classified = classify_concept_request(
        "Will this UL pathway meet fire code requirements for elevator migration?",
        domain="pots",
        enabled=True,
        scope_terms=("fire code", "elevator", "pathway", "pots"),
    )
    assert classified["in_scope"] is True
    assert classified["blocked"] is True
    assert classified["allow_concept"] is False


def test_deterministic_router_concept_payload_is_source_backed() -> None:
    payload = deterministic_concept_payload(
        "How do I describe WAN vs LAN ports to non-technical buyers?",
        domain="router_docs",
    )
    assert payload is not None
    assert payload["meta"]["retrieval_mode"] == "router_wan_lan_concept_fast"
    assert "WAN" in str(payload["result"])
    assert "LAN" in str(payload["result"])
    assert any("FAQ_200_ansers_set_3.csv" in str(src.get("doc") or "") for src in list(payload.get("sources") or []))


def test_deterministic_pots_concept_payload_requires_internal_playbook() -> None:
    payload = deterministic_concept_payload(
        "What's the difference between copper sunset and POTS replacement?",
        domain="pots",
        available_docs=["pots_top100_questions_draft.md"],
    )
    assert payload is not None
    assert payload["meta"]["retrieval_mode"] == "pots_copper_sunset_concept_fast"
    assert "Copper sunset" in str(payload["result"])
    assert any("pots_top100_questions_draft.md" in str(src.get("doc") or "") for src in list(payload.get("sources") or []))


def test_deterministic_masters_contact_center_payload_is_source_backed() -> None:
    payload = deterministic_concept_payload(
        "Explain contact center in plain English for a sales rep.",
        domain="masters",
        available_docs=["MST_Contact Center.pdf"],
    )
    assert payload is not None
    assert payload["meta"]["retrieval_mode"] == "masters_contact_center_concept_fast"
    assert "Contact center in plain English" in str(payload["result"])
    assert any("MST_Contact Center.pdf" in str(src.get("doc") or "") for src in list(payload.get("sources") or []))