import contextlib import io import math import unittest import proactive_agents_toolkit as P def opp(t=600, useful=True, signal=0.9, benefit=1.0, text="x", focus=0.0): return P.Opp(t, useful, signal, benefit, text, focus) EXACT = P.Costs(accept=1.0, fatigue=1.0) class TestHelpers(unittest.TestCase): def test_hhmm(self): self.assertEqual(P.hhmm(0), "00:00") self.assertEqual(P.hhmm(545), "09:05") self.assertEqual(P.hhmm(P.DAY + 61), "01:01") def test_mean_clip(self): self.assertEqual(P.mean([]), 0.0) self.assertEqual(P.mean([1, 2, 6]), 3.0) self.assertEqual(P.clip(1.5), 1.0) self.assertEqual(P.clip(-1), 0.0) self.assertEqual(P.clip(0.4), 0.4) self.assertEqual(P.clip(5, 2, 4), 4) def test_rng_for(self): self.assertEqual(P.rng_for("a", 1).random(), P.rng_for("a", 1).random()) self.assertNotEqual(P.rng_for("a", 1).random(), P.rng_for("a", 2).random()) self.assertNotEqual(P.rng_for("a", 1).random(), P.rng_for("b", 1).random()) class TestCron(unittest.TestCase): def test_fields(self): mins, hrs, dows = P.parse_cron("*/15 9-10 0,2") self.assertEqual(mins, {0, 15, 30, 45}) self.assertEqual(hrs, {9, 10}) self.assertEqual(dows, {0, 2}) def test_start_step(self): self.assertEqual(P.parse_cron("5/20 * *")[0], {5, 25, 45}) self.assertEqual(P.parse_cron("0-10/5 * *")[0], {0, 5, 10}) def test_star_covers_range(self): mins, hrs, dows = P.parse_cron("* * *") self.assertEqual((len(mins), len(hrs), len(dows)), (60, 24, 7)) self.assertEqual(max(mins), 59) self.assertEqual(max(hrs), 23) self.assertEqual(max(dows), 6) def test_bad(self): for bad in ("* *", "60 * *", "* 24 *", "* * 7", "*/0 * *", "5-3 * *", "a * *", "* * * *"): with self.subTest(bad): with self.assertRaises(ValueError): P.parse_cron(bad) def test_matches(self): self.assertTrue(P.cron_matches("0 9 0-4", 9 * 60)) self.assertFalse(P.cron_matches("0 9 0-4", 9 * 60 + 1)) self.assertFalse(P.cron_matches("0 9 0-4", 8 * 60)) self.assertTrue(P.cron_matches("0 9 0-4", 4 * P.DAY + 9 * 60)) self.assertFalse(P.cron_matches("0 9 0-4", 5 * P.DAY + 9 * 60)) self.assertTrue(P.cron_matches("0 9 0-4", 7 * P.DAY + 9 * 60)) def test_minute_field_alone(self): self.assertTrue(P.cron_matches("30 * *", 13 * 60 + 30)) self.assertFalse(P.cron_matches("30 * *", 13 * 60 + 31)) class TestTriggers(unittest.TestCase): def test_cron_trigger_needs_tick(self): c = P.Cron("0 9 *") self.assertTrue(c.fires(P.ev(540, "tick"))) self.assertFalse(c.fires(P.ev(540, "file", path="a"))) with self.assertRaises(ValueError): P.Cron("bad") def test_every(self): e = P.Every(30, start=480) self.assertFalse(e.fires(P.ev(450, "tick"))) self.assertTrue(e.fires(P.ev(480, "tick"))) self.assertFalse(e.fires(P.ev(500, "tick"))) self.assertTrue(e.fires(P.ev(510, "tick"))) self.assertFalse(e.fires(P.ev(510, "email"))) with self.assertRaises(ValueError): P.Every(0) def test_at(self): a = P.At(100) self.assertTrue(a.fires(P.ev(100, "tick"))) self.assertFalse(a.fires(P.ev(101, "tick"))) self.assertFalse(a.fires(P.ev(100, "email"))) def test_webhook_source(self): w = P.Webhook("ci") self.assertTrue(w.fires(P.ev(1, "webhook", source="ci"))) self.assertFalse(w.fires(P.ev(1, "webhook", source="x"))) self.assertFalse(w.fires(P.ev(1, "tick"))) def test_file_prefix_and_debounce(self): f = P.FileChange("src/", debounce=5) self.assertTrue(f.fires(P.ev(10, "file", path="src/a.py"))) self.assertFalse(f.fires(P.ev(14, "file", path="src/b.py"))) self.assertTrue(f.fires(P.ev(15, "file", path="src/b.py"))) self.assertFalse(f.fires(P.ev(30, "file", path="docs/x"))) self.assertFalse(f.fires(P.ev(30, "tick"))) def test_file_debounce_measured_from_last_fire(self): f = P.FileChange("src/", debounce=5) self.assertTrue(f.fires(P.ev(10, "file", path="src/a"))) self.assertFalse(f.fires(P.ev(13, "file", path="src/a"))) self.assertFalse(f.fires(P.ev(14, "file", path="src/a"))) self.assertTrue(f.fires(P.ev(15, "file", path="src/a"))) def test_file_no_debounce(self): f = P.FileChange("") self.assertTrue(f.fires(P.ev(1, "file", path="a"))) self.assertTrue(f.fires(P.ev(1, "file", path="a"))) def test_event_get(self): e = P.ev(5, "x", a=1, b="two") self.assertEqual(e.get("a"), 1) self.assertEqual(e.get("b"), "two") self.assertIsNone(e.get("c")) self.assertEqual(e.get("c", 9), 9) self.assertEqual(e, P.ev(5, "x", b="two", a=1)) def test_base_trigger(self): with self.assertRaises(NotImplementedError): P.Trigger().fires(P.ev(0, "tick")) class TestWindow(unittest.TestCase): def test_none_is_always(self): self.assertTrue(P.in_window(5, None)) def test_plain(self): w = (480, 1080) self.assertFalse(P.in_window(479, w)) self.assertTrue(P.in_window(480, w)) self.assertTrue(P.in_window(1079, w)) self.assertFalse(P.in_window(1080, w)) def test_wrap(self): w = (22 * 60, 8 * 60) self.assertTrue(P.in_window(22 * 60, w)) self.assertTrue(P.in_window(23 * 60 + 59, w)) self.assertTrue(P.in_window(0, w)) self.assertTrue(P.in_window(8 * 60 - 1, w)) self.assertFalse(P.in_window(8 * 60, w)) self.assertFalse(P.in_window(12 * 60, w)) self.assertFalse(P.in_window(22 * 60 - 1, w)) def test_equal_means_whole_day_and_later_days(self): self.assertTrue(P.in_window(100, (60, 60))) self.assertTrue(P.in_window(3 * P.DAY + 500, (480, 1080))) self.assertFalse(P.in_window(3 * P.DAY + 100, (480, 1080))) class TestScheduler(unittest.TestCase): def test_handler_none_delivers_name(self): s = P.Scheduler([P.Job("j", P.At(5))]) self.assertEqual(s.step(5), [("j", "j")]) self.assertEqual(s.step(6), []) def test_no_reply_is_silent_but_logged(self): s = P.Scheduler([P.Job("j", P.Every(10), lambda e, t: P.NO_REPLY if t < 20 else "go")]) out = s.run(range(0, 31)) self.assertEqual(out, [(20, "j", "go"), (30, "j", "go")]) self.assertEqual([(l[0], l[3]) for l in s.log], [(0, False), (10, False), (20, True), (30, True)]) s2 = P.Scheduler([P.Job("j", P.At(1), lambda e, t: None)]) self.assertEqual(s2.step(1), []) self.assertEqual(len(s2.log), 1) def test_active_hours_drop(self): s = P.Scheduler([P.Job("j", P.Every(60), active=(120, 240))]) out = s.run(range(0, 400)) self.assertEqual([t for t, n, r in out], [120, 180]) def test_idle_only_defers_to_first_idle_minute(self): s = P.Scheduler([P.Job("j", P.At(10), idle_only=True)]) out = s.run(range(0, 30), busy_fn=lambda t: 5 <= t < 15) self.assertEqual(out, [(15, "j", "j")]) def test_idle_only_coalesces(self): s = P.Scheduler([P.Job("j", P.Every(10), idle_only=True)]) out = s.run(range(0, 45), busy_fn=lambda t: 5 <= t < 33) self.assertEqual([t for t, n, r in out], [0, 33, 40]) def test_pending_dropped_when_window_closes(self): s = P.Scheduler([P.Job("j", P.At(10), idle_only=True, active=(0, 20))]) out = s.run(range(0, 40), busy_fn=lambda t: t < 25) self.assertEqual(out, []) def test_min_gap(self): s = P.Scheduler([P.Job("j", P.Every(5), min_gap=12)]) out = s.run(range(0, 40)) self.assertEqual([t for t, n, r in out], [0, 15, 30]) def test_min_gap_boundary_is_inclusive(self): s = P.Scheduler([P.Job("j", P.Every(10), min_gap=10)]) self.assertEqual([t for t, n, r in s.run(range(0, 31))], [0, 10, 20, 30]) def test_events_by_t_and_two_jobs(self): jobs = [P.Job("w", P.Webhook("a"), lambda e, t: "w:" + e.get("text")), P.Job("f", P.FileChange("s/"), lambda e, t: "f")] s = P.Scheduler(jobs) out = s.run(range(0, 5), {2: [P.ev(2, "webhook", source="a", text="hi"), P.ev(2, "file", path="s/x")]}) self.assertEqual(out, [(2, "w", "w:hi"), (2, "f", "f")]) def test_last_updates_only_on_run(self): j = P.Job("j", P.At(5), active=(100, 200)) s = P.Scheduler([j]) s.step(5) self.assertIsNone(j.last) j2 = P.Job("j", P.At(5)) P.Scheduler([j2]).step(5) self.assertEqual(j2.last, 5) class TestGate(unittest.TestCase): def test_disabled_by_default(self): self.assertEqual(P.WebhookGate("t").post("t", "a"), (404, "disabled")) def test_token(self): g = P.WebhookGate("tok", enabled=True) self.assertEqual(g.post("bad", "a"), (401, "unauthorized")) self.assertEqual(g.post("tok", "a"), (200, "queued")) def test_coalesce_only_with_last(self): g = P.WebhookGate("t", enabled=True) self.assertEqual(g.post("t", "a"), (200, "queued")) self.assertEqual(g.post("t", "b"), (200, "queued")) self.assertEqual(g.post("t", "b"), (200, "coalesced")) self.assertEqual(g.post("t", "a"), (200, "queued")) self.assertEqual(g.queue, ["a", "b", "a"]) def test_full_refuses_without_evicting_but_still_coalesces(self): g = P.WebhookGate("t", max_queue=2, enabled=True) g.post("t", "a") g.post("t", "b") self.assertEqual(g.post("t", "c"), (503, "full")) self.assertEqual(g.post("t", "b"), (200, "coalesced")) self.assertEqual(g.queue, ["a", "b"]) def test_drain(self): g = P.WebhookGate("t", enabled=True) g.post("t", "a") self.assertEqual(g.drain(), ["a"]) self.assertEqual(g.drain(), []) self.assertEqual(g.post("t", "a"), (200, "queued")) class TestRoute(unittest.TestCase): def test_route(self): self.assertEqual(P.route(P.ev(1, "tick")), "main") self.assertEqual(P.route(P.ev(1, "webhook", source="ci"), ("ci",)), "main") self.assertEqual(P.route(P.ev(1, "webhook", source="x"), ("ci",)), "isolated_reader") self.assertEqual(P.route(P.ev(1, "webhook", source="ci")), "isolated_reader") self.assertEqual(P.route(P.ev(1, "email")), "isolated_reader") self.assertEqual(P.route(P.ev(1, "file")), "isolated_reader") class TestTextHelpers(unittest.TestCase): def test_tokens(self): self.assertEqual(P.tokens("The User is writing a Paper"), ["writing", "paper"]) self.assertEqual(P.tokens("a b"), []) def test_jaccard(self): self.assertEqual(P.jaccard("suggest flights v1", "suggest flights v1"), 1.0) self.assertAlmostEqual(P.jaccard("suggest flights v1", "suggest flights v2"), 0.5) self.assertEqual(P.jaccard("alpha beta", "gamma delta"), 0.0) self.assertEqual(P.jaccard("the", "a"), 1.0) class TestAudit(unittest.TestCase): def test_categories(self): a = P.Audit() self.assertEqual(a.category({"text": "Enter your password"}), "credentials") self.assertEqual(a.category({"text": "bank statement"}), "finance") self.assertEqual(a.category({"text": "new prescription"}), "health") self.assertIsNone(a.category({"text": "paper draft"})) def test_allow_and_block(self): a = P.Audit() self.assertTrue(a.allow({"id": "1", "text": "paper"})) self.assertFalse(a.allow({"id": "2", "text": "bank login"})) self.assertEqual(a.blocked, ["2"]) def test_allowed_category(self): a = P.Audit(allowed=("finance",)) self.assertTrue(a.allow({"id": "1", "text": "bank"})) self.assertFalse(a.allow({"id": "2", "text": "password"})) def O(i, t, text): return {"id": i, "t": t, "source": "s", "text": text} class TestUserModel(unittest.TestCase): def test_propose_confidence(self): m = P.UserModel() one = m.propose(O("a", 0, "deadline")) self.assertEqual([(x[0], round(x[2], 3)) for x in one], [("User has an impending deadline", 0.6)]) two = m.propose(O("a", 0, "deadline due")) self.assertAlmostEqual(two[0][2], 0.75) many = m.propose(O("a", 0, "wedding invite rsvp")) self.assertEqual(many[0][2], 1.0) def test_propose_nothing(self): self.assertEqual(P.UserModel().propose(O("a", 0, "hello there")), []) def test_observe_adds(self): m = P.UserModel() r = m.observe(O("a", 5, "draft paper")) self.assertEqual(r["added"], ["User is writing a paper"]) p = m.find("User is writing a paper") self.assertAlmostEqual(p.conf, 0.75) self.assertEqual((p.alpha, p.last_seen, p.obs), (0.3, 5, ["a"])) def test_revise_noisy_or(self): m = P.UserModel() m.observe(O("a", 0, "deadline")) r = m.observe(O("b", 10, "due")) p = m.find("User has an impending deadline") self.assertEqual(r["revised"], ["User has an impending deadline"]) self.assertAlmostEqual(p.conf, 1 - 0.4 * 0.4) self.assertEqual(p.last_seen, 10) self.assertEqual(p.obs, ["a", "b"]) self.assertEqual(len(m.props), 1) def test_revise_keeps_smaller_staleness(self): m = P.UserModel() m.observe(O("a", 0, "recipe")) self.assertEqual(m.find("User is distracted by recipes while writing").alpha, 0.6) m.props[0].alpha = 0.9 m.observe(O("b", 1, "recipe")) self.assertEqual(m.props[0].alpha, 0.6) m.props[0].alpha = 0.1 m.observe(O("c", 2, "recipe")) self.assertEqual(m.props[0].alpha, 0.1) def test_blocked_by_audit(self): m = P.UserModel(P.Audit()) r = m.observe(O("a", 0, "deadline bank")) self.assertTrue(r["blocked"]) self.assertEqual(m.props, []) def test_confirm_at_boundary(self): m = P.UserModel() m.observe(O("a", 0, "deadline")) m.find("User has an impending deadline").conf = 0.8 r = m.observe(O("b", 1, "vacation")) self.assertEqual(r["pending"], ["User is on vacation"]) self.assertIsNone(m.find("User is on vacation")) self.assertEqual(m.find("User has an impending deadline").conf, 0.8) def test_below_confirm_lowers_target(self): m = P.UserModel() m.observe(O("a", 0, "deadline")) m.find("User has an impending deadline").conf = 0.79 r = m.observe(O("b", 1, "vacation")) self.assertEqual(r["lowered"], ["User has an impending deadline"]) self.assertAlmostEqual(m.find("User has an impending deadline").conf, 0.395) self.assertIsNotNone(m.find("User is on vacation")) def test_target_goes_to_zero_below_five_percent(self): m = P.UserModel() m.observe(O("a", 0, "deadline")) m.find("User has an impending deadline").conf = 0.09 m.observe(O("b", 1, "vacation")) self.assertEqual(m.find("User has an impending deadline").conf, 0.0) m2 = P.UserModel() m2.observe(O("a", 0, "deadline")) m2.find("User has an impending deadline").conf = 0.10 m2.observe(O("b", 1, "vacation")) self.assertAlmostEqual(m2.find("User has an impending deadline").conf, 0.05) def test_confirm_yes_and_no(self): m = P.UserModel() m.observe(O("a", 0, "deadline due submit")) m.observe(O("b", 1, "vacation")) self.assertEqual(len(m.pending), 1) before = m.find("User has an impending deadline").conf self.assertTrue(m.confirm("User is on vacation", True)) self.assertIsNotNone(m.find("User is on vacation")) self.assertAlmostEqual(m.find("User has an impending deadline").conf, before * 0.5) self.assertEqual(m.pending, []) m2 = P.UserModel() m2.observe(O("a", 0, "deadline due submit")) m2.observe(O("b", 1, "vacation")) self.assertTrue(m2.confirm("User is on vacation", False)) self.assertIsNone(m2.find("User is on vacation")) self.assertAlmostEqual(m2.find("User has an impending deadline").conf, 0.9) self.assertFalse(m2.confirm("User is on vacation", True)) def test_retrieve_basic_and_empty(self): m = P.UserModel() self.assertEqual(m.retrieve("paper", 0), []) m.observe(O("a", 0, "draft paper")) m.observe(O("b", 0, "wedding")) got = m.retrieve("paper", 0) self.assertEqual([p.text for p in got], ["User is writing a paper"]) self.assertEqual(m.retrieve("", 0), []) self.assertEqual(m.retrieve("zebra", 0), []) def test_retrieve_skips_zero_confidence(self): m = P.UserModel() m.observe(O("a", 0, "draft paper")) m.props[0].conf = 0.0 self.assertEqual(m.retrieve("paper", 0), []) def test_retrieve_recency(self): m = P.UserModel() m.props = [P.Proposition("alpha one", 1, 0.9, 0), P.Proposition("alpha two", 1, 0.0, 0)] now = 3 * P.DAY self.assertEqual(m.retrieve("alpha", now, k=2, lam=1.0)[0].text, "alpha two") m.props[0].alpha, m.props[1].alpha = 0.0, 0.9 self.assertEqual(m.retrieve("alpha", now, k=2, lam=1.0)[0].text, "alpha one") m.props[1].last_seen = now m.props[1].alpha = 0.9 m.props[0].alpha = 0.9 self.assertEqual(m.retrieve("alpha", now, k=2, lam=1.0)[0].text, "alpha two") def test_retrieve_no_future_age(self): m = P.UserModel() m.observe(O("a", 100, "draft paper")) self.assertEqual(len(m.retrieve("paper", 0)), 1) def test_retrieve_k_and_mmr_diversity(self): m = P.UserModel() m.props = [P.Proposition("alpha beta gamma", 1, 0.0, 0), P.Proposition("alpha beta gamma delta", 1, 0.0, 0), P.Proposition("alpha omega", 1, 0.0, 0)] rel_only = m.retrieve("alpha", 0, k=2, lam=1.0) self.assertEqual(len(rel_only), 2) div = m.retrieve("alpha", 0, k=2, lam=0.5) self.assertEqual(len(div), 2) self.assertIn("alpha omega", [p.text for p in div]) self.assertEqual(len(m.retrieve("alpha", 0, k=1)), 1) self.assertEqual(len(m.retrieve("alpha", 0, k=9)), 3) def test_retrieve_decay_formula(self): m = P.UserModel() m.props = [P.Proposition("alpha one", 1, 0.5, 0), P.Proposition("alpha two", 1, 0.0, 0)] got = m.retrieve("alpha", P.DAY, k=2, lam=1.0, kdecay=2.0) self.assertEqual([p.text for p in got], ["alpha two", "alpha one"]) m.props[0].alpha = 0.0 tie = m.retrieve("alpha", P.DAY, k=2, lam=1.0) self.assertEqual(len(tie), 2) class TestDecisions(unittest.TestCase): def test_from_costs_defaults(self): u = P.from_costs(1.0, 0.2, 0.5) self.assertEqual(u, P.Utilities(1.0, -0.2, -0.5, 0.0, 0.5, -0.1)) v = P.from_costs(2.0, 0.2, 0.5, c_ask=0.3, ask_gain=0.5) self.assertEqual((v.d_g, v.d_ng), (0.7, -0.3)) def test_eu_formulas(self): u = P.Utilities(1.0, -0.4, -0.5, 0.0, 0.5, -0.1) self.assertAlmostEqual(P.eu_act(0.25, u), 0.25 - 0.3) self.assertAlmostEqual(P.eu_wait(0.25, u), -0.125) self.assertAlmostEqual(P.eu_ask(0.25, u), 0.125 - 0.075) def test_decide_ties_prefer_less_intrusive(self): u = P.Utilities(1.0, -1.0, 0.0, 0.0, -5.0, -5.0) self.assertEqual(P.decide(0.5, u), "wait") self.assertEqual(P.decide(0.51, u), "act") tie = P.Utilities(1.0, -1.0, 0.0, 0.0, 1.0, -1.0) self.assertEqual(P.decide(0.51, tie), "ask") u2 = P.Utilities(1.0, -1.0, -1.0, 0.0, 0.0, 0.0) self.assertEqual(P.decide(0.0, u2), "wait") def test_thresholds_match_brute_force(self): for u in (P.from_costs(1.0, 0.195, 0.5), P.from_costs(1.0, 2.0, 0.5, 0.2), P.from_costs(1.0, 0.555, 0.5), P.from_costs(1.0, 0.3, 0.1, 5.0)): lo, hi = P.thresholds(u) for i in range(0, 1001): p = i / 1000 want = "wait" if p < lo else ("ask" if p < hi else "act") if abs(p - lo) < 1e-6 or abs(p - hi) < 1e-6: continue self.assertEqual(P.decide(p, u), want, (u, p)) def test_thresholds_pinned(self): lo, hi = P.thresholds(P.from_costs(1.0, 2.0, 0.5, 0.2)) self.assertAlmostEqual(lo, 0.2 / 1.1) self.assertAlmostEqual(hi, 0.75) lo, hi = P.thresholds(P.from_costs(1.0, 0.555, 0.5)) self.assertAlmostEqual(lo, 0.25230, 4) self.assertAlmostEqual(hi, 0.290576, 5) def test_no_ask_zone_collapses_to_act_threshold(self): u = P.from_costs(1.0, 0.3, 0.2, c_ask=5.0) lo, hi = P.thresholds(u) self.assertEqual(lo, hi) self.assertAlmostEqual(lo, P.gumbo_threshold(1.0, 0.3, 0.2)) def test_interrupt_rule_equals_wait_act_crossing(self): for b, fp, fn in ((1.0, 0.555, 0.5), (2.0, 0.1, 0.3), (0.7, 0.9, 0.0)): u = P.from_costs(b, fp, fn) wa = (u.na_ng - u.a_ng) / ((u.a_g - u.na_g) + (u.na_ng - u.a_ng)) self.assertAlmostEqual(wa, P.gumbo_threshold(b, fp, fn)) thr = P.gumbo_threshold(b, fp, fn) for p in (thr - 0.01, thr + 0.01): i, n = P.gumbo_eu(p, b, fp, fn) self.assertEqual(i > n, p > thr) def test_gumbo_eu_pinned(self): i, n = P.gumbo_eu(0.3, 1.0, 0.555, 0.5) self.assertAlmostEqual(i, 0.3 - 0.7 * 0.555) self.assertAlmostEqual(n, -0.15) self.assertAlmostEqual(P.gumbo_threshold(1.0, 0.555, 0.5), 0.555 / 2.055) def test_cross_parallel(self): with self.assertRaises(ValueError): P._cross(0, 1, 1, 2) def test_attention_cost(self): self.assertAlmostEqual(P.attention_cost(0.15, 0.9), 0.15 * 3.7) self.assertEqual(P.attention_cost(0.15, 0.0), 0.15) self.assertAlmostEqual(P.attention_cost(1.0, 0.5, k=2.0), 2.0) class TestCalibration(unittest.TestCase): def test_brier(self): self.assertAlmostEqual(P.brier([0.9, 0.2], [True, False]), (0.01 + 0.04) / 2) self.assertEqual(P.brier([], []), 0.0) self.assertEqual(P.brier([1.0], [True]), 0.0) def test_reliability(self): rows = P.reliability([0.1, 0.15, 0.9, 1.0], [False, True, True, True], bins=5) self.assertEqual(rows[0], (0.0, 0.2, 2, 0.125, 0.5)) self.assertEqual(len(rows), 2) self.assertEqual(rows[1][2], 2) self.assertAlmostEqual(rows[1][3], 0.95) self.assertEqual(rows[1][4], 1.0) def test_reliability_bin_edges(self): rows = P.reliability([0.2, 0.4], [True, True], bins=5) self.assertEqual([r[0] for r in rows], [0.2, 0.4]) def test_ece(self): self.assertAlmostEqual(P.ece([0.1, 0.9], [False, True], 5), 0.1) self.assertEqual(P.ece([], []), 0.0) self.assertAlmostEqual(P.ece([0.9, 0.9, 0.9, 0.9], [True, True, True, False], 5), 0.15) def test_sharpen(self): self.assertAlmostEqual(P.sharpen(0.3, 1.0), 0.3) self.assertAlmostEqual(P.sharpen(0.5, 3.0), 0.5) self.assertAlmostEqual(P.sharpen(0.8, 2.0), 0.64 / 0.68) self.assertAlmostEqual(P.sharpen(0.8, 0.5), math.sqrt(0.8) / (math.sqrt(0.8) + math.sqrt(0.2))) self.assertEqual(P.sharpen(0.0, 2.0), 0.0) self.assertEqual(P.sharpen(1.0, 2.0), 1.0) def test_calibrator(self): c = P.Calibrator(bins=2).fit([0.1, 0.2, 0.9], [False, False, True]) self.assertAlmostEqual(c.rate[0], 1 / 4) self.assertAlmostEqual(c.rate[1], 2 / 3) self.assertAlmostEqual(c.apply(0.49), 1 / 4) self.assertAlmostEqual(c.apply(0.5), 2 / 3) self.assertAlmostEqual(c.apply(1.0), 2 / 3) def test_calibrator_empty_bin_midpoint(self): c = P.Calibrator(bins=4).fit([0.1], [True]) self.assertAlmostEqual(c.rate[0], 2 / 3) self.assertEqual(c.rate[1:], [0.375, 0.625, 0.875]) def test_fit_calibrator_monotone_enough(self): c = P.fit_calibrator(range(0, 5)) self.assertEqual(len(c.rate), 10) self.assertLess(c.rate[0], c.rate[-1]) self.assertTrue(all(0 < r < 1 for r in c.rate)) class TestRateAndGovernor(unittest.TestCase): def test_bucket(self): b = P.TokenBucket(0.5, 2) self.assertEqual([b.allow(0), b.allow(0), b.allow(0)], [True, True, False]) self.assertFalse(b.allow(1)) self.assertTrue(b.allow(2)) self.assertFalse(b.allow(2)) def test_bucket_capacity_cap_and_time_backwards(self): b = P.TokenBucket(1.0, 2) b.allow(0) self.assertEqual([b.allow(1000), b.allow(1000), b.allow(1000)], [True, True, False]) b2 = P.TokenBucket(1.0, 1) self.assertTrue(b2.allow(100)) self.assertFalse(b2.allow(50)) self.assertTrue(b2.allow(101)) def test_governor_quiet_first(self): g = P.Governor(quiet=(22 * 60, 8 * 60), per_min=1, capacity=1, daily_cap=5) self.assertEqual(g.check(23 * 60, "a b"), (False, "quiet")) self.assertEqual(g.check(9 * 60, "a b"), (True, "ok")) self.assertEqual(g.counts, {"quiet": 1, "ok": 1}) def test_governor_daily_cap_per_day(self): g = P.Governor(daily_cap=2) self.assertTrue(g.check(100, "one")[0]) self.assertTrue(g.check(101, "two")[0]) self.assertEqual(g.check(102, "three"), (False, "cap")) self.assertTrue(g.check(P.DAY + 5, "four")[0]) def test_governor_duplicate_window(self): g = P.Governor(dup_window=100) self.assertTrue(g.check(0, "suggest flights v1")[0]) self.assertEqual(g.check(99, "suggest flights v1"), (False, "duplicate")) self.assertTrue(g.check(100, "suggest flights v1")[0]) self.assertTrue(g.check(150, "suggest flights v2")[0]) def test_governor_duplicate_threshold(self): g = P.Governor(dup_thr=0.5) g.check(0, "suggest flights v1") self.assertEqual(g.check(1, "suggest flights v2"), (False, "duplicate")) g2 = P.Governor(dup_thr=0.51) g2.check(0, "suggest flights v1") self.assertTrue(g2.check(1, "suggest flights v2")[0]) def test_governor_rate_and_token_not_spent_on_refusals(self): g = P.Governor(per_min=0.1, capacity=1, dup_window=1000) self.assertTrue(g.check(0, "aa bb")[0]) self.assertEqual(g.check(1, "cc dd"), (False, "rate")) self.assertEqual(g.check(2, "aa bb"), (False, "duplicate")) self.assertTrue(g.check(10, "ee ff")[0]) self.assertEqual(g.counts["rate"], 1) def test_governor_refused_not_recorded(self): g = P.Governor(quiet=(0, 600)) g.check(5, "aa bb") self.assertEqual(g.sent, []) self.assertTrue(g.check(700, "aa bb")[0]) self.assertEqual(len(g.sent), 1) def test_governor_no_limits(self): g = P.Governor() self.assertTrue(all(g.check(i, "t%d x" % i)[0] for i in range(50))) class TestLedger(unittest.TestCase): def test_unknown_level(self): with self.assertRaises(ValueError): P.Ledger().grant("a", "root", 0) def test_levels_order(self): self.assertEqual(P.LEVELS, ["observe", "suggest", "draft", "act_reversible", "act_irreversible"]) def test_authorize_basic(self): L = P.Ledger() self.assertEqual(L.authorize("s", "observe", 0), ("deny", "no live grant")) L.grant("s", "draft", 0) self.assertEqual(L.authorize("s", "observe", 1)[0], "allow") self.assertEqual(L.authorize("s", "draft", 1)[0], "allow") self.assertEqual(L.authorize("s", "act_reversible", 1), ("deny", "grant is below act_reversible")) self.assertEqual(L.authorize("other", "observe", 1)[0], "deny") def test_irreversible(self): L = P.Ledger() L.grant("b", "act_reversible", 0) self.assertEqual(L.authorize("b", "act_irreversible", 1, True)[0], "deny") L.grant("b", "act_irreversible", 0) self.assertEqual(L.authorize("b", "act_irreversible", 1)[0], "ask") self.assertEqual(L.authorize("b", "act_irreversible", 1, confirm=True)[0], "allow") self.assertEqual(L.authorize("b", "act_reversible", 1)[0], "allow") def test_expiry_and_start(self): L = P.Ledger() L.grant("c", "draft", 10, ttl=20) self.assertEqual(L.authorize("c", "draft", 9)[0], "deny") self.assertEqual(L.authorize("c", "draft", 10)[0], "allow") self.assertEqual(L.authorize("c", "draft", 29)[0], "allow") self.assertEqual(L.authorize("c", "draft", 30)[0], "deny") def test_revoke(self): L = P.Ledger() L.grant("s", "observe", 0) L.grant("s", "suggest", 0) self.assertEqual(L.revoke("s", 50), 2) self.assertEqual(L.authorize("s", "observe", 49)[0], "allow") self.assertEqual(L.authorize("s", "observe", 50)[0], "deny") self.assertEqual(L.revoke("s", 60), 0) self.assertEqual(L.max_level("s", 100), -1) def test_max_level_takes_highest_live(self): L = P.Ledger() L.grant("s", "observe", 0) L.grant("s", "draft", 0, ttl=10) self.assertEqual(L.max_level("s", 5), 2) self.assertEqual(L.max_level("s", 10), 0) def test_log_chain(self): L = P.Ledger() self.assertTrue(L.verify()) L.grant("s", "observe", 0) L.authorize("s", "observe", 1) L.revoke("s", 2) self.assertEqual(len(L.log), 3) self.assertEqual(L.log[0][3], "start") self.assertEqual(L.log[1][3], L.log[0][4]) self.assertTrue(L.verify()) t = L.log[1] L.log[1] = (t[0], t[1], "deny", t[3], t[4]) self.assertFalse(L.verify()) def test_log_chain_detects_wrong_link(self): L = P.Ledger() L.grant("s", "observe", 0) L.grant("s", "suggest", 1) t = L.log[1] L.log[1] = (t[0], t[1], t[2], "forged", t[4]) self.assertFalse(L.verify()) def test_authorize_is_logged(self): L = P.Ledger() L.authorize("s", "observe", 3) self.assertEqual(L.log[0][:3], (3, "authorize s observe", "deny")) class TestSimulation(unittest.TestCase): def test_day_blocks_tile(self): b = P.make_day(1) self.assertEqual(b[0].start, 420) self.assertEqual(b[-1].end, 1380) for x, y in zip(b, b[1:]): self.assertEqual(x.end, y.start) self.assertTrue(all(x.end > x.start for x in b)) self.assertTrue(all(x.kind in {"deep", "meeting", "admin", "break"} for x in b)) self.assertEqual(P.make_day(1), b) self.assertNotEqual(P.make_day(2), b) def test_day_custom_range(self): b = P.make_day(0, start=100, end=130) self.assertEqual((b[0].start, b[-1].end), (100, 130)) def test_focus_at(self): b = [P.Block("deep", 10, 20, 0.9), P.Block("break", 20, 30, 0.1)] self.assertEqual(P.focus_at(b, 10), 0.9) self.assertEqual(P.focus_at(b, 19), 0.9) self.assertEqual(P.focus_at(b, 20), 0.1) self.assertEqual(P.focus_at(b, 30), 0.0) self.assertEqual(P.focus_at(b, 5), 0.0) def test_opps(self): d = P.make_day(3) o = P.make_opps(3, d) self.assertEqual(o, P.make_opps(3, d)) self.assertEqual([x.t for x in o], sorted(x.t for x in o)) self.assertTrue(all(420 <= x.t < 1380 for x in o)) self.assertTrue(all(0 <= x.signal <= 1 and 0.5 <= x.benefit <= 1.5 for x in o)) self.assertTrue(all(x.focus == P.focus_at(d, x.t) for x in o)) self.assertTrue(all(x.text.startswith("suggest ") for x in o)) def test_opps_rate_and_signal_means(self): os_ = [x for s in range(30) for x in P.make_opps(s, P.make_day(s))] per_day = len(os_) / 30 self.assertTrue(125 < per_day < 150, per_day) useful = [x.signal for x in os_ if x.useful] useless = [x.signal for x in os_ if not x.useful] self.assertTrue(0.28 < len(useful) / len(os_) < 0.32) self.assertGreater(P.mean(useful), P.mean(useless) + 0.15) def test_opps_per_hour_and_base(self): d = P.make_day(0) self.assertEqual(P.make_opps(0, d, per_hour=0.0), []) allu = P.make_opps(0, d, base=1.0) self.assertTrue(allu and all(x.useful for x in allu)) none = P.make_opps(0, d, base=0.0) self.assertTrue(none and not any(x.useful for x in none)) def test_prob(self): o = opp(signal=0.8) self.assertEqual(P.prob(o), 0.8) self.assertAlmostEqual(P.prob(o, gamma=2.0), 0.64 / 0.68) c = P.Calibrator(2).fit([0.1], [False]) self.assertEqual(P.prob(o, cal=c), c.apply(0.8)) self.assertTrue(0 < P.prob(opp(signal=0.0)) < 1e-3) self.assertTrue(0 < 1 - P.prob(opp(signal=1.0)) < 1e-3) def test_c_fp(self): self.assertAlmostEqual(P.c_fp(P.Costs(), 0.9), 0.555) self.assertAlmostEqual(P.c_fp(P.Costs(c_fp0=0.2, k_focus=1.0), 0.5), 0.3) def test_costs_defaults(self): c = P.Costs() self.assertEqual((c.c_fp0, c.k_focus, c.miss, c.ignored_frac, c.accept, c.fatigue, c.window), (0.15, 3.0, 0.5, 0.3, 0.9, 0.85, 90)) def test_simple_policies(self): o = opp() self.assertFalse(P.never()(o)) self.assertTrue(P.always()(o)) self.assertTrue(P.fixed(0.8)(opp(signal=0.8))) self.assertFalse(P.fixed(0.81)(opp(signal=0.8))) def test_eu_policy_threshold(self): C = P.Costs() d = P.eu_policy(C) thr = 0.555 / 2.055 self.assertTrue(d(opp(signal=thr + 0.01, focus=0.9))) self.assertFalse(d(opp(signal=thr - 0.01, focus=0.9))) thr2 = 0.15 / 1.65 self.assertTrue(d(opp(signal=thr2 + 0.01, focus=0.0))) self.assertFalse(d(opp(signal=thr2 - 0.01, focus=0.0))) def test_eu_policy_benefit_and_miss(self): C = P.Costs(miss=0.0) thr = 0.15 / (1.0 + 0.0 + 0.15) d = P.eu_policy(C) self.assertTrue(d(opp(signal=thr + 0.01))) self.assertFalse(d(opp(signal=thr - 0.01))) d2 = P.eu_policy(P.Costs(), benefit=2.0) thr2 = 0.15 / (2.0 + 1.0 + 0.15) self.assertTrue(d2(opp(signal=thr2 + 0.01))) self.assertFalse(d2(opp(signal=thr2 - 0.01))) def test_eu_policy_aware(self): C = P.Costs() plain, aware = P.eu_policy(C), P.eu_policy(C, aware=True) o = opp(signal=0.30, focus=0.9) self.assertTrue(plain(o, 3)) self.assertFalse(aware(o, 3)) self.assertTrue(aware(o, 0)) thr = 0.555 / (0.85 ** 3 + 0.5 + 0.555) self.assertTrue(aware(opp(signal=thr + 0.01, focus=0.9), 3)) self.assertFalse(aware(opp(signal=thr - 0.01, focus=0.9), 3)) def test_run_policy_utilities(self): os_ = [opp(t=0, useful=True, benefit=1.2), opp(t=1, useful=False, focus=0.9), opp(t=2, useful=True, benefit=0.8), opp(t=3, useful=False)] r = P.run_policy(os_, lambda o, n=0: o.t in (0, 1), 0, EXACT) u = [x.util for x in r.rows] self.assertAlmostEqual(u[0], 1.2) self.assertAlmostEqual(u[1], -0.555) self.assertAlmostEqual(u[2], -0.4) self.assertAlmostEqual(u[3], 0.0) self.assertEqual([x.shown for x in r.rows], [True, True, False, False]) self.assertEqual([x.accepted for x in r.rows], [True, False, False, False]) self.assertAlmostEqual(r.net(), 1.2 - 0.555 - 0.4) def test_ignored_useful_costs_fraction(self): r = P.run_policy([opp(t=0, useful=True, focus=0.0)], P.always(), 0, P.Costs(accept=0.0, fatigue=1.0)) self.assertAlmostEqual(r.rows[0].util, -0.3 * 0.15) self.assertFalse(r.rows[0].accepted) def test_fatigue_window_boundary(self): C = P.Costs(accept=1.0, fatigue=0.0, window=90) a = [opp(t=0, useful=False), opp(t=89, useful=True)] self.assertFalse(P.run_policy(a, P.always(), 0, C).rows[1].accepted) b = [opp(t=0, useful=False), opp(t=90, useful=True)] self.assertTrue(P.run_policy(b, P.always(), 0, C).rows[1].accepted) def test_fatigue_counts_only_ignored_shown(self): C = P.Costs(accept=1.0, fatigue=0.0, window=90) a = [opp(t=0, useful=False), opp(t=10, useful=True)] r = P.run_policy(a, lambda o, n=0: o.t == 10, 0, C) self.assertTrue(r.rows[1].accepted) def test_policy_gets_recent_ignored_count(self): seen = [] C = P.Costs(accept=1.0, fatigue=1.0, window=50) os_ = [opp(t=0, useful=False), opp(t=10, useful=False), opp(t=60, useful=False), opp(t=70, useful=False)] P.run_policy(os_, lambda o, n=0: seen.append(n) or True, 0, C) self.assertEqual(seen, [0, 1, 0, 1]) def test_governor_in_run(self): g = P.Governor(daily_cap=1) r = P.run_policy([opp(t=0, text="aa bb"), opp(t=1, text="cc dd")], P.always(), 0, EXACT, g) self.assertEqual([x.shown for x in r.rows], [True, False]) self.assertEqual([x.reason for x in r.rows], ["ok", "cap"]) r2 = P.run_policy([opp()], P.never(), 0, EXACT, P.Governor()) self.assertEqual(r2.rows[0].reason, "policy") def test_user_draws_deterministic_and_seeded(self): os_ = P.make_opps(1, P.make_day(1)) a = P.run_policy(os_, P.always(), 1) self.assertEqual([x.accepted for x in a.rows], [x.accepted for x in P.run_policy(os_, P.always(), 1).rows]) self.assertNotEqual([x.accepted for x in a.rows], [x.accepted for x in P.run_policy(os_, P.always(), 2).rows]) def test_result_metrics(self): os_ = [opp(t=60, useful=True), opp(t=120, useful=True), opp(t=180, useful=False), opp(t=240, useful=False)] r = P.run_policy(os_, lambda o, n=0: o.t in (60, 180), 0, EXACT) self.assertEqual((r.shown(), r.accepted(), r.useful()), (2, 1, 2)) self.assertEqual(r.precision(), 0.5) self.assertEqual(r.recall(), 0.5) self.assertAlmostEqual(r.per_hour(), 2 / 3) def test_result_empty(self): r = P.Result([]) self.assertEqual((r.net(), r.precision(), r.recall(), r.per_hour()), (0, 0.0, 0.0, 0.0)) r2 = P.run_policy([opp(t=5)], P.never(), 0, EXACT) self.assertEqual((r2.precision(), r2.recall(), r2.per_hour()), (0.0, 0.0, 0.0)) def test_study_and_summarize(self): res = P.study(P.never, range(3)) self.assertEqual(len(res), 3) s = P.summarize(res) self.assertEqual((s["shown"], s["accepted"], s["precision"]), (0, 0, 0)) self.assertLess(s["net"], 0) self.assertEqual(set(s), {"net", "shown", "accepted", "precision", "recall", "per_hour"}) res2 = P.study(P.always, range(2), governor_factory=lambda: P.Governor(daily_cap=3)) self.assertTrue(all(r.shown() == 3 for r in res2)) def test_policy_ordering_in_the_simulation(self): seeds = range(10, 20) C = P.Costs() cal = P.fit_calibrator(range(0, 10)) never = P.summarize(P.study(P.never, seeds, C))["net"] always = P.summarize(P.study(P.always, seeds, C))["net"] mid = P.summarize(P.study(lambda: P.fixed(0.5, cal=cal), seeds, C))["net"] self.assertGreater(mid, never) self.assertGreater(mid, always) self.assertGreater(never, always) class TestNextAction(unittest.TestCase): def test_world(self): seq, fav = P.habit_world(0, n=500) self.assertEqual(len(seq), 500) self.assertEqual((seq, fav), P.habit_world(0, n=500)) self.assertLessEqual(len(set(seq)), 40) self.assertEqual(len(fav), 40) self.assertTrue(all(v in fav for v in fav.values())) follow = sum(1 for a, b in zip(seq, seq[1:]) if fav[a] == b) / 499 self.assertTrue(0.1 < follow < 0.35, follow) def test_world_strength(self): seq, fav = P.habit_world(1, n=2000, strength=0.0) follow = sum(1 for a, b in zip(seq, seq[1:]) if fav[a] == b) / 1999 self.assertLess(follow, 0.06) seq, fav = P.habit_world(1, n=2000, n_actions=5, strength=0.9) self.assertEqual(len(set(seq)), 5) def test_model_counts_and_sample(self): m = P.NextActionModel(["a", "b", "c"], smooth=0.0).fit(["a", "b", "a", "b", "a", "c"]) self.assertEqual(m.counts["a"], {"b": 2, "c": 1}) r = P.rng_for("t") self.assertEqual(set(m.predict("b", 10, r)), {"a"}) got = m.predict("a", 300, P.rng_for("t2")) self.assertEqual(set(got), {"b", "c"}) self.assertTrue(150 < got.count("b") < 250) self.assertEqual(len(m.predict("a", 7, r)), 7) def test_smoothing_allows_unseen(self): m = P.NextActionModel(["a", "b"], smooth=0.5).fit(["a", "a"]) got = m.predict("a", 400, P.rng_for("s")) self.assertIn("b", got) self.assertTrue(got.count("a") > got.count("b")) def test_agreement(self): self.assertEqual(P.agreement(["a", "a", "b", "c"]), 0.5) self.assertEqual(P.agreement(["a"]), 1.0) def test_nap_eval_invariants(self): seq, _ = P.habit_world(0) r = P.nap_eval(seq, seed=0) self.assertEqual(r["n"], len(seq) - int(len(seq) * 0.7)) self.assertTrue(0 <= r["pass1"] <= r["passk"] <= 1) self.assertGreater(r["passk"], r["pass1"] + 0.2) self.assertGreater(r["sel1"], r["pass1"]) self.assertTrue(0 < r["all_conf"] < 1) self.assertEqual(r, P.nap_eval(seq, seed=0)) def test_nap_eval_pinned(self): seq, _ = P.habit_world(0) r = P.nap_eval(seq, seed=0) self.assertEqual(round(r["pass1"], 3), 0.087) self.assertEqual(round(r["passk"], 3), 0.508) self.assertEqual(round(r["sel1"], 3), 0.195) def test_nap_eval_top_share(self): seq, _ = P.habit_world(0) a = P.nap_eval(seq, seed=0, top=0.1)["sel1"] b = P.nap_eval(seq, seed=0, top=1.0)["sel1"] self.assertAlmostEqual(b, P.nap_eval(seq, seed=0)["pass1"]) self.assertGreater(a, b) def test_nap_eval_k(self): seq, _ = P.habit_world(0) self.assertLess(P.nap_eval(seq, k=1)["passk"], P.nap_eval(seq, k=20)["passk"]) self.assertLess(P.nap_eval(seq, k=5)["passk"], P.nap_eval(seq, k=40)["passk"]) def test_risk_coverage(self): rows = [(0.9, 1), (0.8, 1), (0.7, 0), (0.1, 0)] self.assertEqual(P.risk_coverage(rows, (1.0, 0.5, 0.25)), [(1.0, 0.5), (0.5, 1.0), (0.25, 1.0)]) self.assertEqual(P.risk_coverage(list(reversed(rows)), (0.5,)), [(0.5, 1.0)]) self.assertEqual(P.risk_coverage(rows, (0.01,)), [(0.01, 1.0)]) class TestLogMetrics(unittest.TestCase): def test_precision_from_log(self): self.assertEqual(P.precision_from_log([(True, True), (True, False), (False, False), (True, True)]), 2 / 3) self.assertEqual(P.precision_from_log([(False, False)]), 0.0) self.assertEqual(P.precision_from_log([]), 0.0) def test_estimate_recall(self): self.assertAlmostEqual(P.estimate_recall(10, 100, 20, 4), 10 / 30) self.assertEqual(P.estimate_recall(10, 100, 20, 0), 1.0) self.assertEqual(P.estimate_recall(0, 100, 20, 0), 0.0) with self.assertRaises(ValueError): P.estimate_recall(1, 10, 0, 0) def test_f1(self): self.assertAlmostEqual(P.f1(0.4815, 0.9811), 0.6460, 4) self.assertEqual(P.f1(0, 0), 0.0) self.assertEqual(P.f1(1, 1), 1.0) self.assertAlmostEqual(P.f1(0.5, 1.0), 2 / 3) def test_lu_accept(self): self.assertTrue(P.lu_accept(True, True, True)) self.assertFalse(P.lu_accept(True, True, False)) self.assertFalse(P.lu_accept(True, False, False)) self.assertTrue(P.lu_accept(False, False, False)) self.assertFalse(P.lu_accept(False, True, False)) self.assertTrue(P.lu_accept(True, False, True)) def test_wilson(self): self.assertEqual(P.wilson(0, 0), (0.0, 1.0)) lo, hi = P.wilson(5, 10) self.assertAlmostEqual(lo, 0.2366, 3) self.assertAlmostEqual(hi, 0.7634, 3) lo, hi = P.wilson(0, 10) self.assertEqual(round(lo, 6), 0.0) self.assertAlmostEqual(hi, 0.2775, 3) class TestPinned(unittest.TestCase): """Tests added after the first mutation run: each one pins something that a small change had left alone.""" def test_rule_table(self): want = {"User has an impending deadline": (0.6, 0.5), "User is writing a paper": (0.6, 0.3), "User is invited to a wedding": (0.7, 0.2), "User needs formal wear": (0.4, 0.4), "User is distracted by recipes while writing": (0.5, 0.6), "User ignores billing notifications": (0.3, 0.2), "User is on vacation": (0.6, 0.5)} self.assertEqual({r[1]: (r[3], r[4]) for r in P.RULES}, want) self.assertEqual(P.CONFIRM_AT, 0.8) self.assertEqual([r[1] for r in P.RULES if r[2]], ["User is on vacation"]) def test_jaccard_one_side_empty(self): self.assertEqual(P.jaccard("alpha", "the"), 0.0) self.assertEqual(P.jaccard("the", "alpha"), 0.0) def test_scored_bm25_pinned(self): m = P.UserModel() m.props = [P.Proposition("alpha beta", 1, 0.0, 0), P.Proposition("alpha gamma delta", 1, 0.0, 0), P.Proposition("omega", 1, 0.0, 0)] sc = m.scored("alpha", 0) avg = (2 + 3 + 1) / 3 idf = math.log(1 + (3 - 2 + 0.5) / (2 + 0.5)) want = [idf * 2.5 / (1 + 1.5 * (0.25 + 0.75 * L / avg)) for L in (2, 3)] self.assertAlmostEqual(sc[0][1], want[0]) self.assertAlmostEqual(sc[1][1], want[1]) self.assertEqual(sc[2][1], 0.0) self.assertGreater(sc[0][1], sc[1][1]) def test_scored_decay_pinned(self): m = P.UserModel() m.props = [P.Proposition("alpha", 1, 0.25, 0), P.Proposition("alpha", 1, 0.0, 0)] a, b = m.scored("alpha", 2 * P.DAY) self.assertAlmostEqual(a[1] / b[1], math.exp(-0.25 * 2.0 * 2)) a2, b2 = m.scored("alpha", 2 * P.DAY, kdecay=1.0) self.assertAlmostEqual(a2[1] / b2[1], math.exp(-0.25 * 2)) self.assertEqual(m.scored("alpha", 0)[0][1], m.scored("alpha", 0)[1][1]) def test_scored_term_frequency_and_empty(self): m = P.UserModel() self.assertEqual(m.scored("alpha", 0), []) m.props = [P.Proposition("alpha alpha beta", 1, 0, 0), P.Proposition("alpha beta gamma", 1, 0, 0)] sc = m.scored("alpha", 0) self.assertGreater(sc[0][1], sc[1][1]) self.assertEqual(m.scored("the", 0), []) def test_retrieve_default_k_is_three(self): m = P.UserModel() m.props = [P.Proposition("alpha t%d" % i, 1, 0, 0) for i in range(5)] self.assertEqual(len(m.retrieve("alpha", 0)), 3) def test_mmr_select(self): rel = [1.0, 0.9, 0.4] sim = lambda i, j: {frozenset((0, 1)): 0.4}.get(frozenset((i, j)), 0.0) self.assertEqual(P.mmr_select(rel, sim, 3, lam=1.0), [0, 1, 2]) self.assertEqual(P.mmr_select(rel, sim, 3, lam=0.5), [0, 1, 2]) sim2 = lambda i, j: {frozenset((0, 1)): 0.8}.get(frozenset((i, j)), 0.0) self.assertEqual(P.mmr_select(rel, sim2, 2, lam=0.5), [0, 2]) self.assertEqual(P.mmr_select([0.0, 0.5], sim, 5), [1]) self.assertEqual(P.mmr_select([0.5, 0.5], sim, 1), [0]) def test_mmr_diversity_weight(self): rel = [1.0, 0.9, 0.4] sim = lambda i, j: 0.45 if {i, j} == {0, 1} else 0.0 self.assertEqual(P.mmr_select(rel, sim, 2, lam=0.5), [0, 1]) sim2 = lambda i, j: 0.55 if {i, j} == {0, 1} else 0.0 self.assertEqual(P.mmr_select(rel, sim2, 2, lam=0.5), [0, 2]) def test_confirm_only_touches_own_negation(self): extra = (("zzz",), "User sleeps", "User is writing a paper", 0.5, 0.1) old = P.RULES P.RULES = old + [extra] try: m = P.UserModel() m.observe(O("a", 0, "draft paper")) m.observe(O("b", 1, "deadline due submit")) m.observe(O("c", 2, "vacation")) paper = m.find("User is writing a paper").conf self.assertTrue(m.confirm("User is on vacation", True)) self.assertEqual(m.find("User is writing a paper").conf, paper) finally: P.RULES = old def test_every_one_minute_allowed(self): e = P.Every(1) self.assertTrue(e.fires(P.ev(7, "tick"))) def test_defaults_of_binned_functions(self): ps = [0.1, 0.3, 0.5, 0.7, 0.9] ys = [True] * 5 self.assertEqual(len(P.reliability(ps, ys)), 5) self.assertEqual(len(P.reliability([0.19, 0.21], [True, False])), 2) self.assertAlmostEqual(P.ece(ps, ys), P.ece(ps, ys, 5)) self.assertAlmostEqual(P.ece(ps, ys), mean_abs([0.9, 0.7, 0.5, 0.3, 0.1])) self.assertEqual(len(P.Calibrator().fit([0.5], [True]).rate), 10) def test_calibrator_edges(self): c = P.Calibrator(2).fit([0.5], [True]) self.assertEqual(c.rate, [0.25, 2 / 3]) c2 = P.Calibrator(2).fit([1.0], [True]) self.assertEqual(c2.rate, [0.25, 2 / 3]) def test_sharpen_underflow(self): self.assertEqual(P.sharpen(0.5, 5000.0), 0.5) def test_governor_default_duplicate_threshold(self): g = P.Governor() g.check(0, "aa bb cc dd ee") self.assertEqual(g.check(1, "aa bb cc dd ee ff"), (False, "duplicate")) g2 = P.Governor() g2.check(0, "aa bb cc dd ee") self.assertTrue(g2.check(1, "aa bb cc dd ee ff gg")[0]) def test_governor_default_dup_window_is_240(self): g = P.Governor() g.check(0, "aa bb") self.assertFalse(g.check(239, "aa bb")[0]) self.assertTrue(g.check(240, "aa bb")[0]) def test_eu_wait_with_nonzero_baseline(self): u = P.Utilities(1.0, -0.4, -0.5, -0.2, 0.5, -0.1) self.assertAlmostEqual(P.eu_wait(0.25, u), 0.25 * -0.5 + 0.75 * -0.2) def test_signal_spread_and_benefit_range(self): os_ = [x for s in range(20) for x in P.make_opps(s, P.make_day(s))] sig = [x.signal for x in os_ if x.useful] m = P.mean(sig) sd = math.sqrt(P.mean((v - m) ** 2 for v in sig)) self.assertTrue(0.17 < sd < 0.21, sd) self.assertTrue(0.5 <= min(x.benefit for x in os_) < 0.52) self.assertTrue(1.48 < max(x.benefit for x in os_) <= 1.5) def test_governor_counts_one_per_refusal(self): g = P.Governor() g.check(0, "aa bb") g.check(1, "aa bb") g.check(2, "aa bb") self.assertEqual(dict(g.counts), {"ok": 1, "duplicate": 2}) def test_ece_default_bins_pinned(self): self.assertAlmostEqual(P.ece([0.19, 0.21], [True, False]), 0.51) def test_eu_policy_strictly_above(self): C = P.Costs(c_fp0=0.5, k_focus=0.0, miss=0.5) d = P.eu_policy(C) self.assertFalse(d(opp(signal=0.25))) self.assertTrue(d(opp(signal=0.2501))) def test_per_hour_uses_span_not_sum(self): os_ = [opp(t=60), opp(t=240)] r = P.run_policy(os_, P.always(), 0, EXACT) self.assertAlmostEqual(r.per_hour(), 2 / 3) def test_opp_text_variants(self): texts = {x.text.split()[-1] for s in range(5) for x in P.make_opps(s, P.make_day(s))} self.assertEqual(texts, {"v1", "v2", "v3"}) topics = {x.text.split()[1] for s in range(10) for x in P.make_opps(s, P.make_day(s))} self.assertEqual(topics, set(P.TOPICS)) def test_risk_coverage_order_and_default_cuts(self): rows = [(0.9, 0), (0.5, 1)] self.assertEqual(P.risk_coverage(rows, (0.5,)), [(0.5, 0.0)]) rows2 = [(0.9, 1), (0.8, 0), (0.1, 0)] self.assertEqual(P.risk_coverage(rows2, (0.01,)), [(0.01, 1.0)]) self.assertEqual([c for c, a in P.risk_coverage(rows2)], [1.0, 0.5, 0.25, 0.1]) def test_nap_all_conf_pinned(self): seq, _ = P.habit_world(0) self.assertEqual(round(P.nap_eval(seq, seed=0)["all_conf"], 3), 0.231) def test_estimate_recall_one_audited(self): self.assertAlmostEqual(P.estimate_recall(5, 10, 1, 1), 0.3333333333) def test_f1_small_values(self): self.assertAlmostEqual(P.f1(0.3, 0.3), 0.3) self.assertAlmostEqual(P.f1(0.1, 0.2), 2 * 0.02 / 0.3) def mean_abs(xs): return sum(xs) / len(xs) class TestDemo(unittest.TestCase): @classmethod def setUpClass(cls): buf = io.StringIO() with contextlib.redirect_stdout(buf): P.demo() cls.out = buf.getvalue() def test_sections(self): heads = [l for l in self.out.splitlines() if l.startswith("== ")] self.assertEqual(len(heads), 10) self.assertTrue(heads[0].startswith("== 1. ")) self.assertTrue(heads[-1].startswith("== 10. ")) def test_published_row_identities(self): self.assertIn("F1 from recall 0.9811 and precision 0.4815 is 0.6460; the paper's table says 0.6460", self.out) self.assertIn("F1 from recall 1.0000 and precision 0.4978 is 0.6647; the paper's table says 0.6647", self.out) def test_key_lines(self): self.assertIn("(404, 'disabled')", self.out) self.assertIn("(503, 'full')", self.out) self.assertIn("chain valid True; after editing one entry, chain valid False", self.out) self.assertIn("silent runs (NO_REPLY): 15", self.out) def test_deterministic_and_ascii(self): buf = io.StringIO() with contextlib.redirect_stdout(buf): P.demo() self.assertEqual(buf.getvalue(), self.out) self.assertTrue(all(ord(c) < 128 for c in self.out)) if __name__ == "__main__": unittest.main()