tests/test_deploy.py

#!/usr/bin/env python3
"""Checks for the deploy path. No GPU, no index, no network.



  python tests/test_deploy.py



Everything here guards a failure that is silent in production: a shard filter

that matches nothing, a language list that never loads, an empty CORS flag that

blocks the browser, or a duplicated language map that drifted.

"""
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from deploy.bake import MMS_LANG as BAKE_LANG          # noqa: E402
from deploy.boot import build_cmd                      # noqa: E402
from scripts.build_and_publish import val_shards       # noqa: E402
from src.guardrails import (GuardrailConfig, Guardrails,  # noqa: E402
                            calibrate_topic_threshold)
from src.router import Passage                          # noqa: E402
from src.voice import MMS_LANG as VOICE_LANG           # noqa: E402

REPO = Path(__file__).resolve().parents[1]


def test_val_shards_picks_the_right_languages():
    # The REAL ai4bharat/MSMARCO-XI layout, read off the hub. An earlier version
    # of this test invented "hin_Deva_validation-00000-of-00002.parquet" and
    # passed while the actual build matched nothing -- a fixture is only worth
    # what its resemblance to production is.
    files = [
        ".gitattributes", "README.md", "ms_marco_translations.py",
        "validation/hinval.parquet", "validation/benval.parquet",
        "validation/kanval.parquet", "validation/marval.parquet",
        "validation/tamval.parquet", "validation/urdval.parquet",
        "train/hintrain.parquet", "train/bentrain.parquet",
    ]
    got = val_shards(files, {"hi", "bn", "kn", "mr"})
    assert sorted(got) == ["validation/benval.parquet", "validation/hinval.parquet",
                           "validation/kanval.parquet", "validation/marval.parquet"], got

    # "validation/" contains the substring "val", so filtering on the full path
    # would drag every train shard in with it. The filter must use .name.
    assert not any("train" in f for f in val_shards(files, {"hi"})), "train leaked in"
    assert not any(f.endswith((".md", ".py")) for f in got), got
    assert val_shards(files, {"zz"}) == []


def test_download_lands_where_the_schema_step_looks():
    # 02_inspect_schema.py:find_parquets scans root/hf_cache only. Downloading to
    # root/data made it find zero parquet, exit non-zero, and stop the build
    # before any index existed -- with the failure three steps upstream of the
    # traceback the user actually saw.
    schema = (REPO / "scripts" / "02_inspect_schema.py").read_text(encoding="utf-8")
    build = (REPO / "scripts" / "build_and_publish.py").read_text(encoding="utf-8")
    assert 'root / "hf_cache"' in schema, "the scan root moved; update the download dest"
    assert 'root / "hf_cache" / "MSMARCO-XI"' in build, "download dest is not under hf_cache"


def test_bake_language_map_has_not_drifted():
    # deploy/bake.py duplicates the map because it runs before src/ is copied
    # into the image. If they disagree, the image bakes the wrong voice and the
    # first spoken request downloads 145 MB while a judge watches.
    assert BAKE_LANG == VOICE_LANG, {
        k: (BAKE_LANG.get(k), VOICE_LANG.get(k))
        for k in set(BAKE_LANG) | set(VOICE_LANG)
        if BAKE_LANG.get(k) != VOICE_LANG.get(k)
    }


def test_build_cmd_omits_flags_for_unset_variables():
    base = build_cmd({}, python="py")
    assert "--cors" not in base, base
    assert "--langs" not in base, base
    assert base[-2:] == ["--port", "7860"], base

    full = build_cmd({"PORT": "8010", "VOICERAG_CORS": "https://x.vercel.app",
                      "VOICERAG_PREWARM": "hi,bn", "VOICERAG_LANGS": " hi "},
                     python="py")
    assert "8010" in full, full
    assert full[full.index("--cors") + 1] == "https://x.vercel.app", full
    assert full[full.index("--langs") + 1] == "hi", full   # stripped

    # An empty-string variable is the Space UI's idea of "unset".
    assert "--cors" not in build_cmd({"VOICERAG_CORS": "   "}, python="py")


