LLM Unit Testing: Writing Tests for Your Prompts

"""Deterministic schema assertions for a structured-output prompt.

The contract here is hard: the model either returns an object with these five
fields, at these types, and nothing else, or the test fails. Nothing about this
is fuzzier than testing any other JSON API response.

Run it:  pytest tests/test_structured_output.py -v
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Literal

import pytest
import yaml
from pydantic import BaseModel, ConfigDict, Field, ValidationError

PROMPT_PATH = (
    Path(__file__).resolve().parents[1] / "prompts" / "summarize_ticket" / "v3.yaml"
)

ACCOUNT_ID_PATTERN = r"^(ACCT-\d{6}|unknown)$"


class TicketSummary(BaseModel):
    """The response contract, as a schema instead of a hope.

    extra="forbid" is the part teams skip. Without it, a model that renames a
    field to something plausible still passes, because the field you asked for
    is quietly absent and the one it invented is quietly ignored.
    """

    model_config = ConfigDict(extra="forbid")

    account_id: str = Field(pattern=ACCOUNT_ID_PATTERN)
    category: Literal["billing", "bug", "how_to", "other"]
    urgency: int = Field(ge=1, le=5)
    summary: str = Field(min_length=10)
    needs_human: bool


# Recorded model responses, so this file runs in CI with no credentials and no
# network. Delete this dict and point generate() at your provider to run live.
RECORDED_RESPONSES = {
    "billing_duplicate_charge": (
        '{"account_id": "ACCT-482913", "category": "billing", "urgency": 4, '
        '"summary": "The customer was billed twice on the same day for one '
        'monthly subscription. They are asking for a refund of the duplicate '
        'charge.", "needs_human": true}'
    ),
    "bug_login_redirect_loop": (
        '{"account_id": "ACCT-771204", "category": "bug", "urgency": 4, '
        '"summary": "Signing in on Chrome loops the customer back to the login '
        'screen and blocks the invoices page. Safari is unaffected.", '
        '"needs_human": true}'
    ),
    "how_to_change_billing_email": (
        '{"account_id": "unknown", "category": "how_to", "urgency": 2, '
        '"summary": "The customer wants to change the email address that '
        'receives the monthly invoice. Profile settings only showed the login '
        'email.", "needs_human": false}'
    ),
}


def load_prompt(path: Path = PROMPT_PATH) -> dict:
    with path.open(encoding="utf-8") as handle:
        return yaml.safe_load(handle)


def generate(system: str, user: str, case_id: str | None = None) -> str:
    """Stand-in for a real model call.

    Replace the body with your provider's chat completion, using the model and
    temperature from the fixture. The case_id argument exists only so the offline
    stub can replay a recorded response. A real implementation ignores it.
    """
    if case_id is None or case_id not in RECORDED_RESPONSES:
        raise AssertionError(f"no recorded response for case {case_id!r}")
    return RECORDED_RESPONSES[case_id]


PROMPT = load_prompt()
GOLDEN_CASES = PROMPT["golden_cases"]
CASE_IDS = [case["id"] for case in GOLDEN_CASES]


def render_user_message(case: dict) -> str:
    template = PROMPT["user_template"]
    variables = case["variables"]
    for name, value in variables.items():
        template = template.replace("{" + name + "}", str(value))
    return template


@pytest.mark.parametrize("case", GOLDEN_CASES, ids=CASE_IDS)
def test_response_satisfies_the_schema(case: dict) -> None:
    raw = generate(
        PROMPT["system_prompt"], render_user_message(case), case_id=case["id"]
    )

    # Parsing and validation are one step. If either fails, the test fails.
    summary = TicketSummary.model_validate_json(raw)

    for field, expected_value in case["expected"].items():
        assert getattr(summary, field) == expected_value, (
            f"{case['id']}: expected {field}={expected_value!r}, "
            f"got {getattr(summary, field)!r}"
        )


def test_the_fixture_pins_deterministic_decoding() -> None:
    # A structured-output prompt that is graded on exact fields has no business
    # running at a nonzero temperature.
    assert PROMPT["temperature"] == 0
    assert PROMPT["version"] == 3


def test_a_missing_required_field_fails() -> None:
    payload = json.loads(RECORDED_RESPONSES["billing_duplicate_charge"])
    del payload["account_id"]

    with pytest.raises(ValidationError) as excinfo:
        TicketSummary.model_validate(payload)

    assert "account_id" in str(excinfo.value)


def test_a_wrongly_typed_field_fails() -> None:
    payload = json.loads(RECORDED_RESPONSES["billing_duplicate_charge"])
    payload["urgency"] = "high"  # the model wrote a word where an integer belongs

    with pytest.raises(ValidationError):
        TicketSummary.model_validate(payload)


def test_an_out_of_range_enum_value_fails() -> None:
    payload = json.loads(RECORDED_RESPONSES["billing_duplicate_charge"])
    payload["category"] = "refund_request"  # plausible, and not in the contract

    with pytest.raises(ValidationError):
        TicketSummary.model_validate(payload)


def test_a_renamed_field_fails_instead_of_passing_silently() -> None:
    payload = json.loads(RECORDED_RESPONSES["billing_duplicate_charge"])
    payload["accountId"] = payload.pop("account_id")  # camelCase drift

    with pytest.raises(ValidationError) as excinfo:
        TicketSummary.model_validate(payload)

    message = str(excinfo.value)
    assert "account_id" in message  # required field absent
    assert "accountId" in message  # unexpected field present


def test_an_empty_account_id_fails() -> None:
    # The field is present and correctly typed, and still useless downstream.
    # This is the case a key-presence check waves through.
    payload = json.loads(RECORDED_RESPONSES["billing_duplicate_charge"])
    payload["account_id"] = ""

    with pytest.raises(ValidationError):
        TicketSummary.model_validate(payload)

"""Must-contain and must-not-contain assertions on raw model output.

This is the test that would have caught the regression in the article's opening
story. A teammate tightened the system prompt to stop the model appending a
sign-off, and the model quietly stopped extracting the account ID along with it.
The JSON stayed valid. The schema stayed satisfied. The one field the pipeline
depended on went missing.

A must-contain assertion on the account-id pattern fails on the first commit
instead of six days later.

