#!/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._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")