def test_tau_topic_lets_98_percent_of_real_queries_through():
    scores = [i / 1000 for i in range(1000)]          # 0.000 .. 0.999
    tau = calibrate_topic_threshold(scores, max_false_refusal=0.02)
    refused = sum(1 for s in scores if s < tau)
    assert refused / len(scores) <= 0.02, (tau, refused)
    assert calibrate_topic_threshold([]) == 0.35      # documented placeholder


def test_serve_derives_languages_from_the_manifest():
    # index_build.py writes indices/model/strategies and never a "languages"
    # key. serve.py used to read one, so every run without --langs raised
    # KeyError. Assert the manifest shape and that serve.py no longer asks.
    src = (REPO / "src" / "serve.py").read_text(encoding="utf-8")
    assert '["languages"]' not in src, "serve.py is reading a key that is never written"
    assert 'manifest.get("indices", {})' in src

    manifest = {"model": "BAAI/bge-m3", "strategies": ["FW", "FCC"],
                "default_strategy": "FW", "indices": {
                    "hi__FW": {"lang": "hi", "strategy": "FW"},
                    "hi__FCC": {"lang": "hi", "strategy": "FCC"},
                    "bn__FW": {"lang": "bn", "strategy": "FW"}}}
    langs = sorted({v["lang"] for v in manifest.get("indices", {}).values()})
    assert langs == ["bn", "hi"], langs
    assert json.loads(json.dumps(manifest))["default_strategy"] == "FW"


# --------------------------------------------------------------- gate 3


def _passages(scores):
    return [Passage(f"c{i}", f"text {i}", s, "kn") for i, s in enumerate(scores)]


def _write_cal(tmp, **kw):
    d = {"topic_signal": "spread", "tau_topic": 0.04, "gate_depth": 20, "n": 400}
    d.update(kw)
    tmp.write_text(json.dumps(d))
    return tmp


def test_tau_topic_cannot_return_the_other_signals_scale():
    # Bug 11: `spread` shipped paired with 0.35 -- a top-1-scale value about ten
    # times any spread observed -- and the gate abstained on all 300 benchmark
    # queries while reporting a fast pipeline that never ran the reader.
    top1 = GuardrailConfig()
    assert top1.topic_signal == "top1" and top1.tau_topic == top1.tau_top1

    spread = GuardrailConfig(topic_signal="spread")
    assert spread.tau_topic == spread.tau_spread
    assert spread.tau_topic < 0.25, "a spread tau on a top-1 scale is bug 11"


def test_load_refuses_a_threshold_written_on_the_wrong_scale():
    import tempfile
    d = Path(tempfile.mkdtemp())

    ok = _write_cal(d / "ok.json", topic_signal="spread", tau_topic=0.0393)
    cfg = GuardrailConfig.load(ok)
    assert cfg.topic_signal == "spread" and cfg.tau_topic == 0.0393
    assert cfg.topic_calibrated and cfg.gate_depth == 20

    for bad, why in [
        (_write_cal(d / "b1.json", topic_signal="spread", tau_topic=0.35),
         "spread with a top-1 tau"),
        (_write_cal(d / "b2.json", topic_signal="top1", tau_topic=0.04),
         "top1 with a spread tau"),
        (_write_cal(d / "b3.json", topic_signal="nonsense", tau_topic=0.3),
         "unknown signal"),
    ]:
        try:
            GuardrailConfig.load(bad)
        except ValueError:
            continue
        raise AssertionError(f"load() accepted {why}")

    # A missing file is a placeholder, not a crash.
    assert GuardrailConfig.load(d / "nope.json").topic_calibrated is False