Run it:  pytest tests/test_string_assertions.py -v
"""

from __future__ import annotations

import re
from pathlib import Path

import pytest
import yaml

PROMPT_PATH = (
    Path(__file__).resolve().parents[1] / "prompts" / "summarize_ticket" / "v3.yaml"
)

# Recorded model responses, so this file runs with no credentials and no network.
# Replace generate() with a real call to run it live.
RECORDED_RESPONSES = {
    "billing_duplicate_charge": (
        '{"account_id": "ACCT-482913", "category": "billing", "urgency": 4, '
        '"summary": "The customer was billed twice on the same day for one '
        'monthly subscription. They are asking for a refund of the duplicate '
        'charge.", "needs_human": true}'
    ),
    "bug_login_redirect_loop": (
        '{"account_id": "ACCT-771204", "category": "bug", "urgency": 4, '
        '"summary": "Signing in on Chrome loops the customer back to the login '
        'screen and blocks the invoices page. Safari is unaffected.", '
        '"needs_human": true}'
    ),
    "how_to_change_billing_email": (
        '{"account_id": "unknown", "category": "how_to", "urgency": 2, '
        '"summary": "The customer wants to change the email address that '
        'receives the monthly invoice. Profile settings only showed the login '
        'email.", "needs_human": false}'
    ),
}

# The actual output from the incident: valid JSON, correct shape, account_id
# silently dropped to the fallback on a ticket that plainly contained one.
REGRESSED_RESPONSE = (
    '{"account_id": "unknown", "category": "billing", "urgency": 4, '
    '"summary": "The customer was billed twice on the same day for one monthly '
    'subscription and wants a refund.", "needs_human": true}'
)

# The other half of the same incident: the sign-off the prompt edit was meant to
# remove, reintroduced by a later edit that reverted one sentence too many.
BOILERPLATE_RESPONSE = (
    '{"account_id": "ACCT-482913", "category": "billing", "urgency": 4, '
    '"summary": "The customer was billed twice and wants a refund. Thanks for '
    'reaching out, we are happy to help.", "needs_human": true}'
)


def load_prompt(path: Path = PROMPT_PATH) -> dict:
    with path.open(encoding="utf-8") as handle:
        return yaml.safe_load(handle)


def generate(system: str, user: str, case_id: str | None = None) -> str:
    """Stand-in for a real model call. Swap the body for your provider's client."""
    if case_id is None or case_id not in RECORDED_RESPONSES:
        raise AssertionError(f"no recorded response for case {case_id!r}")
    return RECORDED_RESPONSES[case_id]


def assert_string_contract(
    output: str,
    must_contain: list[str] | None = None,
    must_contain_patterns: list[str] | None = None,
    must_not_contain: list[str] | None = None,
) -> None:
    """Assert presence and absence markers against raw model text.

    Kept as one function so the same contract can be applied to a recorded
    response, a live response, or a regression sample.
    """
    for needle in must_contain or []:
        assert needle in output, f"required substring {needle!r} missing from output"

    for pattern in must_contain_patterns or []:
        assert re.search(pattern, output), (
            f"required pattern {pattern!r} did not match output"
        )

    for needle in must_not_contain or []:
        assert needle.lower() not in output.lower(), (
            f"forbidden substring {needle!r} present in output"
        )


PROMPT = load_prompt()
GOLDEN_CASES = PROMPT["golden_cases"]
CASE_IDS = [case["id"] for case in GOLDEN_CASES]


def render_user_message(case: dict) -> str:
    template = PROMPT["user_template"]
    for name, value in case["variables"].items():
        template = template.replace("{" + name + "}", str(value))
    return template


@pytest.mark.parametrize("case", GOLDEN_CASES, ids=CASE_IDS)
def test_output_meets_its_string_contract(case: dict) -> None:
    output = generate(
        PROMPT["system_prompt"], render_user_message(case), case_id=case["id"]
    )

    assert_string_contract(
        output,
        must_contain=case.get("must_contain"),
        must_contain_patterns=case.get("must_contain_patterns"),
        must_not_contain=case.get("must_not_contain"),
    )


def test_the_dropped_account_id_regression_is_caught() -> None:
    """Proof the assertion above is load-bearing and not decorative.

    Feed it the exact output from the incident and it fails. This is the six-days
    of-silence bug, turned into a red test.
    """
    case = next(c for c in GOLDEN_CASES if c["id"] == "billing_duplicate_charge")

    with pytest.raises(AssertionError) as excinfo:
        assert_string_contract(
            REGRESSED_RESPONSE,
            must_contain=case["must_contain"],
            must_contain_patterns=case["must_contain_patterns"],
            must_not_contain=case["must_not_contain"],
        )

    assert "ACCT-482913" in str(excinfo.value)


def test_reintroduced_boilerplate_is_caught() -> None:
    case = next(c for c in GOLDEN_CASES if c["id"] == "billing_duplicate_charge")

    with pytest.raises(AssertionError) as excinfo:
        assert_string_contract(
            BOILERPLATE_RESPONSE,
            must_contain=case["must_contain"],
            must_not_contain=case["must_not_contain"],
        )

    assert "Thanks for reaching out" in str(excinfo.value)


def test_forbidden_markers_are_matched_case_insensitively() -> None:
    # Models reword their own boilerplate. "BEST REGARDS" is the same bug.
    with pytest.raises(AssertionError):
        assert_string_contract(
            '{"summary": "Refund issued. BEST REGARDS, support"}',
            must_not_contain=["Best regards"],
        )


def test_every_golden_case_declares_at_least_one_marker() -> None:
    # A golden case with no assertions is a golden case that cannot fail.
    for case in GOLDEN_CASES:
        markers = (
            case.get("must_contain", [])
            + case.get("must_contain_patterns", [])
            + case.get("must_not_contain", [])
        )
        assert markers, f"{case['id']} declares no string markers"

"""Semantic similarity assertions, with the threshold chosen empirically.

Use this when a correct answer has no fixed wording. The interesting part is not
the cosine formula, it is where the threshold comes from: calibrate_threshold()
scores a set of known-good answers against the reference and puts the floor just
below the worst true positive. Picking 0.8 because it looks like a reasonable
number is how semantic tests become the flakiest tests in a suite.

Run it:            pytest tests/test_semantic_similarity.py -v
See the numbers:   pytest tests/test_semantic_similarity.py -s -k calibration
"""

from __future__ import annotations

import hashlib
import re
import statistics
from pathlib import Path

import numpy as np
import pytest
import yaml

PROMPT_PATH = (
    Path(__file__).resolve().parents[1] / "prompts" / "summarize_ticket" / "v3.yaml"
)

# Margin below the worst known-good score. Small enough that a real regression
# still fails, wide enough to absorb ordinary embedding noise.
CALIBRATION_MARGIN = 0.05

EMBEDDING_DIM = 512

_WORD = re.compile(r"[a-z0-9]+")
_STOPWORDS = frozenset(
    """
    a an and are as at be been but by can for from had has have how in into is it
    its of on only or that the their them there they this to was were what when
    which who will with would you your
    """.split()
)


def _tokens(text: str) -> list[str]:
    return [
        token
        for token in _WORD.findall(text.lower())
        if len(token) > 2 and token not in _STOPWORDS
    ]


