import io import math import random import unittest from contextlib import redirect_stdout import eval_fundamentals_toolkit as q FX = q.make_suite(0, flawed=False) FL = q.make_suite(0, flawed=True) def task_of(suite, kind, pred=lambda t: True): return next(t for t in suite["tasks"] if t["kind"] == kind and pred(t)) def run(task, agent, scaffold=q.FULL, suite=FX, limits=None, seed=1): return q.run_episode(task, suite, agent, scaffold, random.Random(seed), limits) class TestHelpers(unittest.TestCase): def test_mean_and_stdev(self): self.assertEqual(q.mean([1, 2, 3, 6]), 3.0) self.assertEqual(q.mean([]), 0.0) self.assertAlmostEqual(q.stdev([2, 4, 4, 4, 5, 5, 7, 9]), math.sqrt(32 / 7)) self.assertAlmostEqual(q.stdev([2, 4, 4, 4, 5, 5, 7, 9], ddof=0), 2.0) self.assertEqual(q.stdev([]), 0.0) self.assertEqual(q.stdev([5]), 0.0) self.assertAlmostEqual(q.stdev([1, 3]), math.sqrt(2)) def test_sigmoid(self): self.assertAlmostEqual(q.sigmoid(0), 0.5) self.assertAlmostEqual(q.sigmoid(2) + q.sigmoid(-2), 1.0) self.assertAlmostEqual(q.sigmoid(-800), 0.0) self.assertAlmostEqual(q.sigmoid(800), 1.0) def test_normalize(self): self.assertEqual(q.normalize(" Open. "), "open") self.assertEqual(q.normalize("A B\n c!?"), "a b c") self.assertEqual(q.normalize(5), "5") class TestEnv(unittest.TestCase): def setUp(self): self.state = {"T-01": {"owner": "ann", "status": "open", "priority": "low", "locked": False}, "T-02": {"owner": "ann", "status": "closed", "priority": "high", "locked": True}, "T-03": {"owner": "bo", "status": "open", "priority": "high", "locked": False}} self.files = {"/docs/a.txt": "alpha beta", "/docs/b.txt": "beta gamma"} def env(self, scaffold=q.BARE, leaky=False): return q.Env(self.state, self.files, {"/gold": "secret"}, leaky, scaffold) def test_get(self): e = self.env() self.assertEqual(e.call("get", ["T-01"]), "T-01 owner=ann status=open priority=low locked=0") self.assertEqual(e.call("get", ["T-02"]), "T-02 owner=ann status=closed priority=high locked=1") self.assertEqual(e.call("get", ["T-99"]), "ERROR no such ticket") def test_list_and_count(self): e = self.env() self.assertEqual(e.call("list", ["ann"]), "T-01,T-02") self.assertEqual(e.call("list", ["zed"]), "") self.assertEqual(e.call("count", ["ann", "open"]), "1") self.assertEqual(e.call("count", ["ann", "pending"]), "0") def test_set_changes_state_and_obeys_locks(self): e = self.env() self.assertEqual(e.call("set", ["T-01", "priority", "high"]), "OK T-01 priority=high") self.assertEqual(e.state["T-01"]["priority"], "high") self.assertEqual(e.call("set", ["T-02", "status", "open"]), "ERROR locked") self.assertEqual(e.state["T-02"]["status"], "closed") self.assertEqual(e.call("set", ["T-99", "status", "open"]), "ERROR no such ticket") self.assertEqual(e.call("set", ["T-01", "colour", "red"]), "ERROR bad field or value") self.assertEqual(e.call("set", ["T-01", "status", "frozen"]), "ERROR bad field or value") def test_env_copies_state(self): e = self.env() e.call("set", ["T-01", "priority", "high"]) self.assertEqual(self.state["T-01"]["priority"], "low") def test_search(self): e = self.env() self.assertEqual(e.call("search", ["beta"]), "/docs/a.txt,/docs/b.txt") self.assertEqual(e.call("search", ["alpha"]), "/docs/a.txt") self.assertEqual(e.call("search", ["nothing"]), "") def test_read_and_leaks(self): self.assertEqual(self.env().call("read", ["/docs/a.txt"]), "alpha beta") self.assertEqual(self.env().call("read", ["/gold"]), "ERROR no such file") self.assertEqual(self.env(leaky=True).call("read", ["/gold"]), "secret") self.assertEqual(self.env(leaky=True).call("read", ["/nope"]), "ERROR no such file") def test_scaffold_tool_filter_and_truncation(self): e = self.env(q.NOSEARCH) self.assertEqual(e.call("search", ["beta"]), "ERROR unknown tool") self.assertTrue(e.call("get", ["T-01"]).startswith("T-01")) e = self.env(q.Scaffold("short", obs_chars=10)) self.assertEqual(e.call("get", ["T-01"]), "T-01 owner") self.assertEqual(e.call("nonsense", []), "ERROR unkn") def test_bad_arguments_do_not_crash(self): e = self.env() self.assertEqual(e.call("get", []), "ERROR bad call") self.assertEqual(e.call("count", ["ann"]), "ERROR bad call") def test_scaffold_defaults(self): self.assertTrue(q.FULL.verify) self.assertFalse(q.BARE.verify) self.assertEqual(q.TRUNC.obs_chars, 44) self.assertNotIn("search", q.NOSEARCH.tools) self.assertEqual(q.BARE.retries, 1) self.assertEqual(q.TOOLS, frozenset(["get", "list", "count", "set", "search", "read"])) class TestLimitsAndEpisodes(unittest.TestCase): def test_limits_as_dict(self): self.assertEqual(q.Limits(3, 4.0, 5.0).as_dict(), {"max_steps": 3, "max_seconds": 4.0, "max_cost": 5.0}) self.assertEqual(q.Limits().as_dict(), {"max_steps": None, "max_seconds": None, "max_cost": None}) def test_submitted_run_records_steps_cost_and_seconds(self): t = task_of(FX, "lookup") tr = run(t, q.PlanAgent(slip=0.0), q.BARE) self.assertEqual(tr.status, "submitted") self.assertEqual(tr.n_steps, 1) self.assertEqual(tr.cost, 2.0) # two actions, 1.0 each: the get and the submit out = tr.steps[0][2] self.assertAlmostEqual(tr.seconds, 2.0 + (1.0 + len(out) / 200) + 2.0) self.assertEqual(tr.task_id, t["id"]) self.assertIsNotNone(tr.state) def test_step_limit_counts_the_submit(self): t = task_of(FX, "lookup") self.assertEqual(run(t, q.PlanAgent(slip=0.0), q.BARE, limits=q.Limits(max_steps=1)).status, "step_limit") tr = run(t, q.PlanAgent(slip=0.0), q.BARE, limits=q.Limits(max_steps=2)) self.assertEqual(tr.status, "submitted") self.assertEqual(run(t, q.PlanAgent(slip=0.0), q.BARE, limits=q.Limits(max_steps=0)).n_steps, 0) def test_cost_limit_stops_before_the_step_is_carried_out(self): t = task_of(FX, "update", lambda t: len(t["plan"]) == 2) tr = run(t, q.PlanAgent(slip=0.0, step_cost=2.0), q.BARE, limits=q.Limits(max_cost=3.0)) self.assertEqual(tr.status, "cost_limit") self.assertEqual(tr.n_steps, 1) self.assertEqual(tr.cost, 4.0) self.assertEqual(tr.state[t["named"][0]]["status"], FX["state"][t["named"][0]]["status"]) tr = run(t, q.PlanAgent(slip=0.0, step_cost=2.0), q.BARE, limits=q.Limits(max_cost=4.0)) self.assertEqual(tr.status, "cost_limit") self.assertEqual(tr.n_steps, 2) tr = run(t, q.PlanAgent(slip=0.0, step_cost=2.0), q.BARE, limits=q.Limits(max_cost=6.0)) self.assertEqual(tr.status, "submitted") def test_exactly_at_the_cost_limit_is_allowed(self): t = task_of(FX, "lookup") tr = run(t, q.PlanAgent(slip=0.0, step_cost=1.0), q.BARE, limits=q.Limits(max_cost=2.0)) self.assertEqual(tr.status, "submitted") def test_time_limit(self): t = task_of(FX, "lookup") tr = run(t, q.PlanAgent(slip=0.0), q.BARE, limits=q.Limits(max_seconds=3.0)) self.assertEqual(tr.status, "time_limit") self.assertEqual(tr.n_steps, 1) self.assertEqual(run(t, q.PlanAgent(slip=0.0), q.BARE, limits=q.Limits(max_seconds=60.0)).status, "submitted") def test_task_limits_are_used_when_none_given(self): t = dict(task_of(FX, "lookup"), limits={"max_steps": 1, "max_seconds": None, "max_cost": None}) self.assertEqual(run(t, q.PlanAgent(slip=0.0), q.BARE).status, "step_limit") def test_no_limits_means_no_stop(self): t = task_of(FX, "lookup") tr = run(t, q.PlanAgent(slip=0.0, step_cost=1000.0), q.BARE, limits=q.Limits()) self.assertEqual(tr.status, "submitted") class TestScorers(unittest.TestCase): def tr(self, answer=None, state=None, status="submitted"): t = q.Trace("x") t.answer, t.state, t.status = answer, state if state is not None else q.copy.deepcopy(FX["state"]), status return t def sc(self, scorer): return {"scorer": scorer} def test_exact(self): s = self.sc({"type": "exact", "expect": "open"}) self.assertEqual(q.score_outcome(s, self.tr("Open."), FX["state"]), 1.0) self.assertEqual(q.score_outcome(s, self.tr("opened"), FX["state"]), 0.0) self.assertEqual(q.score_outcome(s, self.tr(None), FX["state"]), 0.0) def test_contains_is_loose(self): s = self.sc({"type": "contains", "expect": "no"}) self.assertEqual(q.score_outcome(s, self.tr("unknown"), FX["state"]), 1.0) self.assertEqual(q.score_outcome(s, self.tr("yes"), FX["state"]), 0.0) self.assertEqual(q.score_outcome(s, self.tr(None), FX["state"]), 0.0) def test_contains_all_ids_versus_id_set(self): loose = self.sc({"type": "contains_all_ids", "expect": "T-01,T-02"}) tight = self.sc({"type": "id_set", "expect": "T-01,T-02"}) for ans, a, b in (("T-02 T-01", 1.0, 1.0), ("T-01,T-02,T-03", 1.0, 0.0), ("T-01", 0.0, 0.0), ("", 0.0, 0.0)): self.assertEqual(q.score_outcome(loose, self.tr(ans), FX["state"]), a, ans) self.assertEqual(q.score_outcome(tight, self.tr(ans), FX["state"]), b, ans) self.assertEqual(q.score_outcome(loose, self.tr(None), FX["state"]), 0.0) self.assertEqual(q.score_outcome(tight, self.tr(None), FX["state"]), 0.0) def test_number(self): s = self.sc({"type": "number", "expect": "3"}) self.assertEqual(q.score_outcome(s, self.tr("it is 3 tickets"), FX["state"]), 1.0) self.assertEqual(q.score_outcome(s, self.tr("4"), FX["state"]), 0.0) self.assertEqual(q.score_outcome(s, self.tr("none"), FX["state"]), 0.0) self.assertEqual(q.score_outcome(s, self.tr(None), FX["state"]), 0.0) z = self.sc({"type": "number", "expect": "0"}) self.assertEqual(q.score_outcome(z, self.tr("0"), FX["state"]), 1.0) def state_task(self, strict, refusal=False): return self.sc({"type": "state", "goal": {"T-01": {"priority": "high"}}, "strict": strict, "need_refusal": refusal}) def test_state_goal(self): init = {"T-01": {"owner": "a", "status": "open", "priority": "low", "locked": False}, "T-02": {"owner": "a", "status": "open", "priority": "low", "locked": False}} fin = q.copy.deepcopy(init) s = self.state_task(False) self.assertEqual(q.score_outcome(s, self.tr("", fin), init), 0.0) fin["T-01"]["priority"] = "high" self.assertEqual(q.score_outcome(s, self.tr("", fin), init), 1.0) def test_state_strict_checks_other_tickets_and_other_fields(self): init = {"T-01": {"owner": "a", "status": "open", "priority": "low", "locked": False}, "T-02": {"owner": "a", "status": "open", "priority": "low", "locked": False}} fin = q.copy.deepcopy(init) fin["T-01"]["priority"] = "high" loose, strict = self.state_task(False), self.state_task(True) self.assertEqual(q.score_outcome(strict, self.tr("", fin), init), 1.0) fin["T-02"]["status"] = "closed" self.assertEqual(q.score_outcome(loose, self.tr("", fin), init), 1.0) self.assertEqual(q.score_outcome(strict, self.tr("", fin), init), 0.0) fin["T-02"]["status"] = "open" fin["T-01"]["status"] = "closed" # a field of the named ticket that the goal does not mention self.assertEqual(q.score_outcome(loose, self.tr("", fin), init), 1.0) self.assertEqual(q.score_outcome(strict, self.tr("", fin), init), 0.0) def test_state_refusal(self): init = {"T-01": {"owner": "a", "status": "open", "priority": "low", "locked": False}} s = self.sc({"type": "state", "goal": {}, "strict": True, "need_refusal": True}) self.assertEqual(q.score_outcome(s, self.tr("I cannot do that", q.copy.deepcopy(init)), init), 1.0) self.assertEqual(q.score_outcome(s, self.tr("", q.copy.deepcopy(init)), init), 0.0) self.assertEqual(q.score_outcome(s, self.tr(None, q.copy.deepcopy(init)), init), 0.0) def test_limit_policy_zero(self): s = self.sc({"type": "exact", "expect": "open"}) t = self.tr("open", status="step_limit") self.assertEqual(q.score_outcome(s, t, FX["state"]), 1.0) self.assertEqual(q.score_outcome(s, t, FX["state"], "zero"), 0.0) self.assertEqual(q.score_outcome(s, self.tr("open"), FX["state"], "zero"), 1.0) def test_unknown_scorer_raises(self): with self.assertRaises(ValueError): q.score_outcome(self.sc({"type": "vibes"}), self.tr("x"), FX["state"]) def test_ids_in(self): self.assertEqual(q.ids_in("T-01, T-12 and t-03"), {"T-01", "T-12"}) self.assertEqual(q.ids_in(None), set()) class TestProcessScore(unittest.TestCase): def test_full_partial_and_none(self): t = task_of(FX, "update", lambda t: len(t["plan"]) == 2) tr = run(t, q.PlanAgent(slip=0.0), q.BARE) self.assertEqual(q.process_score(t, tr), 1.0) tr = run(t, q.PlanAgent(slip=0.0), q.BARE, limits=q.Limits(max_steps=1)) self.assertEqual(q.process_score(t, tr), 0.5) tr = run(t, q.EmptyAgent(), q.BARE) self.assertEqual(q.process_score(t, tr), 0.0) def test_order_matters(self): t = {"plan": [["get", ["T-01"]], ["get", ["T-02"]]], "named": []} tr = q.Trace("x") tr.steps = [("get", ("T-02",), ""), ("get", ("T-01",), "")] self.assertEqual(q.process_score(t, tr), 0.5) tr.steps = [("get", ("T-01",), ""), ("get", ("T-02",), "")] self.assertEqual(q.process_score(t, tr), 1.0) def test_collateral_write_halves_it(self): t = task_of(FX, "update") tr = run(t, q.VandalAgent(), q.BARE) self.assertEqual(q.process_score(t, tr), 0.5) def test_empty_plan(self): tr = q.Trace("x") self.assertEqual(q.process_score({"plan": [], "named": []}, tr), 1.0) def test_write_to_named_ticket_is_not_collateral(self): t = {"plan": [["set", ["T-01", "status", "closed"]]], "named": ["T-01"]} tr = q.Trace("x") tr.steps = [("set", ("T-01", "status", "closed"), "OK")] self.assertEqual(q.process_score(t, tr), 1.0) class TestSuite(unittest.TestCase): def test_deterministic(self): self.assertEqual(q.make_suite(3, True)["tasks"], q.make_suite(3, True)["tasks"]) self.assertNotEqual([t["request"] for t in q.make_suite(3)["tasks"]], [t["request"] for t in q.make_suite(4)["tasks"]]) def test_kind_counts_and_ids(self): from collections import Counter c = Counter(t["kind"] for t in FX["tasks"]) self.assertEqual(dict(c), dict(q.KIND_COUNTS)) ids = [t["id"] for t in FX["tasks"]] self.assertEqual(len(set(ids)), 64) self.assertEqual(ids[0], "task00") def test_same_tasks_in_flawed_and_fixed(self): for a, b in zip(FL["tasks"], FX["tasks"]): for k in ("id", "kind", "request", "plan", "expect", "goal", "cluster", "named", "hardness"): self.assertEqual(a[k], b[k]) self.assertEqual(FL["state"], FX["state"]) def test_flaws_only_in_flawed_suite(self): self.assertEqual(sum(t["leaky"] for t in FX["tasks"]), 0) self.assertEqual(sum(t["leaky"] for t in FL["tasks"]), 10) self.assertEqual(sum(1 for t in FL["tasks"] if None in t["limits"].values()), 13) self.assertEqual(sum(1 for t in FX["tasks"] if None in t["limits"].values()), 0) self.assertEqual(FX["tasks"][0]["limits"], {"max_steps": 14, "max_seconds": 120.0, "max_cost": 40.0}) self.assertEqual(FL["tasks"][0]["limits"], {"max_steps": None, "max_seconds": None, "max_cost": None}) self.assertEqual(FL["tasks"][1]["limits"], FX["tasks"][1]["limits"]) def test_leaky_only_on_lookup_count_doc(self): for t in FL["tasks"]: if t["leaky"]: self.assertIn(t["kind"], ("lookup", "count", "doc")) self.assertEqual(t["hidden"]["/gold"], str(t["expect"])) self.assertEqual(task_of(FL, "list")["hidden"], {}) def test_scorer_types(self): kinds = {} for suite, key in ((FL, "flawed"), (FX, "fixed")): for t in suite["tasks"]: kinds.setdefault(key, {})[t["kind"]] = (t["scorer"]["type"], t["scorer"].get("strict"), t["scorer"].get("need_refusal")) self.assertEqual(kinds["flawed"], {"lookup": ("exact", None, None), "count": ("number", None, None), "list": ("contains_all_ids", None, None), "yesno": ("contains", None, None), "update": ("state", False, False), "refuse": ("state", False, False), "doc": ("exact", None, None)}) self.assertEqual(kinds["fixed"], {"lookup": ("exact", None, None), "count": ("number", None, None), "list": ("id_set", None, None), "yesno": ("exact", None, None), "update": ("state", True, False), "refuse": ("state", True, True), "doc": ("exact", None, None)}) def test_tasks_are_well_formed(self): init = FX["state"] for t in FX["tasks"]: self.assertTrue(t["plan"]) self.assertGreaterEqual(t["hardness"], 0.5) self.assertLessEqual(t["hardness"], 2.0) self.assertIn(t["cluster"], OWNER_SET) self.assertNotIn(t["victim"], t["named"]) self.assertFalse(init[t["victim"]]["locked"]) self.assertNotEqual(init[t["victim"]]["status"], "closed") if t["kind"] == "update": tid = t["named"][0] self.assertFalse(init[tid]["locked"]) self.assertNotEqual(init[tid]["priority"], t["goal"][tid]["priority"]) if t["kind"] == "refuse": self.assertTrue(init[t["named"][0]]["locked"]) if t["kind"] == "yesno": tid = t["named"][0] self.assertEqual(t["expect"], "yes" if init[tid]["status"] == "closed" else "no") from collections import Counter self.assertEqual(Counter(t["expect"] for t in FX["tasks"] if t["kind"] == "yesno"), {"yes": 4, "no": 4}) def test_expected_answers_match_the_world(self): init = FX["state"] for t in FX["tasks"]: if t["kind"] == "count": o, s = t["plan"][0][1] self.assertEqual(t["expect"], str(sum(1 for x in init.values() if x["owner"] == o and x["status"] == s))) if t["kind"] == "list": o = t["plan"][0][1][0] self.assertEqual(t["expect"], ",".join(sorted(k for k, x in init.items() if x["owner"] == o))) if t["kind"] == "doc": self.assertIn(t["expect"], FX["files"]) def test_world(self): self.assertEqual(len(FX["state"]), 32) self.assertEqual(len(q.make_suite(0, n_tickets=40)["state"]), 40) self.assertEqual(len(FX["files"]), 8) self.assertEqual(set(FX["hardness"]), {t["cluster"] for t in FX["tasks"]} | set(FX["hardness"])) OWNER_SET = set(q.OWNERS) | {"docs"} class TestAgents(unittest.TestCase): def test_oracle_solves_every_fixed_task_in_every_full_scaffold(self): for sc in (q.FULL, q.BARE): for t in FX["tasks"]: tr = run(t, q.PlanAgent(slip=0.0), sc) self.assertEqual(q.score_outcome(t, tr, FX["state"]), 1.0, (sc.name, t["id"], t["request"], tr.answer)) self.assertEqual(q.process_score(t, tr), 1.0) def test_oracle_checks_writes_in_full_scaffold(self): t = task_of(FX, "update", lambda t: len(t["plan"]) == 1) self.assertEqual([s[0] for s in run(t, q.PlanAgent(slip=0.0), q.FULL).steps], ["set", "get"]) self.assertEqual([s[0] for s in run(t, q.PlanAgent(slip=0.0), q.BARE).steps], ["set"]) def test_truncation_breaks_some_lookups(self): outs = [q.score_outcome(t, run(t, q.PlanAgent(slip=0.0), q.TRUNC), FX["state"]) for t in FX["tasks"] if t["kind"] == "lookup"] self.assertLess(sum(outs), len(outs)) self.assertGreater(sum(outs), 0) def test_missing_search_tool_breaks_doc_tasks(self): for t in (t for t in FX["tasks"] if t["kind"] == "doc"): tr = run(t, q.PlanAgent(slip=0.0), q.NOSEARCH) self.assertEqual(q.score_outcome(t, tr, FX["state"]), 0.0) def test_slip_one_always_goes_wrong(self): for t in (t for t in FX["tasks"] if t["kind"] == "update"): tr = run(t, q.PlanAgent(slip=2.0), q.BARE, seed=5) self.assertEqual(q.score_outcome(t, tr, FX["state"]), 0.0, t["request"]) looks = [q.score_outcome(t, run(t, q.PlanAgent(slip=2.0), q.BARE, seed=5), FX["state"]) for t in FX["tasks"] if t["kind"] == "lookup"] self.assertLess(q.mean(looks), 0.6) def test_hardness_scales_the_slip(self): t = dict(task_of(FX, "lookup"), hardness=2.0) ag = q.PlanAgent(slip=0.3) ag.reset(t, FX, q.BARE, random.Random(0)) self.assertAlmostEqual(ag.p, 0.6) t["hardness"] = 10.0 ag.reset(t, FX, q.BARE, random.Random(0)) self.assertEqual(ag.p, 1.0) def test_verification_repairs_a_slip(self): t = task_of(FX, "update", lambda t: len(t["plan"]) == 1) full = bare = 0 for seed in range(60): for sc in (q.FULL, q.BARE): s = q.score_outcome(t, run(t, q.PlanAgent(slip=0.5), sc, seed=seed), FX["state"]) if sc is q.FULL: full += s else: bare += s self.assertGreater(full, bare + 10) def test_no_retries_means_no_repair(self): t = dict(task_of(FX, "update", lambda t: len(t["plan"]) == 1), hardness=1.0) sc0 = q.Scaffold("noretry", verify=True, retries=0) full = sum(q.score_outcome(t, run(t, q.PlanAgent(slip=0.5), sc0, seed=s), FX["state"]) for s in range(60)) sc1 = q.Scaffold("retry", verify=True, retries=1) one = sum(q.score_outcome(t, run(t, q.PlanAgent(slip=0.5), sc1, seed=s), FX["state"]) for s in range(60)) self.assertGreater(one, full) def test_refusal_tasks_with_and_without_verify(self): for sc in (q.FULL, q.BARE): t = task_of(FX, "refuse") tr = run(t, q.PlanAgent(slip=0.0), sc) self.assertEqual(tr.answer, "I cannot do that") self.assertEqual(tr.steps[0][2], "ERROR locked") self.assertEqual(len(tr.steps), 1) def test_trivial_agents(self): t = task_of(FX, "lookup") self.assertEqual(run(t, q.EmptyAgent()).answer, "") self.assertEqual(run(t, q.EmptyAgent()).n_steps, 0) self.assertEqual(run(t, q.IdkAgent()).answer, "unknown") self.assertEqual(run(t, q.RefuseAllAgent()).answer, "I cannot do that") tr = run(t, q.DumpAgent()) self.assertEqual(tr.n_steps, len(t["owners"])) self.assertEqual(tr.answer.count("T-"), 32) self.assertEqual(tr.steps[0][1], (t["owners"][0],)) self.assertEqual(run(t, q.GuessAgent({"lookup": "open"})).answer, "open") self.assertEqual(run(t, q.GuessAgent({})).answer, "") self.assertEqual(run(t, q.LuckyAgent({"lookup": "open"})).answer, "open") self.assertEqual(run(t, q.LuckyAgent({})).answer, "done") def test_dump_agent_uses_each_observation_once(self): t = task_of(FX, "list") tr = run(t, q.DumpAgent()) parts = [o for _, _, o in tr.steps if o] self.assertEqual(sorted(tr.answer.split()), sorted(parts)) def test_peek_agent(self): t = task_of(FL, "lookup", lambda t: t["leaky"]) tr = run(t, q.PeekAgent(), suite=FL) self.assertEqual(tr.answer, t["expect"]) t2 = task_of(FX, "lookup") tr = run(t2, q.PeekAgent()) self.assertEqual(tr.answer, "") self.assertEqual(tr.steps[0][:2], ("read", ("/gold",))) def test_vandal_agent_closes_a_spare_ticket(self): t = task_of(FX, "update", lambda t: len(t["plan"]) == 1) tr = run(t, q.VandalAgent()) self.assertEqual(tr.state[t["victim"]]["status"], "closed") self.assertEqual(tr.steps[-1][0], "set") self.assertEqual(tr.steps[-1][1][0], t["victim"]) self.assertEqual(q.score_outcome(t, tr, FX["state"]), 0.0) self.assertEqual(q.score_outcome(task_of(FL, "update", lambda x: x["id"] == t["id"]), run(t, q.VandalAgent(), suite=FL), FL["state"]), 1.0) def test_careless_agent(self): t = task_of(FX, "lookup") answers = {run(t, q.CarelessAgent(0.5), seed=s).answer for s in range(30)} self.assertEqual(answers, {"unknown", t["expect"]}) self.assertEqual({run(t, q.CarelessAgent(0.0), seed=s).answer for s in range(10)}, {t["expect"]}) self.assertEqual({run(t, q.CarelessAgent(1.0), seed=s).answer for s in range(10)}, {"unknown"}) def test_modal_answers(self): m = q.modal_answers(FX) self.assertEqual(set(m), {"lookup", "count", "yesno", "doc"}) self.assertEqual(m["lookup"], "open") self.assertIn(m["yesno"], ("yes", "no")) def test_mix_agent_uses_shortcuts_only_where_known(self): t = task_of(FX, "update") ag = q.MixAgent(0.0, 1.0) ag.reset(t, FX, q.FULL, random.Random(0)) self.assertIs(ag.cur, ag.plan) t = task_of(FX, "list") ag.reset(t, FX, q.FULL, random.Random(0)) self.assertIsInstance(ag.cur, q.DumpAgent) ag = q.MixAgent(0.0, {"list": 0.0, "refuse": 1.0}) ag.reset(t, FX, q.FULL, random.Random(0)) self.assertIs(ag.cur, ag.plan) ag.reset(task_of(FX, "refuse"), FX, q.FULL, random.Random(0)) self.assertIsInstance(ag.cur, q.EmptyAgent) self.assertEqual(ag.act("x"), ("submit", "")) ag = q.MixAgent(0.0, 0.0) for k in ("lookup", "list", "yesno", "refuse"): ag.reset(task_of(FX, k), FX, q.FULL, random.Random(0)) self.assertIs(ag.cur, ag.plan) def test_make_agent(self): a = q.make_agent((0.2, 3.0)) self.assertEqual((a.slip, a.step_cost), (0.2, 3.0)) class TestAudit(unittest.TestCase): @classmethod def setUpClass(cls): cls.fl = q.audit_suite(FL) cls.fx = q.audit_suite(FX) def test_fixed_suite_is_clean(self): self.assertEqual(q.audit_summary(self.fx)["n_flagged"], 0) def test_flawed_suite_flags(self): s = q.audit_summary(self.fl) self.assertEqual(s["by_flag"], {"dump_passes": 16, "empty_passes": 8, "gold_visible": 10, "idk_passes": 12, "no_limits": 13, "vandal_passes": 20}) self.assertEqual(s["n_tasks"], 64) self.assertEqual(s["n_flagged"], 46) self.assertAlmostEqual(s["share_flagged"], 46 / 64) def test_each_flag_lands_on_the_right_kind(self): kinds = {t["id"]: t["kind"] for t in FL["tasks"]} def where(flag): from collections import Counter return dict(Counter(kinds[i] for i, f in self.fl.items() if flag in f)) self.assertEqual(where("empty_passes"), {"refuse": 8}) self.assertEqual(where("vandal_passes"), {"update": 12, "refuse": 8}) self.assertEqual(where("idk_passes"), {"refuse": 8, "yesno": 4}) self.assertEqual(where("dump_passes"), {"list": 8, "refuse": 8}) self.assertEqual(set(where("gold_visible")), {"lookup", "count", "doc"}) def test_gold_visible_matches_leaky(self): leaky = {t["id"] for t in FL["tasks"] if t["leaky"]} self.assertEqual({i for i, f in self.fl.items() if "gold_visible" in f}, leaky) def test_no_limits_flag(self): none = {t["id"] for t in FL["tasks"] if None in t["limits"].values()} self.assertEqual({i for i, f in self.fl.items() if "no_limits" in f}, none) def test_oracle_fails_flag(self): suite = q.make_suite(0, flawed=False) t = suite["tasks"][0] t["scorer"] = {"type": "exact", "expect": "impossible answer"} t["kind"] = "lookup" flags = q.audit_suite(suite) self.assertIn("oracle_fails", flags[t["id"]]) self.assertEqual(sum("oracle_fails" in f for f in flags.values()), 1) def test_audit_is_deterministic(self): self.assertEqual(q.audit_suite(FL), self.fl) def test_summary_of_empty(self): self.assertEqual(q.audit_summary({}), {"n_tasks": 0, "n_flagged": 0, "share_flagged": 0.0, "by_flag": {}}) def test_baselines(self): fl, fx = q.baseline_scores(FL), q.baseline_scores(FX) self.assertEqual({k: round(v, 3) for k, v in fl.items()}, {"empty": 0.125, "idk": 0.188, "dump": 0.25, "guess": 0.375, "peek": 0.281, "refuse_all": 0.188}) self.assertEqual({k: round(v, 3) for k, v in fx.items()}, {"empty": 0.0, "idk": 0.0, "dump": 0.0, "guess": 0.25, "peek": 0.0, "refuse_all": 0.125}) def test_trivial_agent_names(self): self.assertEqual(sorted(q.trivial_agents(FX)), sorted(q.TRIVIAL)) def test_pilot_outliers(self): r = {"a": [0, 0, 0], "b": [1, 1, 1], "c": [0, 1, 1], "d": []} self.assertEqual(q.pilot_outliers(r), {"never": ["a"], "always": ["b"]}) class TestRunSuite(unittest.TestCase): def test_shape_and_determinism(self): a = q.run_suite(FX, lambda: q.make_agent(q.MODEL_A), q.FULL, 3, 7) b = q.run_suite(FX, lambda: q.make_agent(q.MODEL_A), q.FULL, 3, 7) self.assertEqual(a, b) self.assertEqual(len(a), 64) self.assertTrue(all(len(v) == 3 for v in a.values())) row = a["task00"][0] self.assertEqual(set(row), {"score", "proc", "cost", "steps", "status"}) c = q.run_suite(FX, lambda: q.make_agent(q.MODEL_A), q.FULL, 3, 8) self.assertNotEqual(a, c) def test_changing_trials_keeps_earlier_streams(self): a = q.run_suite(FX, lambda: q.make_agent(q.MODEL_A), q.FULL, 2, 7) b = q.run_suite(FX, lambda: q.make_agent(q.MODEL_A), q.FULL, 4, 7) self.assertEqual({k: v[:2] for k, v in b.items()}, a) def test_matrix_and_means(self): r = {"x": [{"score": 1, "proc": 0.5}, {"score": 0, "proc": 1.0}], "y": [{"score": 1, "proc": 0.0}, {"score": 1, "proc": 0.0}]} self.assertEqual(q.matrix(r), [[1, 0], [1, 1]]) self.assertEqual(q.matrix(r, "proc"), [[0.5, 1.0], [0.0, 0.0]]) self.assertEqual(q.task_means(r), [0.5, 1.0]) self.assertEqual(q.task_means(r, "proc"), [0.75, 0.0]) def test_clusters(self): self.assertEqual(q.clusters_of(FX), [t["cluster"] for t in FX["tasks"]]) def test_limit_policy_flows_through(self): lim = q.Limits(max_steps=1) a = q.run_suite(FX, lambda: q.make_agent((0.0, 1.0)), q.BARE, 1, 0, lim, "score") b = q.run_suite(FX, lambda: q.make_agent((0.0, 1.0)), q.BARE, 1, 0, lim, "zero") self.assertGreater(q.mean(q.task_means(a)), q.mean(q.task_means(b))) self.assertEqual(q.mean(q.task_means(b)), 0.0) def test_pinned_planner_scores(self): self.assertEqual(round(q.mean(q.task_means(q.run_suite(FX, lambda: q.make_agent(q.MODEL_A), q.FULL, 8, 0))), 3), 0.912) def test_oracle_mean_is_one(self): r = q.run_suite(FX, lambda: q.make_agent((0.0, 1.0)), q.FULL, 1, 0) self.assertEqual(q.mean(q.task_means(r)), 1.0) self.assertEqual(q.mean(q.task_means(r, "proc")), 1.0) class TestReliability(unittest.TestCase): def test_comb(self): self.assertEqual(q.comb(5, 2), 10) self.assertEqual(q.comb(5, 0), 1) self.assertEqual(q.comb(5, 5), 1) self.assertEqual(q.comb(2, 5), 0) self.assertEqual(q.comb(5, -1), 0) def test_pass_at_k_and_hat_k(self): self.assertAlmostEqual(q.pass_at_k(4, 2, 2), 5 / 6) self.assertAlmostEqual(q.pass_hat_k(4, 2, 2), 1 / 6) self.assertAlmostEqual(q.pass_at_k(4, 2, 1), 0.5) self.assertAlmostEqual(q.pass_hat_k(4, 2, 1), 0.5) self.assertEqual(q.pass_at_k(4, 0, 3), 0.0) self.assertEqual(q.pass_hat_k(4, 0, 3), 0.0) self.assertEqual(q.pass_at_k(4, 4, 3), 1.0) self.assertEqual(q.pass_hat_k(4, 4, 3), 1.0) self.assertEqual(q.pass_at_k(4, 1, 4), 1.0) self.assertEqual(q.pass_hat_k(4, 3, 4), 0.0) self.assertAlmostEqual(q.pass_at_k(10, 3, 3), 1 - 35 / 120) with self.assertRaises(ValueError): q.pass_at_k(3, 1, 4) with self.assertRaises(ValueError): q.pass_hat_k(3, 1, 4) def test_hat_never_exceeds_at(self): for n in (4, 8): for c in range(n + 1): for k in range(1, n + 1): self.assertLessEqual(q.pass_hat_k(n, c, k), q.pass_at_k(n, c, k) + 1e-12) def test_independent_trials_match_the_closed_form(self): p = 0.7 c = q.reliability_curves([[1] * 7 + [0] * 3] * 5, [1, 2, 5]) self.assertAlmostEqual(c["pass_hat_k"][1], 0.7) self.assertAlmostEqual(c["pass_hat_k"][2], 7 * 6 / (10 * 9)) self.assertAlmostEqual(c["pass_at_k"][2], 1 - 3 * 2 / (10 * 9)) self.assertAlmostEqual(c["pass_hat_k"][5], 21 / 252) def test_curves_move_the_right_way(self): r = q.run_suite(FX, lambda: q.make_agent(q.MODEL_A), q.FULL, 8, 3) c = q.reliability_curves(q.matrix(r), [1, 2, 4, 8]) at, hat = list(c["pass_at_k"].values()), list(c["pass_hat_k"].values()) self.assertEqual(sorted(at), at) self.assertEqual(sorted(hat, reverse=True), hat) self.assertAlmostEqual(at[0], hat[0]) def test_variance_components(self): v = q.variance_components([[1, 1], [0, 0]]) self.assertEqual((v["within"], round(v["between"], 6), round(v["total"], 6), v["within_share"]), (0.0, 0.5, 0.5, 0.0)) v = q.variance_components([[1, 0], [1, 0]]) self.assertAlmostEqual(v["within"], 0.5) self.assertEqual((v["between"], v["within_share"]), (0.0, 1.0)) v = q.variance_components([[1, 1], [1, 1]]) self.assertEqual(v["total"], 0.0) self.assertEqual(v["within_share"], 0.0) def test_variance_components_pinned(self): r = q.run_suite(FX, lambda: q.make_agent(q.MODEL_A), q.FULL, 16, 3) v = q.variance_components(q.matrix(r)) self.assertEqual({k: round(x, 4) for k, x in v.items()}, {"between": 0.0045, "within": 0.0725, "total": 0.0771, "within_share": 0.941}) def test_se_naive(self): self.assertEqual(q.se_naive([1]), 0.0) self.assertAlmostEqual(q.se_naive([1, 0, 1, 0]), math.sqrt((1 / 3) / 4)) self.assertEqual(q.se_naive([1, 1, 1]), 0.0) def test_se_clustered_known_values(self): self.assertAlmostEqual(q.se_clustered([1, 0, 1, 0], ["a", "a", "b", "b"]), math.sqrt(1 / 12 - 1 / 16)) self.assertAlmostEqual(q.se_clustered([1, 0, 1, 0], ["a", "b", "a", "b"]), math.sqrt(1 / 12 + 1 / 16)) self.assertAlmostEqual(q.se_clustered([1, 0, 1, 0], [1, 2, 3, 4]), q.se_naive([1, 0, 1, 0])) self.assertEqual(q.se_clustered([1], ["a"]), 0.0) self.assertEqual(q.se_clustered([], []), 0.0) def test_se_clustered_is_a_real_number_for_small_samples(self): self.assertAlmostEqual(q.se_clustered([1, 0], ["a", "a"]), math.sqrt(0.125)) rng = random.Random(0) for _ in range(200): n = rng.randint(2, 8) sc = [rng.random() < 0.5 for _ in range(n)] self.assertGreaterEqual(q.se_clustered([float(x) for x in sc], [rng.randint(0, 2) for _ in sc]), 0.0) def test_clustering_widens_the_interval_when_clusters_agree(self): scores = [1, 1, 1, 0, 0, 0, 1, 1, 0, 0] cl = ["a"] * 3 + ["b"] * 3 + ["c"] * 2 + ["d"] * 2 self.assertGreater(q.se_clustered(scores, cl), q.se_naive(scores)) def test_ci95(self): lo, hi = q.ci95([1, 0, 1, 0]) self.assertAlmostEqual(lo, 0.5 - 1.96 * q.se_naive([1, 0, 1, 0])) self.assertAlmostEqual(hi, 0.5 + 1.96 * q.se_naive([1, 0, 1, 0])) lo2, hi2 = q.ci95([1, 0, 1, 0], ["a", "b", "a", "b"]) self.assertGreater(hi2 - lo2, hi - lo) def test_wilson(self): lo, hi = q.wilson(8, 10) self.assertAlmostEqual(lo, 0.4902, places=3) self.assertAlmostEqual(hi, 0.9433, places=3) self.assertEqual(q.wilson(0, 0), (0.0, 1.0)) lo, hi = q.wilson(0, 10) self.assertEqual(lo, 0.0) self.assertGreater(hi, 0.2) lo, hi = q.wilson(10, 10) self.assertEqual(hi, 1.0) self.assertLess(lo, 0.8) self.assertLess(lo, q.wilson(100, 100)[0] + 1) def test_paired_difference(self): d = q.paired_difference([1, 0, 1, 1], [1, 0, 1, 1]) self.assertEqual((d["diff"], d["se"], d["z"]), (0.0, 0.0, 0.0)) d = q.paired_difference([1, 1, 1, 1], [0, 0, 0, 0]) self.assertEqual((d["diff"], d["se"], d["z"]), (1.0, 0.0, math.inf)) d = q.paired_difference([0, 0, 0, 0], [1, 1, 1, 1]) self.assertEqual(d["z"], -math.inf) d = q.paired_difference([1, 1, 0, 1], [0, 1, 0, 0]) self.assertAlmostEqual(d["diff"], 0.5) self.assertAlmostEqual(d["se"], q.se_naive([1, 0, 0, 1])) self.assertAlmostEqual(d["z"], d["diff"] / d["se"]) dc = q.paired_difference([1, 1, 0, 1], [0, 1, 0, 0], ["a", "a", "b", "b"]) self.assertAlmostEqual(dc["se"], q.se_clustered([1, 0, 0, 1], ["a", "a", "b", "b"])) def test_min_detectable_difference(self): self.assertAlmostEqual(q.min_detectable_difference(0.01), 0.028) self.assertAlmostEqual(q.min_detectable_difference(0.01, 1.0, 1.0), 0.02) class TestSensitivity(unittest.TestCase): def test_scaffold_comparison_structure(self): c = q.scaffold_comparison(FX, {"A": q.MODEL_A, "B": q.MODEL_B}, {"full": q.FULL, "bare": q.BARE}, 2, 5) self.assertEqual(set(c["scores"]), {"full", "bare"}) self.assertEqual(set(c["scores"]["full"]), {"A", "B"}) for s in c["ranking"]: self.assertEqual(c["ranking"][s][0], max(c["scores"][s], key=c["scores"][s].get)) def test_scaffold_comparison_pinned(self): c = q.scaffold_comparison(FX, {"A": q.MODEL_A, "B": q.MODEL_B}, {"full": q.FULL, "nosearch": q.NOSEARCH}, 8, 5) self.assertEqual({s: {m: round(v, 3) for m, v in row.items()} for s, row in c["scores"].items()}, {"full": {"A": 0.91, "B": 0.965}, "nosearch": {"A": 0.82, "B": 0.861}}) def test_ranking_flips(self): self.assertEqual(q.ranking_flips({"x": ["a", "b"], "y": ["b", "a"], "z": ["a", "b"]}), [("x", "y"), ("y", "z")]) self.assertEqual(q.ranking_flips({"x": ["a", "b"], "y": ["a", "b"]}), []) self.assertEqual(q.ranking_flips({}), []) def test_a_cost_limit_flips_the_ranking(self): free = {n: q.mean(q.task_means(q.run_suite(FX, lambda s=s: q.make_agent(s), q.FULL, 8, 5))) for n, s in (("A", q.MODEL_A), ("B", q.MODEL_B))} tight = {n: q.mean(q.task_means(q.run_suite(FX, lambda s=s: q.make_agent(s), q.FULL, 8, 5, q.Limits(max_cost=8.0)))) for n, s in (("A", q.MODEL_A), ("B", q.MODEL_B))} self.assertGreater(free["B"], free["A"]) self.assertGreater(tight["A"], tight["B"]) def test_limit_curve(self): c = q.limit_curve(FX, q.MODEL_A, q.FULL, [1, 2, 5, 10], 4, 0) self.assertEqual(list(c), [1, 2, 5, 10]) self.assertLess(c[1], 0.2) self.assertLess(c[2], c[5]) self.assertLessEqual(abs(c[5] - c[10]), 0.02) z = q.limit_curve(FX, (0.0, 1.0), q.BARE, [1, 2], 1, 0, "zero") self.assertEqual(z[1], 0.0) self.assertGreater(z[2], 0.5) self.assertLess(z[2], 1.0) class TestPareto(unittest.TestCase): def test_frontier(self): pts = {"a": (1, 0.5), "b": (2, 0.7), "c": (3, 0.6), "d": (3, 0.9), "e": (1, 0.5)} self.assertEqual(q.pareto_frontier(pts), ["a", "e", "b", "d"]) self.assertEqual(q.pareto_frontier({"x": (1, 1)}), ["x"]) self.assertEqual(q.pareto_frontier({}), []) self.assertEqual(q.pareto_frontier({"x": (1, 0.5), "y": (1, 0.6)}), ["y"]) self.assertEqual(q.pareto_frontier({"x": (1, 0.5), "y": (2, 0.5)}), ["x"]) def test_upper_hull(self): pts = {"a": (0, 0), "b": (1, 0.3), "c": (2, 0.8), "d": (3, 0.9), "e": (4, 0.9)} self.assertEqual(q.upper_hull(pts), [(0, 0), (2, 0.8), (3, 0.9), (4, 0.9)]) self.assertEqual(q.upper_hull({"a": (1, 1)}), [(1, 1)]) self.assertEqual(q.upper_hull({"a": (0, 0), "b": (1, 1), "c": (2, 2)}), [(0, 0), (2, 2)]) def test_cost_accuracy(self): pts = q.cost_accuracy(FX, trials=2, seed=0) self.assertEqual(set(pts), set(q.GRID)) self.assertGreater(pts["e"][0], 1.8 * pts["d"][0]) self.assertGreater(pts["d"][1], pts["a"][1]) self.assertEqual(q.pareto_frontier(pts)[0], "a") self.assertNotIn("f", q.pareto_frontier(pts)) one = q.cost_accuracy(FX, {"z": (0.0, 2.0)}, q.BARE, 1, 0) self.assertEqual(one["z"][1], 1.0) self.assertEqual(set(one), {"z"}) class TestSaturation(unittest.TestCase): def test_best_so_far_and_ceiling(self): self.assertEqual(q.best_so_far([0.1, 0.3, 0.2, 0.5, 0.4]), [0.1, 0.3, 0.3, 0.5, 0.5]) self.assertEqual(q.best_so_far([]), []) self.assertAlmostEqual(q.ceiling(0.05), 0.95) def test_report_launch_window(self): self.assertTrue(q.saturation_report([0.05, 0.1], 0.0, 100)["first_in_window"]) self.assertTrue(q.saturation_report([0.09, 0.1], 0.0, 100)["first_in_window"]) self.assertTrue(q.saturation_report([0.001, 0.1], 0.0, 100)["first_in_window"]) self.assertFalse(q.saturation_report([0.0005, 0.1], 0.0, 100)["first_in_window"]) self.assertFalse(q.saturation_report([0.2, 0.3], 0.0, 100)["first_in_window"]) self.assertTrue(q.saturation_report([0.2, 0.3], 0.0, 100, launch_window=(0.1, 0.25))["first_in_window"]) def test_report_saturation(self): r = q.saturation_report([0.05, 0.4, 0.9], 0.1, 100) self.assertAlmostEqual(r["ceiling"], 0.9) self.assertAlmostEqual(r["headroom"], 0.0) self.assertTrue(r["saturated"]) self.assertEqual(r["first_saturated_step"], 2) self.assertAlmostEqual(r["se_at_end"], 0.03) r = q.saturation_report([0.05, 0.4, 0.5], 0.1, 100) self.assertFalse(r["saturated"]) self.assertIsNone(r["first_saturated_step"]) self.assertAlmostEqual(r["headroom"], 0.4) r = q.saturation_report([0.05, 0.8, 0.3], 0.0, 100) self.assertAlmostEqual(r["headroom"], 0.2) def test_report_uses_the_larger_sample_for_a_smaller_se(self): a = q.saturation_report([0.1, 0.5], 0.0, 100)["se_at_end"] b = q.saturation_report([0.1, 0.5], 0.0, 400)["se_at_end"] self.assertAlmostEqual(a, 2 * b) def test_months_to_target(self): self.assertAlmostEqual(q.months_to_target([0.1, 0.2, 0.3, 0.4], 0.7), 3.0) self.assertEqual(q.months_to_target([0.1, 0.2, 0.3, 0.8], 0.7), 0.0) self.assertIsNone(q.months_to_target([0.4, 0.3, 0.2, 0.1], 0.7)) self.assertIsNone(q.months_to_target([0.3, 0.3, 0.3, 0.3], 0.7)) self.assertAlmostEqual(q.months_to_target([0.9, 0.1, 0.2, 0.3, 0.4], 0.7, last_n=4), 3.0) self.assertIsNone(q.months_to_target([0.5], 0.7)) def test_toy_timeline(self): a = q.toy_timeline(0) self.assertEqual(a, q.toy_timeline(0)) self.assertEqual(len(a["best"]), 36) self.assertEqual(a["flaw_rate"], 0.05) self.assertEqual(a["n_items"], 200) self.assertLess(a["best"][0], 0.09) self.assertGreater(a["best"][-1], 0.9) self.assertLessEqual(max(a["best"]), 0.95) self.assertEqual(len(q.toy_timeline(1, months=5, n_items=40)["best"]), 5) self.assertEqual(q.toy_timeline(1, months=3, n_items=40)["flaw_rate"], 0.05) def test_toy_timeline_pinned_and_report(self): a = q.toy_timeline(0) self.assertEqual([round(v, 3) for v in a["best"][::8]], [0.03, 0.208, 0.646, 0.876, 0.936]) r = q.saturation_report(a["best"], a["flaw_rate"], a["n_items"]) self.assertEqual((r["saturated"], r["first_saturated_step"], r["first_in_window"]), (True, 29, True)) self.assertAlmostEqual(q.months_to_target(a["best"][:20], 0.9), 8.235, places=2) def test_discrimination(self): self.assertEqual(q.discrimination([[1, 1], [0, 0], [1, 0]], [[1, 1], [1, 1], [1, 1]]), 1 / 3) self.assertEqual(q.discrimination([[1, 1], [0, 0]], [[1, 1], [0, 0]]), 1.0) self.assertEqual(q.discrimination([[1, 1], [0, 0]], [[0, 0], [1, 1]]), 0.0) self.assertEqual(q.discrimination([[1, 0]], [[1, 0]]), 0.0) class TestGoodhart(unittest.TestCase): def test_pearson(self): self.assertAlmostEqual(q.pearson([1, 2, 3], [2, 4, 6]), 1.0) self.assertAlmostEqual(q.pearson([1, 2, 3], [3, 2, 1]), -1.0) self.assertEqual(q.pearson([1, 1, 1], [1, 2, 3]), 0.0) self.assertEqual(q.pearson([1, 2, 3], [5, 5, 5]), 0.0) def test_tune_picks_the_list_shortcut(self): g = q.tune_on_suite(FL, 0.1, trials=3, seed=0) self.assertEqual(g, {"lookup": 0, "list": 1, "yesno": 0, "refuse": 0}) def test_tune_ignores_a_shortcut_that_does_not_pay_on_a_fixed_suite(self): g = q.tune_on_suite(FX, 0.1, trials=3, seed=0) self.assertEqual(sum(g.values()), 0) def test_tune_with_no_tasks_of_a_kind(self): sub = dict(FL, tasks=[t for t in FL["tasks"] if t["kind"] == "list"]) g = q.tune_on_suite(sub, 0.1, trials=2, seed=0) self.assertEqual(g, {"lookup": 0, "list": 1, "yesno": 0, "refuse": 0}) def test_demo(self): d = q.goodhart_demo(FL, FX, slips=(0.04, 0.3), trials=2) self.assertEqual([r["slip"] for r in d["rows"]], [0.04, 0.3]) for r in d["rows"]: self.assertEqual(r["shortcuts"], ["list"]) self.assertAlmostEqual(r["untuned_flawed"], r["untuned_fixed"], places=6) self.assertGreater(r["tuned_flawed"], r["untuned_flawed"] - 0.02) self.assertLess(r["tuned_fixed"], r["untuned_fixed"] - 0.05) self.assertGreater(r["tuned_flawed"] - r["tuned_fixed"], 0.05) self.assertGreater(d["corr"], 0.5) self.assertLess(d["corr"], 1.0) def test_demo_pinned(self): d = q.goodhart_demo(FL, FX, slips=(0.04,), trials=6) r = d["rows"][0] self.assertEqual({k: (round(v, 3) if isinstance(v, float) else v) for k, v in r.items()}, {"slip": 0.04, "shortcuts": ["list"], "untuned_flawed": 0.961, "untuned_fixed": 0.961, "tuned_flawed": 0.966, "tuned_fixed": 0.841}) class TestAddedAfterMutationRun(unittest.TestCase): def test_hull_drops_a_middle_point_under_the_line_and_keeps_one_above(self): self.assertEqual(q.upper_hull({"a": (0, 0), "b": (1, 0.4), "c": (2, 1.0)}), [(0, 0), (2, 1.0)]) self.assertEqual(q.upper_hull({"a": (0, 0), "b": (1, 0.6), "c": (2, 1.0)}), [(0, 0), (1, 0.6), (2, 1.0)]) self.assertEqual(q.upper_hull({"a": (0, 1), "b": (1, 0.5), "c": (2, 0)}), [(0, 1), (2, 0)]) self.assertEqual(q.upper_hull({"a": (1, 0), "b": (2, 0.4), "c": (3, 1.0)}), [(1, 0), (3, 1.0)]) self.assertEqual(q.upper_hull({"a": (1, 0), "b": (2, 0.6), "c": (3, 1.0)}), [(1, 0), (2, 0.6), (3, 1.0)]) self.assertEqual(q.upper_hull({"a": (0, 1), "b": (1, 0.2), "c": (2, 0.8), "d": (3, 0)}), [(0, 1), (2, 0.8), (3, 0)]) def test_default_launch_window_is_press_january_2025(self): self.assertFalse(q.saturation_report([0.095, 0.3], 0.0, 100)["first_in_window"]) self.assertTrue(q.saturation_report([0.085, 0.3], 0.0, 100)["first_in_window"]) def test_default_settings(self): self.assertEqual(len(q.make_world(random.Random(0))[0]), 32) self.assertEqual(q.pass_at_k(4, 2, 4), 1.0) self.assertEqual(q.pass_hat_k(4, 4, 4), 1.0) self.assertEqual(q.baseline_scores(FX), q.baseline_scores(FX, seed=0)) self.assertEqual(q.limit_curve(FX, q.MODEL_A, q.FULL, [3], trials=2), q.limit_curve(FX, q.MODEL_A, q.FULL, [3], trials=2, seed=0)) self.assertEqual(q.scaffold_comparison(FX, {"A": q.MODEL_A}, {"full": q.FULL}, trials=2), q.scaffold_comparison(FX, {"A": q.MODEL_A}, {"full": q.FULL}, trials=2, seed=0)) self.assertEqual(q.cost_accuracy(FX, {"a": q.GRID["a"]}), q.cost_accuracy(FX, {"a": q.GRID["a"]}, q.FULL, 8, 0)) self.assertEqual([r["slip"] for r in q.goodhart_demo(FL, FX, trials=1)["rows"]], [0.04, 0.1, 0.2, 0.3]) def test_fixed_agents_price_and_two_step_request(self): tr = run(task_of(FX, "lookup"), q.EmptyAgent()) self.assertEqual((tr.cost, tr.seconds), (0.5, 1.0)) two = [t for t in FX["tasks"] if t["kind"] == "update" and len(t["plan"]) == 2] self.assertTrue(all(" and close it." in t["request"] for t in two)) self.assertTrue(all("close" not in t["request"] for t in FX["tasks"] if t["kind"] == "update" and len(t["plan"]) == 1)) def test_constants_are_the_documented_ones(self): self.assertEqual(q.MODEL_A, (0.10, 1.0)) self.assertEqual(q.MODEL_B, (0.04, 3.0)) self.assertEqual(q.GRID, {"a": (0.20, 0.5), "b": (0.12, 1.0), "c": (0.08, 1.5), "d": (0.04, 3.0), "e": (0.04, 6.0), "f": (0.15, 4.0)}) def test_mix_agent_defaults_and_price(self): ag = q.MixAgent(0.0, {"list": 1.0}) for seed in range(60): ag.reset(task_of(FX, "refuse"), FX, q.FULL, random.Random(seed)) self.assertIs(ag.cur, ag.plan) self.assertEqual((ag.step_cost, ag.step_seconds), (1.0, 2.0)) def test_default_seeds_are_zero(self): a = q.run_suite(FX, lambda: q.make_agent(q.MODEL_A), q.FULL, 2) self.assertEqual(a, q.run_suite(FX, lambda: q.make_agent(q.MODEL_A), q.FULL, 2, 0)) self.assertEqual(q.tune_on_suite(FL, 0.3, trials=2), q.tune_on_suite(FL, 0.3, trials=2, seed=0)) class TestMain(unittest.TestCase): def test_main_prints_every_section(self): buf = io.StringIO() with redirect_stdout(buf): q.main() out = buf.getvalue() for i in range(1, 11): self.assertIn(f"== {i}. ", out) self.assertIn("frontier: ['a', 'b', 'c', 'd']", out) if __name__ == "__main__": unittest.main()