def test_spread_reads_shape_not_height():
    cfg = GuardrailConfig(topic_signal="spread", tau_spread=0.05, gate_depth=20)
    g = Guardrails(cfg)

    # A peaked neighbourhood: high rank-1, long flat tail. Answerable.
    peaked = _passages([0.72] + [0.40] * 19)
    v, detail = g.topic_value(peaked)
    assert abs(v - 0.32) < 1e-6, v
    assert g.check_topical(peaked).action.name == "ALLOW"

    # A hub: rank-1 just as high, but everything else is nearly as close. This
    # is the case top-1 cannot see -- both have top_score 0.72.
    hub = _passages([0.72] + [0.70] * 19)
    v2, _ = g.topic_value(hub)
    assert abs(v2 - 0.02) < 1e-6, v2
    assert g.check_topical(hub).action.name == "ABSTAIN"

    top1 = Guardrails(GuardrailConfig(topic_signal="top1", tau_top1=0.35))
    assert top1.check_topical(hub).action.name == "ALLOW",         "top-1 allows the hub; that is exactly why spread was measured"


def test_short_depth_is_reported_not_hidden():
    g = Guardrails(GuardrailConfig(topic_signal="spread", gate_depth=20))
    _, detail = g.topic_value(_passages([0.7, 0.6, 0.5]))
    assert detail["depth"] == 3
    assert "short_depth" in detail, "a shape from 3 points must say so"


def test_gate_3_retrieves_the_depth_it_was_calibrated_on():
    # "A shape read from 5 points is not the shape the threshold was calibrated
    # on." The harness must retrieve gate_depth and slice k for the reader.
    h = (REPO / "src" / "harness.py").read_text(encoding="utf-8")
    assert "max(req.k, self.guards.cfg.gate_depth)" in h
    assert 'ctx["gate_passages"] = deep' in h,         "the gate must see the raw dense set, in its original order"
    # The reader's k may be reordered by the lexical rerank, but it must still
    # be derived from `deep` and it must not write back into gate_passages --
    # tau was calibrated on the shape of an untouched dense neighbourhood.
    assert 'ctx["passages"] = _lexical_rerank(ctx["query"], deep, req.k)' in h
    r = h[h.index("def _lexical_rerank"):h.index("class Harness")]
    body = r.split('"""')[2]          # skip the docstring; it names both on purpose
    assert ".score =" not in body and "gate_passages" not in body,         "the rerank must not mutate scores or touch the gate's set"


def test_describe_quotes_the_measured_verbatim_rate():
    """Bug 13 twice over. 11.1% is the English *absent* rate, and the 8.2% the

    handoff offered as its correction reconciles with nothing in the committed

    JSON. Assert /guardrails against the evidence file, not against either."""
    d = Guardrails(GuardrailConfig()).describe()
    cost = d["hallucination"]["cost"]

    ext = json.loads((REPO / "results" / "extractability_translated.json")
                     .read_text(encoding="utf-8"))
    by = ext["by_level"]
    verbatim = 100 * by["exact"] / sum(by.values())
    assert abs(verbatim - 9.2) < 0.1, verbatim
    assert f"{verbatim:.1f}%" in cost, (verbatim, cost)

    eng = json.loads((REPO / "results" / "extractability_english.json")
                     .read_text(encoding="utf-8"))
    eby = eng["by_level"]
    assert f"{100 * eby['exact'] / sum(eby.values()):.1f}%" in cost
    assert "11.1% appear verbatim" not in cost
    assert "8.2%" not in cost, "the unreconciled handoff figure is back"


# ----------------------------------------------------------- space card