def get_embedding(text: str) -> np.ndarray:
    """Offline stand-in for an embeddings API.

    Signed hashing of content words into a fixed-width vector. It is not a
    language model, but it is deterministic, needs no credentials, and behaves
    like an embedding for the purpose of this test: paraphrases of the same
    answer land near each other, unrelated text lands near zero.

    Replace the body with a real call, for example:

        from openai import OpenAI
        client = OpenAI()
        vector = client.embeddings.create(
            model="text-embedding-3-small", input=text
        ).data[0].embedding
        return np.asarray(vector, dtype=np.float64)

    The rest of this file, including the calibration routine, works unchanged.
    """
    vector = np.zeros(EMBEDDING_DIM, dtype=np.float64)
    for token in _tokens(text):
        digest = hashlib.sha256(token.encode("utf-8")).digest()
        index = int.from_bytes(digest[:4], "big") % EMBEDDING_DIM
        sign = 1.0 if digest[4] % 2 == 0 else -1.0
        vector[index] += sign

    norm = float(np.linalg.norm(vector))
    if norm == 0.0:
        return vector
    return vector / norm


def cosine_similarity(left: np.ndarray, right: np.ndarray) -> float:
    denominator = float(np.linalg.norm(left) * np.linalg.norm(right))
    if denominator == 0.0:
        return 0.0
    return float(np.dot(left, right) / denominator)


def similarity(candidate: str, reference: str) -> float:
    return cosine_similarity(get_embedding(candidate), get_embedding(reference))


def calibrate_threshold(
    reference: str,
    known_good: list[str],
    margin: float = CALIBRATION_MARGIN,
    label: str = "",
) -> float:
    """Derive a threshold from data instead of guessing one.

    Scores every known-good answer against the reference, prints the distribution
    so a human can look at it, and returns a floor just below the worst true
    positive. If that floor lands uncomfortably low, the honest read is that this
    check does not discriminate on this case and you need a different assertion,
    not a nudged number.
    """
    scores = sorted(similarity(answer, reference) for answer in known_good)

    print(f"\ncalibration {label or 'case'}: n={len(scores)}")
    for score, answer in zip(scores, sorted(known_good, key=lambda a: similarity(a, reference))):
        print(f"  {score:.3f}  {answer[:72]}")
    print(
        f"  min={scores[0]:.3f} median={statistics.median(scores):.3f} "
        f"max={scores[-1]:.3f}"
    )

    threshold = round(scores[0] - margin, 3)
    print(f"  threshold = worst true positive - {margin} = {threshold:.3f}")
    return threshold


# Recorded model output, so this file runs with no credentials and no network.
# Each summary is worded differently from every calibration answer on purpose:
# the point is to score meaning, not overlap with a memorized string.
RECORDED_SUMMARIES = {
    "billing_duplicate_charge": (
        "The customer received two identical subscription charges on one day and "
        "is requesting a refund for the duplicate charge."
    ),
    "bug_login_redirect_loop": (
        "A login redirect loop on Chrome blocks the customer from reaching the "
        "invoices page, while Safari signs in normally."
    ),
    "how_to_change_billing_email": (
        "The customer is asking where to change the email address that receives "
        "the monthly invoice, having found only the login email in settings."
    ),
}

OFF_TOPIC_SUMMARY = (
    "The office cafeteria will be closed on Friday for scheduled maintenance."
)


def load_prompt(path: Path = PROMPT_PATH) -> dict:
    with path.open(encoding="utf-8") as handle:
        return yaml.safe_load(handle)


PROMPT = load_prompt()
GOLDEN_CASES = PROMPT["golden_cases"]
CASE_IDS = [case["id"] for case in GOLDEN_CASES]


@pytest.mark.parametrize("case", GOLDEN_CASES, ids=CASE_IDS)
def test_summary_is_semantically_close_to_the_reference(case: dict) -> None:
    reference = case["reference_summary"]
    threshold = calibrate_threshold(
        reference, case["accepted_summaries"], label=case["id"]
    )

    score = similarity(RECORDED_SUMMARIES[case["id"]], reference)

    assert score >= threshold, (
        f"{case['id']}: similarity {score:.3f} below calibrated threshold "
        f"{threshold:.3f}"
    )


@pytest.mark.parametrize("case", GOLDEN_CASES, ids=CASE_IDS)
def test_an_off_topic_answer_falls_below_the_threshold(case: dict) -> None:
    """A threshold that nothing can fail is not a threshold."""
    reference = case["reference_summary"]
    threshold = calibrate_threshold(
        reference, case["accepted_summaries"], label=case["id"]
    )

    score = similarity(OFF_TOPIC_SUMMARY, reference)

    assert score < threshold, (
        f"{case['id']}: off-topic answer scored {score:.3f}, at or above the "
        f"threshold {threshold:.3f}. The threshold is too low to be useful."
    )


def test_the_threshold_is_derived_from_the_worst_true_positive() -> None:
    case = GOLDEN_CASES[0]
    scores = [
        similarity(answer, case["reference_summary"])
        for answer in case["accepted_summaries"]
    ]
    threshold = calibrate_threshold(
        case["reference_summary"], case["accepted_summaries"], label=case["id"]
    )

    assert threshold == pytest.approx(min(scores) - CALIBRATION_MARGIN, abs=1e-3)
    assert threshold < min(scores)  # every known-good answer still passes
    assert threshold > 0.0  # a non-positive floor would accept anything


def test_identical_text_scores_one() -> None:
    reference = GOLDEN_CASES[0]["reference_summary"]
    assert similarity(reference, reference) == pytest.approx(1.0, abs=1e-9)


def test_similarity_cannot_catch_a_factually_wrong_answer() -> None:
    """The documented ceiling of this assertion type, asserted rather than assumed.

    Same wording, wrong number. Cosine similarity measures closeness in meaning
    space, not truth, so this scores like a correct answer and passes the
    threshold. If a wrong number is the failure you care about, this is the wrong
    test, and no amount of threshold tuning fixes it.
    """
    case = next(c for c in GOLDEN_CASES if c["id"] == "billing_duplicate_charge")
    threshold = calibrate_threshold(
        case["reference_summary"], case["accepted_summaries"], label=case["id"]
    )

    confidently_wrong = (
        "The customer was charged seventeen times for the same monthly "
        "subscription on the same day and is asking for one of the duplicate "
        "charges to be refunded."
    )
    score = similarity(confidently_wrong, case["reference_summary"])

    assert score >= threshold

"""LLM-as-judge, for the residual nothing cheaper can express.

Three rules make the difference between a judge and a random number generator:

1. An explicit rubric. Named criteria with a definition each, not "rate this
   response 1 to 5", which asks for a plausible number rather than a measurement.
2. A pinned judge model. Not "the latest", or your judge drifts underneath you at
   the same time as the model you are testing, and you cannot tell which moved.
3. Temperature 0. It does not make a judge deterministic, it stops it from
   volunteering extra variance on top of what is already there.

This is the most expensive and least stable assertion type in the suite. Reach
for it last, on the small set of properties a schema and an embedding score
genuinely cannot express.

Run it:  pytest tests/test_llm_judge.py -v
"""

from __future__ import annotations

