"""Offline tests for synthesize.py. Run: python3 -m unittest -v No network: the gateway is replaced by a fake that returns canned JSON shaped like a model reply, so these exercise parsing, validation, caching and rendering. """ import json import os import sys import tempfile import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import synthesize as S # noqa: E402 FIX = Path(__file__).parent / "fixtures" def _reply(text: str, finish: str = "stop") -> dict: return {"choices": [{"message": {"role": "assistant", "content": text}, "finish_reason": finish}], "usage": {"prompt_tokens": 100, "completion_tokens": 50}} class FakeGateway(S.Gateway): """Answers tagging calls with real quotes from the batch and the cluster call with a fixed grouping.""" def __init__(self, cache_dir=None, **kw): super().__init__("https://example.invalid/v1", "tinfoil/glm-5.3", cache_dir, **kw) self.posted = [] def _http_post(self, body): self.posted.append(body) user = body["messages"][1]["content"] if user.startswith("i0"): # cluster stage: item lines ids = [line.split()[0] for line in user.splitlines() if line.strip()] half = max(1, len(ids) // 2) obj = {"themes": [{"name": "Institutional access", "summary": "Group and org accounts.", "item_ids": ids[:half]}, {"name": "Own the stack", "summary": "Self-hosting versus buying tokens.", "item_ids": ids[half:]}], "novel": ids[:1], "dissent": [{"theme": "Own the stack", "item_ids": ids[-1:], "summary": "Cost doubts."}], "open_questions": [{"question": "Institutional tier?", "item_ids": ids[:2]}, "Bare string question"]} return _reply("Here you go:\n```json\n" + json.dumps(obj) + "\n```") # tag stage: emit one item per answered question using the first 6 words as the quote, # plus one fabricated quote that must be dropped. items = [] rid = None for line in user.splitlines(): if line.startswith("### "): head, _, _ = line[4:].partition(": ") rid, q = head.split(" / ") items.append({"id": rid, "q": q, "label": f"point about {q} from {rid}", "kind": "idea", "stance": "for", "quote": None}) elif items and items[-1]["quote"] is None and line.strip(): items[-1]["quote"] = " ".join(line.split()[:6]) items.append({"id": rid, "q": "Q1", "label": "fabricated", "kind": "concern", "stance": "against", "quote": "this text is not in the source"}) return _reply("```json\n" + json.dumps({"items": items}) + "\n```") class LoadTests(unittest.TestCase): def test_loads_fixture_and_detects_choice_column(self): loaded = S.load_responses(FIX / "sample.csv") self.assertEqual(len(loaded["responses"]), 12) self.assertEqual(loaded["choice_questions"], ["Priorities (pick up to 3)"]) self.assertEqual(len(loaded["questions"]), 7) r01 = loaded["responses"][0] self.assertEqual(r01["id"], "r01") self.assertIn("privacy", r01["choices"]["Priorities (pick up to 3)"]) # empty cells are skipped, not stored as "" self.assertNotIn("Anything else?", r01["answers"]) def test_meta_columns_dropped_and_limit(self): with tempfile.TemporaryDirectory() as d: p = Path(d) / "x.csv" p.write_text("Timestamp,Ideas,Notes\n2026-10-19,Build a thing,\n2026-10-20,Try another thing,\n2026-10-21,Third,\n") loaded = S.load_responses(p, limit=2) self.assertEqual(loaded["questions"], ["Ideas"]) # Timestamp dropped as meta, Notes dropped as empty self.assertEqual([r["id"] for r in loaded["responses"]], ["r01", "r02"]) def test_choice_heuristic_rejects_long_text(self): self.assertFalse(S.looks_like_choice_column(["a long free-text answer that goes on and on about many things"] * 5)) self.assertTrue(S.looks_like_choice_column(["privacy, green", "green", "privacy; docs", "docs"])) class JsonTests(unittest.TestCase): def test_fenced_with_prose(self): obj = S.extract_json('Sure! Here is the result:\n```json\n{"items": [1, 2]}\n```\nLet me know.') self.assertEqual(obj, {"items": [1, 2]}) def test_bare_object_and_trailing_comma(self): self.assertEqual(S.extract_json('{"a": 1,}'), {"a": 1}) self.assertEqual(S.extract_json('noise {"a": {"b": 2}} tail'), {"a": {"b": 2}}) def test_no_json_raises(self): with self.assertRaises(ValueError): S.extract_json("I cannot do that.") def test_message_text_prefers_tool_call_arguments(self): data = {"choices": [{"message": {"content": None, "tool_calls": [{"function": {"arguments": '{"items": []}'}}]}}]} self.assertEqual(S.message_text(data), '{"items": []}') class ValidationTests(unittest.TestCase): def test_quote_must_be_substring(self): loaded = S.load_responses(FIX / "sample.csv", limit=1) qidx = S.q_index(loaded["questions"]) batch = loaded["responses"] raw = [ {"id": "r01", "q": "Q1", "label": "privacy for client work", "kind": "want", "stance": "for", "quote": "use for client work without feeding a big vendor"}, {"id": "r01", "q": "Q1", "label": "made up", "kind": "want", "stance": "for", "quote": "never said this"}, {"id": "r99", "q": "Q1", "label": "unknown respondent", "kind": "want", "stance": "for", "quote": "Privacy."}, {"id": "r01", "q": "Q1", "label": "bad enums fall back", "kind": "banana", "stance": "maybe", "quote": "Privacy."}, ] kept, dropped = S.validate_items(raw, batch, qidx) self.assertEqual([k["label"] for k in kept], ["privacy for client work", "bad enums fall back"]) self.assertEqual(kept[1]["kind"], "idea") self.assertEqual(kept[1]["stance"], "neutral") self.assertEqual(len(dropped), 2) def test_cluster_validation_dedupes_and_tracks_unthemed(self): items = [{"iid": f"i00{n}"} for n in range(1, 5)] obj = {"themes": [{"name": "A", "summary": "", "item_ids": ["i001", "i002", "i002", "i999"]}, {"name": "B", "summary": "", "item_ids": ["i002", "i003"]}, {"name": "Empty", "summary": "", "item_ids": ["i999"]}], "novel": ["i004", "i999"], "dissent": [{"theme": "A", "item_ids": ["i999"], "summary": "x"}], "open_questions": []} c = S.validate_clusters(obj, items) self.assertEqual([t["item_ids"] for t in c["themes"]], [["i001", "i002"], ["i003"]]) self.assertEqual(c["unthemed"], ["i004"]) self.assertEqual(c["novel"], ["i004"]) self.assertEqual(c["dissent"], []) # all its ids were unknown class PipelineTests(unittest.TestCase): def test_end_to_end_offline_and_cache(self): os.environ[S.KEY_ENV] = "sk-test-not-real" loaded = S.load_responses(FIX / "sample.csv") with tempfile.TemporaryDirectory() as d: gw = FakeGateway(cache_dir=Path(d) / "cache") items, dropped = S.stage_tag(gw, loaded, batch_size=5, use_tools=False) self.assertEqual(gw.calls, 3) # 12 responses / 5 per batch self.assertEqual(len(dropped), 3) # one fabricated quote per batch self.assertTrue(all(it["quote"] in " ".join(loaded["responses"][int(it["rid"][1:]) - 1]["answers"].values()) for it in items)) clusters = S.stage_cluster(gw, items, use_tools=False) self.assertEqual(gw.calls, 4) self.assertEqual(len(clusters["themes"]), 2) self.assertEqual(clusters["open_questions"][1], {"question": "Bare string question", "item_ids": []}) # Same inputs again: everything served from cache, no network. gw2 = FakeGateway(cache_dir=Path(d) / "cache") S.stage_tag(gw2, loaded, batch_size=5, use_tools=False) S.stage_cluster(gw2, items, use_tools=False) self.assertEqual(gw2.calls, 0) self.assertEqual(gw2.cache_hits, 4) meta = {"date": "2026-10-19", "source": "sample.csv", "model": "tinfoil/glm-5.3", "dropped": dropped} md = S.render(loaded, items, clusters, meta) self.assertIn("# Braindump synthesis", md) self.assertIn("## Priorities (pick up to 3)", md) self.assertIn("**4** ยท privacy", md) # r01, r03, r06, r09 picked privacy self.assertIn("### 1. ", md) self.assertIn("Institutional access", md) self.assertIn("Own the stack", md) self.assertIn("2. Bare string question", md) self.assertIn("## Notable dissent", md) self.assertIn("## Appendix: all responses, verbatim", md) self.assertIn("3 extracted point(s) were discarded", md) # Every quote in the doc is verbatim from some response. all_text = " ".join(a for r in loaded["responses"] for a in r["answers"].values()) for line in md.splitlines(): if line.startswith('> "'): q = line[3:line.rindex('" (')] self.assertIn(q, S._norm_ws(all_text)) # Persist a rendering fixture for eyeballing. (FIX / "sample.review.md").write_text(md) (FIX / "sample.synthesis.json").write_text(json.dumps( {"items": items, "clusters": clusters, "meta": meta}, indent=1)) def test_retry_on_unparseable_reply(self): os.environ[S.KEY_ENV] = "sk-test-not-real" class Flaky(S.Gateway): n = 0 def _http_post(self, body): Flaky.n += 1 if Flaky.n == 1: return _reply("Sorry, here is prose only.") self_check = "could not be parsed" in body["messages"][1]["content"] return _reply('```json\n{"items": [], "retried": %s}\n```' % ("true" if self_check else "false")) gw = Flaky("https://example.invalid/v1", "tinfoil/glm-5.3", None) obj = S.call_json(gw, "sys", "user", ["items"], tag="t") self.assertEqual(obj, {"items": [], "retried": True}) self.assertEqual(gw.calls, 2) class ReasoningExhaustionTests(unittest.TestCase): def test_fails_fast_without_retry(self): os.environ[S.KEY_ENV] = "sk-test-not-real" class Exhausted(S.Gateway): def _http_post(self, body): return {"choices": [{"message": {"role": "assistant", "content": "", "reasoning_content": "..."}, "finish_reason": "length"}], "usage": {"prompt_tokens": 10, "completion_tokens": 8000, "completion_tokens_details": {"reasoning_tokens": 8000}}} gw = Exhausted("https://example.invalid/v1", "tinfoil/glm-5.3", None) with self.assertRaises(SystemExit) as cm: S.call_json(gw, "sys", "user", ["items"], tag="t") self.assertIn("hidden reasoning", str(cm.exception)) self.assertEqual(gw.calls, 1) class CliTests(unittest.TestCase): def test_refuses_non_tee_without_flag(self): self.assertEqual(S.main([str(FIX / "sample.csv"), "--model", "greenpt/glm-5.3", "--dry-run"]), 2) def test_dry_run_makes_no_calls(self): os.environ.pop(S.KEY_ENV, None) # would fail loudly if any call were attempted self.assertEqual(S.main([str(FIX / "sample.csv"), "--dry-run"]), 0) def test_estimate_is_sane(self): loaded = S.load_responses(FIX / "sample.csv") est = S.estimate(loaded, "tinfoil/glm-5.3", 15) self.assertEqual(est["batches"], 1) self.assertLess(est["est_cost_usd"], 0.25) if __name__ == "__main__": unittest.main()