def test_space_card_matches_the_entrypoint_that_exists():
    # Docker SDK is PRO-only, so this Space runs the Gradio base image and
    # executes app_file. A card naming a file that is not there fails at build
    # time with a message about Gradio, not about the missing file.
    card = (REPO / "README.md").read_text(encoding="utf-8")
    assert card.startswith("---"), "HF reads the card from the first bytes"
    front = card.split("---", 2)[1]
    assert "sdk: gradio" in front, front
    assert "app_file: app.py" in front, front
    assert "app_port" not in front, "app_port is Docker-only and is ignored here"
    assert (REPO / "app.py").exists()

    # The Gradio SDK installs the ROOT requirements.txt, not env/.
    reqs = (REPO / "requirements.txt").read_text(encoding="utf-8")
    pinned = [ln.strip() for ln in reqs.splitlines()
              if ln.strip() and not ln.lstrip().startswith("#")]
    # ZeroGPU rejects anything outside this set, and matches the exact
    # version string -- so a +cpu local version is rejected too.
    assert any(ln == f"torch=={v}" for ln in pinned
               for v in ("2.8.0", "2.9.1", "2.10.0", "2.11.0")), pinned
    assert not any(ln.startswith("gradio") for ln in pinned),         "gradio comes from the base image; pinning it fights sdk_version"


def test_both_entrypoints_agree_on_serve_flags():
    # app.py slices build_cmd()[2:] to drop the interpreter and the script path,
    # then feeds the rest to serve.py's own parser. If those two ever disagree
    # the Space starts with the wrong configuration rather than failing.
    import src.serve as serve

    env = {"PORT": "7860", "VOICERAG_CORS": "https://x.vercel.app",
           "VOICERAG_PREWARM": "hi,bn,kn,mr", "VOICERAG_LANGS": "hi,bn"}
    cmd = build_cmd(env)
    assert cmd[1] == "src/serve.py", cmd

    args = serve.parser().parse_args(cmd[2:])
    assert args.port == 7860
    assert args.host == "0.0.0.0"
    assert args.cors == "https://x.vercel.app"
    assert args.prewarm_tts == "hi,bn,kn,mr"
    assert args.langs == "hi,bn"

    # And the defaults path, which is what a bare Space actually runs.
    bare = serve.parser().parse_args(build_cmd({})[2:])
    assert bare.port == 7860 and bare.cors == "" and bare.prewarm_tts == ""

    app_src = (REPO / "app.py").read_text(encoding="utf-8")
    assert "build_cmd({**os.environ, \"PORT\": str(PORT)})[2:]" in app_src


def test_serve_can_build_the_app_without_binding_a_port():
    # The whole Gradio adaptation rests on this split: bootstrap() does
    # everything main() does except bind, so a host that owns the port reuses it.
    import src.serve as serve

    assert callable(serve.bootstrap) and callable(serve.parser)
    assert callable(serve.main) and callable(serve.build_app)
    src_txt = (REPO / "src" / "serve.py").read_text(encoding="utf-8")
    assert "uvicorn.run" not in src_txt.split("def main()")[0],         "bootstrap must not bind a port"


def test_chunk_meta_is_json_serialisable():
    """The build embeds first and serialises second, so a bound method in the

    meta dict costs the whole GPU run before it surfaces. Chunk.chunk_id is a

    method while Chunk.n_words beside it is a property, which is the trap."""
    from src.chunkers.base import Document
    from src.chunkers.strategies import build as build_chunker

    doc = Document.from_blocks(
        "q1", "kn", ["a corporation is a legal entity " * 12,
                     "shareholders elect a board " * 12], ["q1:0", "q1:1"])
    chunks = build_chunker("FW").chunk(doc)
    assert chunks, "fixture produced no chunks"

    meta = [{"chunk_id": c.chunk_id(), "doc_id": doc.doc_id,
             "block_ids": list(c.block_ids or [])} for c in chunks]
    json.dumps({"texts": [c.text for c in chunks], "meta": meta},
               ensure_ascii=False)

    assert callable(chunks[0].chunk_id), "chunk_id stopped being a method"
    assert isinstance(chunks[0].n_words, int), "n_words stopped being a property"

    src = (REPO / "src" / "index_build.py").read_text(encoding="utf-8")
    assert '"chunk_id": c.chunk_id()' in src, "index_build lost the parens again"