import json
import re

import pytest
from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator

# Pinned to a dated snapshot on purpose. "gpt-4o" is a moving target: the weights
# behind that alias change, and when they do, every judged score in your history
# shifts with no commit to blame. A dated snapshot makes a judge upgrade an
# explicit, reviewable, one-line diff.
JUDGE_MODEL = "gpt-4o-2024-08-06"

# Not negotiable for a grader.
JUDGE_TEMPERATURE = 0.0

# The rubric. Each criterion is a name plus a definition precise enough that two
# people reading the same response would agree on pass or fail.
RUBRIC: tuple[tuple[str, str], ...] = (
    (
        "addresses_the_specific_problem",
        "The reply restates the customer's actual problem, with its specifics, "
        "rather than a generic category of problem.",
    ),
    (
        "states_a_concrete_next_step",
        "The reply names one concrete next step and who takes it, rather than "
        "saying the issue is being looked into.",
    ),
    (
        "makes_no_unfounded_promise",
        "The reply promises no refund, deadline, credit, or outcome that the "
        "provided context does not already authorize.",
    ),
    (
        "tone_is_calm_and_non_defensive",
        "The reply neither blames the customer nor over-apologizes, and does not "
        "argue with the customer's account of events.",
    ),
)

RUBRIC_NAMES = tuple(name for name, _ in RUBRIC)


class CriterionResult(BaseModel):
    """One rubric line, graded."""

    model_config = ConfigDict(extra="forbid")

    criterion: str
    passed: bool
    evidence: str = Field(min_length=1)  # a verdict with no quote is not a verdict


class JudgeVerdict(BaseModel):
    """The judge's structured output, validated like any other model response."""

    model_config = ConfigDict(extra="forbid")

    results: list[CriterionResult]

    @model_validator(mode="after")
    def _must_cover_the_whole_rubric(self) -> "JudgeVerdict":
        graded = [result.criterion for result in self.results]
        if sorted(graded) != sorted(RUBRIC_NAMES):
            missing = sorted(set(RUBRIC_NAMES) - set(graded))
            unexpected = sorted(set(graded) - set(RUBRIC_NAMES))
            raise ValueError(
                f"verdict does not cover the rubric; missing={missing} "
                f"unexpected={unexpected}"
            )
        return self

    @property
    def failures(self) -> list[str]:
        return [result.criterion for result in self.results if not result.passed]

    @property
    def passed(self) -> bool:
        return not self.failures


def build_judge_prompt(customer_message: str, reply: str) -> str:
    """Render the rubric into a grading prompt.

    Everything the judge is allowed to consider is in this string. No score
    scale, no "use your judgment", no room to average four things into one number.
    """
    criteria_block = "\n".join(
        f"- {name}: {definition}" for name, definition in RUBRIC
    )
    return (
        "You are grading a support reply against a fixed rubric.\n"
        "Grade each criterion independently as pass or fail. Do not average them "
        "and do not produce an overall score.\n"
        "For each criterion, quote the span of the reply that decided it.\n\n"
        f"Criteria:\n{criteria_block}\n\n"
        f"Customer message:\n{customer_message}\n\n"
        f"Reply under test:\n{reply}\n\n"
        'Return only JSON of the form {"results": [{"criterion": "<name>", '
        '"passed": true, "evidence": "<quote>"}]} with one entry per criterion, '
        "using the criterion names exactly as written above."
    )


# Every judge call this module makes, recorded so the tests can assert on how the
# judge was invoked and not just on what it returned.
JUDGE_CALLS: list[dict] = []

# Recorded judge verdicts, so this file runs with no credentials and no network.
RECORDED_VERDICTS = {
    "good_reply": {
        "results": [
            {
                "criterion": "addresses_the_specific_problem",
                "passed": True,
                "evidence": "two charges of 49 dollars on ACCT-482913",
            },
            {
                "criterion": "states_a_concrete_next_step",
                "passed": True,
                "evidence": "a billing specialist will review both charges today",
            },
            {
                "criterion": "makes_no_unfounded_promise",
                "passed": True,
                "evidence": "if the second charge is confirmed as duplicate",
            },
            {
                "criterion": "tone_is_calm_and_non_defensive",
                "passed": True,
                "evidence": "Thanks for flagging this.",
            },
        ]
    },
    "evasive_reply": {
        "results": [
            {
                "criterion": "addresses_the_specific_problem",
                "passed": False,
                "evidence": "we have received your billing inquiry",
            },
            {
                "criterion": "states_a_concrete_next_step",
                "passed": False,
                "evidence": "our team is looking into it",
            },
            {
                "criterion": "makes_no_unfounded_promise",
                "passed": True,
                "evidence": "no promise is made",
            },
            {
                "criterion": "tone_is_calm_and_non_defensive",
                "passed": True,
                "evidence": "we appreciate your patience",
            },
        ]
    },
    "overpromising_reply": {
        "results": [
            {
                "criterion": "addresses_the_specific_problem",
                "passed": True,
                "evidence": "the duplicate 49 dollar charge",
            },
            {
                "criterion": "states_a_concrete_next_step",
                "passed": True,
                "evidence": "I am refunding it now",
            },
            {
                "criterion": "makes_no_unfounded_promise",
                "passed": False,
                "evidence": "you will see the money back within one hour, guaranteed",
            },
            {
                "criterion": "tone_is_calm_and_non_defensive",
                "passed": True,
                "evidence": "Sorry about that.",
            },
        ]
    },
}


def call_judge(
    prompt: str,
    model: str = JUDGE_MODEL,
    temperature: float = JUDGE_TEMPERATURE,
    case_id: str | None = None,
) -> str:
    """Stand-in for a real judge call.

    Replace the body with your provider's chat completion, passing model and
    temperature straight through. The case_id argument exists only so the offline
    stub can replay a recorded verdict. A real implementation ignores it.
    """
    JUDGE_CALLS.append({"prompt": prompt, "model": model, "temperature": temperature})
    if case_id is None or case_id not in RECORDED_VERDICTS:
        raise AssertionError(f"no recorded verdict for case {case_id!r}")
    return json.dumps(RECORDED_VERDICTS[case_id])


def judge_reply(customer_message: str, reply: str, case_id: str) -> JudgeVerdict:
    prompt = build_judge_prompt(customer_message, reply)
    raw = call_judge(
        prompt,
        model=JUDGE_MODEL,
        temperature=JUDGE_TEMPERATURE,
        case_id=case_id,
    )
    return JudgeVerdict.model_validate_json(raw)


CUSTOMER_MESSAGE = (
    "I was charged twice for my subscription this month on ACCT-482913, two "
    "charges of 49 dollars on the same day. Please refund one of them."
)

GOOD_REPLY = (
    "Thanks for flagging this. I can see two charges of 49 dollars on "
    "ACCT-482913 dated the same day. A billing specialist will review both "
    "charges today, and if the second charge is confirmed as duplicate it will "
    "be reversed to your original payment method."
)

