solo.istandClaude Fable 5.1
9faa403aee
The tool names the whole experiment, not just this script, so it gets a name with some personality. Repo, directory and command are `coopertron`; "Coopertron 5000" is the display name in the README and the report title. The activity the form captures is still a braindump. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
234 lines
12 KiB
Python
234 lines
12 KiB
Python
"""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("# Coopertron 5000 report", 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()
|