def test_zerogpu_function_is_declared():
    """ZeroGPU refuses to start a Space with no @spaces.GPU function, restarts

    it, and the second process dies binding 7860 -- so the crash reads as a

    port clash and the cause reads as a warning. Declaring one costs nothing:

    quota is billed per call, and nothing calls this one."""
    src = (REPO / "app.py").read_text(encoding="utf-8")
    assert "@spaces.GPU" in src, "ZeroGPU startup check will fail the Space"
    assert "import spaces" in src, "@spaces.GPU without the import is a no-op"
    # It is now referenced on purpose -- ZeroGPU only finds a function a gradio
    # event reaches. What must never appear is a CALL, which is the only thing
    # that spends the 5-minute daily allowance. Passing it to .click() is not
    # a call; "_gpu_probe(" is, and the def line is the sole legitimate one.
    assert src.count("_gpu_probe(") == 1, "the probe is invoked -- that burns quota"
    assert ".click(_gpu_probe" in src, "unreferenced: ZeroGPU will not detect it"


def test_zerogpu_shape_is_launch_not_uvicorn():
    """ZeroGPU is Gradio-SDK-only and its startup scan runs off gradio's

    launch, so serving uvicorn ourselves fails the Space even when the port

    binds cleanly -- which is exactly what happened. On a Space gradio must

    launch and the API rides inside it; off-Space the simpler shape stands."""
    src = (REPO / "app.py").read_text(encoding="utf-8")
    assert "demo.launch(" in src, "no launch: ZeroGPU will not detect the Space"
    assert "ON_SPACE" in src, "the two shapes are not separated"
    # A Mount, not a route splice: routes would lose the API's CORS middleware,
    # and middleware cannot be re-added after the app has started.
    assert 'Mount("/", app=app)' in src, "API must mount, to keep its middleware"


def test_gpu_probe_survives_a_missing_spaces_package():
    """Off-Space the import fails, and a bare `except: print` would leave

    _gpu_probe undefined for the console to reference -- a NameError that only

    ever fires where nobody is watching."""
    src = (REPO / "app.py").read_text(encoding="utf-8")
    blk = src[src.index("try:" + chr(10) + "    import spaces"):src.index("from deploy.boot")]
    assert "_gpu_probe = None" in blk, "except branch leaves _gpu_probe undefined"
    ns = {}
    exec(compile(blk.replace("import spaces", "import spaces_absent_xyz"),
                 "app.py", "exec"), ns)
    assert ns["_gpu_probe"] is None


def test_gradio_ssr_is_off_before_the_import():
    """Gradio 5+ SSR spawns a Node server that takes the PUBLIC port and

    proxies to Python. That is what produced "[Errno 98] address already in

    use" on 7860. The env var must be set BEFORE gradio is imported, or gradio

    has already resolved the default."""
    src = (REPO / "app.py").read_text(encoding="utf-8")
    assert "GRADIO_SSR_MODE" in src, "SSR default restored; Node will take 7860"
    assert (src.index('os.environ["GRADIO_SSR_MODE"]') < src.index("import gradio")),         "GRADIO_SSR_MODE is set after the gradio import, which is too late"
    assert "ssr_mode=False" in src, "launch/mount must also pass ssr_mode"


def test_pages_read_timing_keys_the_harness_emits():
    """The harness names every timing after its stage: speak_ms, transcribe_ms.

    Both pages read tts_ms and asr_ms, which never exist -- so the budget line

    said "retrieval + guardrails + reader" over a number carrying 877 ms of

    speech synthesis. A missing key reads as absent, not as an error, which is

    why it survived. Any t.<x>_ms a page reads must be a key harness emits."""
    import re

    # Stage names are declared inline in _stages; read them from the source so
    # this test needs no built index to run.
    hsrc = (REPO / "src" / "harness.py").read_text(encoding="utf-8")
    stages = re.findall(r'Stage\("([a-z_]+)"', hsrc)
    assert len(stages) >= 7, f"expected 7 stages, found {stages}"
    allowed = {n + "_ms" for n in stages} | {"total_ms", "pipeline_ms", "budget_ms"}

    for rel in ("web/index.html", "src/serve.py"):
        text = (REPO / rel).read_text(encoding="utf-8")
        used = set(re.findall(r"t\.([a-z_]+_ms)", text))
        unknown = used - allowed
        assert not unknown, f"{rel} reads timing keys the harness never emits: {sorted(unknown)}"