EVASIVE_REPLY = (
    "Hello, we have received your billing inquiry and our team is looking into "
    "it. We appreciate your patience and will be in touch."
)

OVERPROMISING_REPLY = (
    "Sorry about that. I am refunding the duplicate 49 dollar charge now and you "
    "will see the money back within one hour, guaranteed."
)


@pytest.fixture(autouse=True)
def _clear_recorded_calls():
    JUDGE_CALLS.clear()
    yield
    JUDGE_CALLS.clear()


def test_rubric_is_written_into_the_prompt() -> None:
    prompt = build_judge_prompt(CUSTOMER_MESSAGE, GOOD_REPLY)

    for name, definition in RUBRIC:
        assert name in prompt
        assert definition in prompt


def test_prompt_asks_for_named_criteria_not_a_vague_score() -> None:
    prompt = build_judge_prompt(CUSTOMER_MESSAGE, GOOD_REPLY).lower()

    for banned in ("1 to 5", "1-5", "out of 10", "rate this response", "score from"):
        assert banned not in prompt, f"judge prompt fell back to {banned!r}"

    assert "pass or fail" in prompt


def test_judge_model_is_pinned_to_a_dated_snapshot() -> None:
    # A floating alias would let the grader change without a commit.
    assert re.search(r"-\d{4}-\d{2}-\d{2}$", JUDGE_MODEL), (
        f"judge model {JUDGE_MODEL!r} is not pinned to a dated snapshot"
    )


def test_judge_is_called_at_temperature_zero_with_the_pinned_model() -> None:
    judge_reply(CUSTOMER_MESSAGE, GOOD_REPLY, case_id="good_reply")

    assert len(JUDGE_CALLS) == 1
    assert JUDGE_CALLS[0]["temperature"] == 0.0
    assert JUDGE_CALLS[0]["model"] == JUDGE_MODEL


def test_a_good_reply_passes_every_criterion() -> None:
    verdict = judge_reply(CUSTOMER_MESSAGE, GOOD_REPLY, case_id="good_reply")

    assert verdict.passed
    assert verdict.failures == []


def test_an_evasive_reply_fails_the_criteria_it_should() -> None:
    verdict = judge_reply(CUSTOMER_MESSAGE, EVASIVE_REPLY, case_id="evasive_reply")

    assert not verdict.passed
    # Named failures, so a red test says which property broke.
    assert verdict.failures == [
        "addresses_the_specific_problem",
        "states_a_concrete_next_step",
    ]


def test_an_overpromising_reply_fails_only_the_promise_criterion() -> None:
    verdict = judge_reply(
        CUSTOMER_MESSAGE, OVERPROMISING_REPLY, case_id="overpromising_reply"
    )

    assert verdict.failures == ["makes_no_unfounded_promise"]


def test_a_verdict_missing_a_criterion_is_rejected() -> None:
    partial = {"results": RECORDED_VERDICTS["good_reply"]["results"][:2]}

    with pytest.raises(ValidationError) as excinfo:
        JudgeVerdict.model_validate(partial)

    assert "does not cover the rubric" in str(excinfo.value)


def test_a_verdict_with_an_invented_criterion_is_rejected() -> None:
    invented = {
        "results": [
            *RECORDED_VERDICTS["good_reply"]["results"][:3],
            {"criterion": "vibes", "passed": True, "evidence": "felt fine"},
        ]
    }

    with pytest.raises(ValidationError):
        JudgeVerdict.model_validate(invented)


def test_a_verdict_without_evidence_is_rejected() -> None:
    unsupported = {
        "results": [
            {**result, "evidence": ""}
            for result in RECORDED_VERDICTS["good_reply"]["results"]
        ]
    }

    with pytest.raises(ValidationError):
        JudgeVerdict.model_validate(unsupported)

# Versioned prompt fixture for the ticket summarizer.
#
# One file per prompt version. The tests load this file rather than importing a
# string from application code, so a prompt edit shows up in a pull request as a
# reviewable diff with a test result attached.
#
# Key contract (do not rename without updating the tests that read them):
#   version, model, temperature   read by every test, asserted in CI
#   system_prompt, user_template  rendered by summarizer/prompt_runtime.py
#   golden_cases[].id             used to look up recorded responses in replay mode
#   golden_cases[].variables      inputs interpolated into user_template
#   golden_cases[].expected       deterministic field values (tests/test_structured_output.py)
#   golden_cases[].must_contain           literal substrings (tests/test_string_assertions.py)
#   golden_cases[].must_contain_patterns  regexes           (tests/test_string_assertions.py)
#   golden_cases[].must_not_contain       forbidden strings (tests/test_string_assertions.py)
#   golden_cases[].reference_summary      similarity target (tests/test_semantic_similarity.py)
#   golden_cases[].accepted_summaries     calibration set   (tests/test_semantic_similarity.py)

version: 3
model: gpt-4o-mini
temperature: 0
max_input_tokens: 1200

system_prompt: |
  You are a support ticket summarizer for a subscription billing product.

  Return a single JSON object and nothing else. No prose before or after it.

  Required fields:
    account_id    the account identifier exactly as it appears in the ticket, in the form ACCT-000000
    category      one of: billing, bug, how_to, other
    urgency       integer 1 to 5, where 5 means the customer is blocked from paying or using the product
    summary       two sentences at most, describing what the customer wants, in plain language
    needs_human   true when resolving the ticket requires an action only a human can take

  Rules:
    Never invent an account_id. If the ticket contains no identifier in the ACCT-000000 form, use the literal string "unknown".
    Never add a greeting, a sign-off, or an offer to help further. The output is consumed by a pipeline, not read by a customer.
    Never restate the ticket verbatim. Summarize it.

user_template: |
  Ticket ID: {ticket_id}
  Channel: {channel}

  Ticket body:
  {ticket_body}