def test_typed_questions_do_not_pay_for_synthesis():
    """MMS runs on CPU on the Space: 3365 ms for one Kannada sentence, against

    110 ms for retrieval + guardrails + reader. Typed questions asked for it

    unconditionally, so the page reported a 3.5 s pipeline for work nobody

    requested. Speech out belongs to the speech path; the voice route still

    sends no audio flag and the server default speaks."""
    page = (REPO / "web" / "index.html").read_text(encoding="utf-8")
    typed = page[page.index("function ask()"):page.index("/api/voice")]
    assert "audio:true" not in typed.replace(" ", ""),         "the typed path hardcodes audio again"
    assert 'audio:$("#speak").checked' in typed.replace(" ", ""),         "the typed path no longer honours the speak toggle"
    assert 'id="speak"' in page, "the toggle it reads does not exist"


def test_gate_3b_refuses_a_coherent_answer_about_the_wrong_thing():
    """Gate 3 asks whether the corpus has anything coherent to say. It cannot

    ask what the coherent thing is ABOUT, and a live Kannada query for the

    capital of India retrieved a peaked neighbourhood of crore, coordinates and

    tourism -- allowed by gate 3, confirmed verbatim by gate 4, and wrong.

    Grounded is not relevant."""
    from src.guardrails import GuardrailConfig, Guardrails
    from src.router import Passage

    g = Guardrails(GuardrailConfig(entity_gate=True))
    india = "ಭಾರತದ ರಾಜಧಾನಿ"
    crore = "ಅಥವಾ ಕೋಟಿ ಹತ್ತು ದಶಲಕ್ಷ"
    tour  = "ಉತ್ತಮ ತಾಣ ಪ್ರವಾಸ"
    bad = [Passage("c1", crore, .485), Passage("c2", tour, .478),
           Passage("c3", crore, .471)]
    cov, _ = g.rare_term_coverage(india, bad)
    assert cov == 0.0, f"expected no coverage, got {cov}"
    v = g.check_topical(bad, query=india)
    assert v.blocked and v.gate == "entity_mismatch", (v.gate, v.action)

    # And the query that legitimately worked must still get through, including
    # across inflection: the passage carries a suffixed form of the term.
    corp = "ಕಾರ್ಪೋರೇಷನ್ ಎಂದರೇನು"
    hit  = "ಕಾರ್ಪೋರೇಷನ್‌ಗಳು ಎಂದು"
    good = [Passage("k1", hit, .673), Passage("k2", crore, .61), Passage("k3", tour, .60)]
    v2 = g.check_topical(good, query=corp)
    assert not v2.blocked, f"a good answer was refused: {v2.gate}"

    # A text query with no query string must not be silently refused.
    assert not g.check_topical(good, query="").blocked, "empty query must skip 3b"

    # And the reason it ships off: with IDF over only the retrieved passages,
    # question words look rarer than entities and a good question is refused.
    hi = [Passage("h0", "कैंटालूप की बेलों को बढ़ने में १० दिन", .91),
          Passage("h1", "परागित फूलों से कैंटालूप विकसित", .62),
          Passage("h2", "तरबूज की खेती गर्म", .20)]
    good_q = "कैंटालूप को पकने में कितना समय लगता है"
    cov3, _ = g.rare_term_coverage(good_q, hi)
    assert cov3 == 0.0, (
        "the known false refusal disappeared -- if IDF is now corpus-level, "
        "re-measure the false-refusal rate and flip entity_gate on")
    assert GuardrailConfig().entity_gate is False,         "gate 3b must stay off until its IDF comes from the corpus"


if __name__ == "__main__":
    tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
    for t in tests:
        t()
        print(f"ok  {t.__name__}")
    print(f"\n{len(tests)} passed")