golden_cases:
  - id: billing_duplicate_charge
    description: A clear billing complaint with an account id present in the body.
    variables:
      ticket_id: T-1041
      channel: email
      ticket_body: |
        Hi, I think something went wrong with my subscription payment this month.
        My account is ACCT-482913. I see two charges of 49 dollars on the same day,
        both from you, and I only have one subscription. Can you refund one of them?
    expected:
      account_id: ACCT-482913
      category: billing
      needs_human: true
    must_contain:
      - ACCT-482913
    must_contain_patterns:
      - ACCT-\d{6}
    must_not_contain:
      - Best regards
      - Thanks for reaching out
      - Let me know if there is anything else
    reference_summary: >-
      The customer was charged twice for the same monthly subscription on the same day
      and is asking for one of the two duplicate charges to be refunded.
    accepted_summaries:
      - >-
        The customer sees two identical subscription charges on the same day and wants
        one of the duplicate charges refunded.
      - >-
        A duplicate subscription charge was billed to the customer on the same day, and
        the customer is requesting a refund for the second charge.
      - >-
        The customer reports being billed twice for one monthly subscription and asks
        for a refund of the duplicate charge.

  - id: bug_login_redirect_loop
    description: A bug report where the account id appears mid sentence.
    variables:
      ticket_id: T-1103
      channel: in_app
      ticket_body: |
        Every time I sign in on Chrome the page sends me back to the login screen
        again. It just loops. Works fine in Safari. Account ACCT-771204 if that helps.
        I cannot get to the invoices page at all right now.
    expected:
      account_id: ACCT-771204
      category: bug
      needs_human: true
    must_contain:
      - ACCT-771204
    must_contain_patterns:
      - ACCT-\d{6}
    must_not_contain:
      - Best regards
      - Thanks for reaching out
      - Let me know if there is anything else
    reference_summary: >-
      Signing in on Chrome sends the customer back to the login screen in a redirect
      loop, which blocks the customer from reaching the invoices page, while Safari works.
    accepted_summaries:
      - >-
        The customer is stuck in a login redirect loop on Chrome and cannot reach the
        invoices page, though signing in on Safari works.
      - >-
        On Chrome the login page loops back to itself for this customer, blocking access
        to the invoices page, while the same sign in works on Safari.
      - >-
        A redirect loop on the Chrome login screen prevents the customer from reaching
        invoices, and Safari is unaffected.

  - id: how_to_change_billing_email
    description: A how-to question with no account id in the body, so the fallback applies.
    variables:
      ticket_id: T-1188
      channel: email
      ticket_body: |
        Where do I change the email address that receives the monthly invoice?
        I looked in the profile settings and only found the login email.
    expected:
      account_id: unknown
      category: how_to
      needs_human: false
    must_contain:
      - unknown
    must_contain_patterns:
      - "\"account_id\"\\s*:\\s*\"unknown\""
    must_not_contain:
      - Best regards
      - Thanks for reaching out
      - Let me know if there is anything else
    reference_summary: >-
      The customer wants to know where to change the email address that receives the
      monthly invoice, having only found the login email in profile settings.
    accepted_summaries:
      - >-
        The customer is asking where the invoice email address is changed, since profile
        settings only exposed the login email.
      - >-
        The customer wants to change which email address receives the monthly invoice and
        could not find that setting next to the login email.
      - >-
        Asking how to change the billing email that receives the monthly invoice, because
        only the login email appears in profile settings.

# yaml-language-server: $schema=https://promptfoo.dev/config-schema.json
#
# Config-first prompt testing: prompts, cases, and graders declared as data.
# Run it with:
#
#   npx promptfoo@latest eval          # full matrix
#   npx promptfoo@latest eval --filter-first-n 2   # the pull-request subset
#   npx promptfoo@latest view          # browse the result grid
#
# Needs OPENAI_API_KEY. The assertions below are Promptfoo built-ins, matching the
# assertion types in tests/: is-json for the schema, contains and regex and
# not-icontains for the string contract, similar for the semantic check, and
# llm-rubric for the one property nothing cheaper expresses.

description: summarize_ticket v3, matrix eval across providers and golden cases

prompts:
  - file://prompts/summarize_ticket/promptfoo_prompt.yaml

providers:
  - id: openai:chat:gpt-4o-mini
    config:
      temperature: 0
      response_format:
        type: json_object

  # Uncomment to matrix the same cases across a second model. This is what
  # Promptfoo is good at: the grid, not the individual assertion.
  # - id: openai:chat:gpt-4o-2024-08-06
  #   config:
  #     temperature: 0
  #     response_format:
  #       type: json_object

# Applied to every case below, so a shared contract is declared once.
defaultTest:
  assert:
    - type: is-json
      value:
        type: object
        required:
          - account_id
          - category
          - urgency
          - summary
          - needs_human
        properties:
          account_id:
            type: string
            pattern: "^(ACCT-[0-9]{6}|unknown)$"
          category:
            type: string
            enum: [billing, bug, how_to, other]
          urgency:
            type: integer
            minimum: 1
            maximum: 5
          summary:
            type: string
            minLength: 10
          needs_human:
            type: boolean
        additionalProperties: false

    # The sign-off boilerplate the prompt forbids.
    - type: not-icontains
      value: Best regards
    - type: not-icontains
      value: Thanks for reaching out
    - type: not-icontains
      value: Let me know if there is anything else

tests:
  - description: billing duplicate charge, account id present in the body
    vars:
      ticket_id: T-1041
      channel: email
      ticket_body: |
        Hi, I think something went wrong with my subscription payment this month.
        My account is ACCT-482913. I see two charges of 49 dollars on the same day,
        both from you, and I only have one subscription. Can you refund one of them?
    assert:
      - type: contains
        value: ACCT-482913
      - type: regex
        value: "ACCT-[0-9]{6}"
      - type: javascript
        value: JSON.parse(output).category === 'billing'
      - type: javascript
        value: JSON.parse(output).needs_human === true
      # Threshold calibrated the same way as tests/test_semantic_similarity.py:
      # just below the worst known-good paraphrase, not at a round number.
      - type: similar
        value: >-
          The customer was charged twice for the same monthly subscription on the same day
          and is asking for one of the two duplicate charges to be refunded.
        threshold: 0.72

  - description: bug report, account id mid sentence
    vars:
      ticket_id: T-1103
      channel: in_app
      ticket_body: |
        Every time I sign in on Chrome the page sends me back to the login screen
        again. It just loops. Works fine in Safari. Account ACCT-771204 if that helps.
        I cannot get to the invoices page at all right now.
    assert:
      - type: contains
        value: ACCT-771204
      - type: javascript
        value: JSON.parse(output).category === 'bug'
      - type: similar
        value: >-
          Signing in on Chrome sends the customer back to the login screen in a redirect
          loop, which blocks the customer from reaching the invoices page, while Safari works.
        threshold: 0.72

  - description: how-to question with no account id, fallback applies
    vars:
      ticket_id: T-1188
      channel: email
      ticket_body: |
        Where do I change the email address that receives the monthly invoice?
        I looked in the profile settings and only found the login email.
    assert:
      - type: javascript
        value: JSON.parse(output).account_id === 'unknown'
      - type: javascript
        value: JSON.parse(output).category === 'how_to'
      # No invented identifier, which is the failure mode this case exists for.
      - type: not-regex
        value: "ACCT-[0-9]{6}"

  - description: ambiguous one-line ticket, the case a judge earns its cost on
    vars:
      ticket_id: T-1201
      channel: chat
      ticket_body: |
        it broke again
    assert:
      - type: javascript
        value: "['bug', 'other'].includes(JSON.parse(output).category)"
      - type: llm-rubric
        value: >-
          The summary states plainly that the ticket does not say what broke, and does
          not invent a specific product area, error, or cause that the ticket never
          mentioned. Pass or fail only.

"""The same idea as the rest of this suite, expressed in DeepEval.

Contrast with promptfooconfig.yaml. That file declares prompts, cases, and
graders as data and runs them with `promptfoo eval`. This file is an ordinary
pytest module: DeepEval metrics are constructed in Python and handed to
assert_test, so prompt tests live next to every other test you already have.

Neither is a better tool. Pick the one that matches where your team already
works. Both are ways to run the assertion types in this repo with less
boilerplate, not different quality tiers.

This module is marked `live`: DeepEval's metrics are themselves model calls, and
the summarizer output being graded comes from a real model. It is skipped when
DeepEval is not installed or no key is set, which is why the every-commit tier
stays free.

Run it:  pip install -r requirements-eval.txt && OPENAI_API_KEY=... pytest -m live tests/test_deepeval_metrics.py
"""

from __future__ import annotations

import os

import pytest

deepeval = pytest.importorskip(
    "deepeval", reason="pip install -r requirements-eval.txt to run this tier"
)

from deepeval import assert_test  # noqa: E402
from deepeval.metrics import AnswerRelevancyMetric  # noqa: E402
from deepeval.test_case import LLMTestCase  # noqa: E402

from summarizer.prompt_runtime import (  # noqa: E402
    load_prompt,
    set_model_backend,
    summarize_ticket,
)

pytestmark = [
    pytest.mark.live,
    pytest.mark.skipif(
        not os.getenv("OPENAI_API_KEY"),
        reason="live tier needs OPENAI_API_KEY",
    ),
]

# The grader model, pinned for the same reason the judge model is pinned in
# tests/test_llm_judge.py: a metric whose model drifts is a metric whose history
# is not comparable.
GRADER_MODEL = "gpt-4o-mini-2024-07-18"

RELEVANCY_THRESHOLD = 0.7

PROMPT = load_prompt()
GOLDEN_CASES = PROMPT["golden_cases"]
CASE_IDS = [case["id"] for case in GOLDEN_CASES]


def openai_backend(system: str, user: str) -> str:
    """A real model call, using the model and temperature from the fixture."""
    from openai import OpenAI

    client = OpenAI()
    response = client.chat.completions.create(
        model=PROMPT["model"],
        temperature=PROMPT["temperature"],
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
    )
    return response.choices[0].message.content or ""


@pytest.fixture(autouse=True)
def _use_a_real_model() -> None:
    set_model_backend(openai_backend)


@pytest.mark.parametrize("case", GOLDEN_CASES, ids=CASE_IDS)
def test_summary_is_relevant_to_the_ticket(case: dict) -> None:
    result = summarize_ticket(case["variables"])

    test_case = LLMTestCase(
        input=case["variables"]["ticket_body"],
        actual_output=result["summary"],
        expected_output=case["reference_summary"],
    )
    relevancy = AnswerRelevancyMetric(
        threshold=RELEVANCY_THRESHOLD,
        model=GRADER_MODEL,
        include_reason=True,
    )

    # assert_test raises with the metric's reason attached on failure, so a red
    # CI run tells you why the summary was judged irrelevant.
    assert_test(test_case, [relevancy])


@pytest.mark.parametrize("case", GOLDEN_CASES, ids=CASE_IDS)
def test_deterministic_fields_still_hold_against_a_live_model(case: dict) -> None:
    """The cheap assertions do not stop mattering once a metric is involved.

    Same expectations as tests/test_structured_output.py, run against a real
    response. This is the pull-request tier: a handful of golden cases, real cost,
    bounded.
    """
    result = summarize_ticket(case["variables"])

    for field, expected_value in case["expected"].items():
        assert result[field] == expected_value, (
            f"{case['id']}: expected {field}={expected_value!r}, got {result[field]!r}"
        )

"""Tier 1: everything that tests your code rather than the model.

Template rendering, variable interpolation, the token budget guard, and what
happens when the model returns garbage. None of it needs a live API call, all of
it runs in milliseconds, and the _no_network fixture below makes that a checked
property rather than a claim: any attempt to open a socket in this file fails the
test.

This is the tier that runs on every commit. There is no cost argument against it.

Run it:  pytest tests/test_prompt_logic_mocked.py -v
"""

from __future__ import annotations

import socket
from unittest.mock import Mock

import pytest

from summarizer.prompt_runtime import (
    CHARS_PER_TOKEN,
    MalformedModelResponse,
    MissingPromptVariable,
    ModelBackendNotConfigured,
    PromptTooLarge,
    build_request,
    check_token_budget,
    estimate_tokens,
    generate,
    load_prompt,
    parse_model_json,
    render_prompt,
    reset_model_backend,
    set_model_backend,
    summarize_ticket,
    template_variables,
)

VALID_RESPONSE = (
    '{"account_id": "ACCT-482913", "category": "billing", "urgency": 4, '
    '"summary": "Two charges for one subscription. The customer wants one '
    'refunded.", "needs_human": true}'
)


@pytest.fixture(autouse=True)
def _no_network(monkeypatch: pytest.MonkeyPatch) -> None:
    """Turn "this tier makes no API calls" into an assertion."""

    def _blocked(*args, **kwargs):
        raise AssertionError(
            "tier 1 opened a network connection; it is supposed to be mocked"
        )

    monkeypatch.setattr(socket, "socket", _blocked)
    monkeypatch.setattr(socket, "create_connection", _blocked)


@pytest.fixture(autouse=True)
def _isolated_backend():
    """No test in this file inherits another test's model backend."""
    reset_model_backend()
    yield
    reset_model_backend()


@pytest.fixture
def prompt() -> dict:
    return load_prompt()


@pytest.fixture
def variables(prompt: dict) -> dict:
    return dict(prompt["golden_cases"][0]["variables"])


def test_template_declares_the_variables_we_think_it_does(prompt: dict) -> None:
    assert template_variables(prompt["user_template"]) == {
        "ticket_id",
        "channel",
        "ticket_body",
    }


def test_render_interpolates_every_variable(prompt: dict, variables: dict) -> None:
    rendered = render_prompt(prompt["user_template"], variables)

    assert "T-1041" in rendered
    assert "ACCT-482913" in rendered
    assert "{" not in rendered  # no placeholder survived


def test_render_fails_loudly_on_a_missing_variable(prompt: dict) -> None:
    with pytest.raises(MissingPromptVariable) as excinfo:
        render_prompt(prompt["user_template"], {"ticket_id": "T-1"})

    # Silently rendering "None" into a prompt is how a prompt bug becomes a
    # data bug three services downstream.
    assert excinfo.value.missing == ["channel", "ticket_body"]


def test_render_leaves_literal_json_braces_alone() -> None:
    template = 'Return {"ok": true} for ticket {ticket_id}.'

    assert render_prompt(template, {"ticket_id": "T-9"}) == (
        'Return {"ok": true} for ticket T-9.'
    )


def test_render_ignores_extra_variables(prompt: dict, variables: dict) -> None:
    rendered = render_prompt(prompt["user_template"], {**variables, "unused": "x"})

    assert "unused" not in rendered
    assert "T-1041" in rendered


def test_token_estimate_tracks_length() -> None:
    assert estimate_tokens("") == 0
    assert estimate_tokens("a" * (CHARS_PER_TOKEN * 10)) == 10
    assert estimate_tokens("a" * (CHARS_PER_TOKEN * 10 + 1)) == 11


def test_token_budget_passes_under_the_limit() -> None:
    assert check_token_budget("a" * 40, max_tokens=100) == 10


def test_token_budget_fails_over_the_limit() -> None:
    with pytest.raises(PromptTooLarge) as excinfo:
        check_token_budget("a" * 40_000, max_tokens=1_200)

    assert excinfo.value.max_tokens == 1_200
    assert excinfo.value.estimated_tokens == 10_000


def test_a_pasted_log_dump_is_rejected_before_it_costs_anything(
    prompt: dict, variables: dict
) -> None:
    # The realistic failure: a customer pastes a 50k character stack trace.
    variables["ticket_body"] = "ERROR at line 1\n" * 4_000

    with pytest.raises(PromptTooLarge):
        build_request(variables, prompt=prompt)


def test_build_request_reports_a_token_estimate(prompt: dict, variables: dict) -> None:
    request = build_request(variables, prompt=prompt)

    assert request.system == prompt["system_prompt"]
    assert "T-1041" in request.user
    assert 0 < request.estimated_tokens <= prompt["max_input_tokens"]


def test_summarize_ticket_parses_a_mocked_response(
    prompt: dict, variables: dict
) -> None:
    backend = Mock(return_value=VALID_RESPONSE)
    set_model_backend(backend)

    result = summarize_ticket(variables, prompt=prompt)

    assert result["account_id"] == "ACCT-482913"
    assert result["needs_human"] is True

    # The mock also lets us assert on what we sent, which a live call makes awkward.
    backend.assert_called_once()
    sent_system, sent_user = backend.call_args.args
    assert "Return a single JSON object" in sent_system
    assert "ACCT-482913" in sent_user


def test_malformed_json_raises_and_keeps_the_raw_text(
    prompt: dict, variables: dict
) -> None:
    # The classic: a chatty preamble in front of otherwise valid JSON.
    set_model_backend(lambda system, user: 'Sure! Here you go:\n{"account_id": "A"}')

    with pytest.raises(MalformedModelResponse) as excinfo:
        summarize_ticket(variables, prompt=prompt)

    # Keeping the raw response on the exception is what makes the CI log useful.
    assert excinfo.value.raw.startswith("Sure!")


def test_truncated_json_raises(prompt: dict, variables: dict) -> None:
    set_model_backend(lambda system, user: '{"account_id": "ACCT-482913", "cat')

    with pytest.raises(MalformedModelResponse):
        summarize_ticket(variables, prompt=prompt)


def test_a_json_array_is_not_an_acceptable_object() -> None:
    with pytest.raises(MalformedModelResponse) as excinfo:
        parse_model_json('[{"account_id": "ACCT-482913"}]')

    assert "expected a JSON object" in str(excinfo.value)


def test_an_empty_response_raises() -> None:
    with pytest.raises(MalformedModelResponse):
        parse_model_json("")


def test_generate_refuses_to_run_without_a_configured_backend() -> None:
    # The default state of this module is "no network", which is why the tier is
    # safe to run anywhere.
    with pytest.raises(ModelBackendNotConfigured):
        generate("system", "user")


def test_the_mocked_backend_is_the_only_thing_called(
    prompt: dict, variables: dict
) -> None:
    calls: list[tuple[str, str]] = []

    def fake_model(system: str, user: str) -> str:
        calls.append((system, user))
        return VALID_RESPONSE

    set_model_backend(fake_model)
    summarize_ticket(variables, prompt=prompt)
    summarize_ticket(variables, prompt=prompt)

    assert len(calls) == 2

name: prompt-tests

# Three tiers, three costs.
#
#   Tier 1  every push          mocked and replayed tests, no credentials, seconds
#   Tier 2  every pull request  a small live golden subset, real cost, bounded
#   Tier 3  nightly schedule    the full eval suite, paid once a day
#
# Each tier maps to specific files, listed explicitly rather than hidden behind a
# glob, so adding a slow test to the fast tier has to be a deliberate edit.

on:
  push:
    branches: ["**"]
  pull_request:
  schedule:
    # 06:00 UTC daily.
    - cron: "0 6 * * *"
  workflow_dispatch:

concurrency:
  group: prompt-tests-${{ github.ref }}
  cancel-in-progress: true

permissions:
  contents: read

jobs:
  tier-1-mocked:
    name: "Tier 1: mocked and replayed (every push)"
    runs-on: ubuntu-latest
    timeout-minutes: 10
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: pip

      - name: Install core dependencies
        run: pip install -r requirements.txt

      # No secrets are provided to this job at all, which is the point: if one of
      # these tests ever needs a key, it does not belong in this tier.
      - name: Run tier 1
        run: |
          pytest -m "not live" \
            tests/test_prompt_logic_mocked.py \
            tests/test_structured_output.py \
            tests/test_string_assertions.py \
            tests/test_semantic_similarity.py \
            tests/test_llm_judge.py

  tier-2-live-subset:
    name: "Tier 2: live golden subset (pull requests)"
    if: github.event_name == 'pull_request'
    needs: tier-1-mocked
    runs-on: ubuntu-latest
    timeout-minutes: 20
    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: pip

      - uses: actions/setup-node@v4
        with:
          node-version: "20"

      - name: Install core and eval dependencies
        run: pip install -r requirements.txt -r requirements-eval.txt

      - name: DeepEval metrics on the golden cases
        run: pytest -m live tests/test_deepeval_metrics.py

      # A bounded slice of the Promptfoo matrix, so per-PR spend is predictable.
      - name: Promptfoo, first two cases only
        run: npx --yes promptfoo@latest eval --filter-first-n 2 --no-cache

      - name: Upload Promptfoo results
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: promptfoo-pr-subset
          path: ~/.promptfoo/output
          if-no-files-found: ignore

  tier-3-full-eval:
    name: "Tier 3: full eval suite (nightly)"
    if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch'
    needs: tier-1-mocked
    runs-on: ubuntu-latest
    timeout-minutes: 60
    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: pip

      - uses: actions/setup-node@v4
        with:
          node-version: "20"

      - name: Install core and eval dependencies
        run: pip install -r requirements.txt -r requirements-eval.txt

      - name: Every test, mocked and live
        run: pytest tests/

      - name: Promptfoo, full matrix
        run: npx --yes promptfoo@latest eval --no-cache

      - name: Upload Promptfoo results
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: promptfoo-nightly
          path: ~/.promptfoo/output
          if-no-files-found: ignore