import json import random import tempfile import unittest from pathlib import Path import accept import crosscheck import voter_model class VoterModelTests(unittest.TestCase): def test_all_a_is_absorbing(self): self.assertEqual(voter_model.run_once(10, 10, 7)["winner"], "A") def test_all_b_is_absorbing(self): self.assertEqual(voter_model.run_once(10, 0, 7)["winner"], "B") def test_seed_is_reproducible(self): self.assertEqual(voter_model.run_once(12, 6, 7), voter_model.run_once(12, 6, 7)) def test_invalid_initial_count(self): with self.assertRaises(ValueError): voter_model.run_once(10, 11, 7) def test_experiment_counts_runs(self): out = voter_model.run_experiment(8, [3, 4], [7, 8], 10000) self.assertEqual([c["n"] for c in out["conditions"]], [2, 2]) def test_closed_form_probability(self): out = voter_model.run_experiment(10, [4], [7], 10000) self.assertEqual(out["conditions"][0]["closed_form_probability_a"], 0.4) def test_crosscheck_matches(self): out = voter_model.run_experiment(10, [4], [7, 8], 10000) self.assertEqual(crosscheck.check(out)["mismatches"], 0) def test_crosscheck_inequality_mutant_rings(self): out = voter_model.run_experiment(10, [4], [7], 10000) self.assertGreater(crosscheck.check(out, mutant=True)["mismatches"], 0) def test_crosscheck_seed_shift_rings(self): out = voter_model.run_experiment(10, [4], [7, 8], 10000) self.assertGreater(crosscheck.check(out, seed_shift=1)["mismatches"], 0) def test_pair_matches_two_runs_small_and_production_size(self): for n, a, seeds in [(8, 4, range(1, 20)), (100, 50, [7, 3001, 3456, 4000])]: for seed in seeds: with self.subTest(n=n, seed=seed): pair = voter_model.pair_once(seed, n, a) base = voter_model.run_once(n, a, seed) one = voter_model.run_once(n, a + 1, seed) self.assertEqual((pair["winner_base"], pair["steps_base"]), (base["winner"], base["steps"])) self.assertEqual((pair["winner_one"], pair["steps_one"]), (one["winner"], one["steps"])) def test_pair_never_runs_contrary(self): for seed in range(1, 501): row = voter_model.pair_once(seed, 20, 10) self.assertNotEqual((row["winner_base"], row["winner_one"]), ("A", "B")) def test_incremental_d_matches_full_recount_every_step(self): for seed in range(1, 31): n, a = 9, 4 rng = random.Random(seed) idx = list(range(n)); rng.shuffle(idx) base = [int(x < a) for x in idx] one = [int(x < a + 1) for x in idx] cb, co, d = a, a + 1, 1 for _ in range(10000): if not ((0 < cb < n) or (0 < co < n)): break listener = rng.randrange(n) speaker = rng.randrange(n - 1) speaker += speaker >= listener old = base[listener] != one[listener] if 0 < cb < n: before = base[listener]; base[listener] = base[speaker]; cb += base[listener] - before if 0 < co < n: before = one[listener]; one[listener] = one[speaker]; co += one[listener] - before d += int(base[listener] != one[listener]) - int(old) self.assertEqual(d, sum(x != y for x, y in zip(base, one))) def test_vanish_null_exactly_when_outcomes_differ(self): for seed in range(1, 301): row = voter_model.pair_once(seed, 12, 6) self.assertEqual(row["vanish"] is None, row["winner_base"] != row["winner_one"]) def test_snapshot_rule_and_states(self): row = voter_model.snapshots(7, 20, 10) steps = row["snap_steps"] self.assertEqual(steps[0], 0) self.assertEqual(steps[-1], max(row["steps_base"], row["steps_one"])) self.assertEqual(steps, sorted(set(steps))) self.assertEqual(len(steps), len(row["base"])) for base, one, d in zip(row["base"], row["one"], row["d"]): self.assertEqual((len(base), len(one)), (20, 20)) self.assertEqual(d, sum(x != y for x, y in zip(base, one))) def test_expected_steps_n4_exact_boundary_and_symmetry(self): values = voter_model.expected_steps(4) self.assertEqual(values[0], 0) self.assertEqual(values[4], 0) self.assertAlmostEqual(values[1], 5.5) self.assertAlmostEqual(values[2], 7.0) for left, right in zip(values, reversed(values)): self.assertAlmostEqual(left, right) def test_prediction_kinds_positive_negative_and_null(self): cases = [ ({"kind": "eq", "value": 3}, 3, True), ({"kind": "eq", "value": 3}, 2, False), ({"kind": "bool", "value": True}, True, True), ({"kind": "bool", "value": True}, 1, False), ({"kind": "range", "lo": 2, "hi": 4}, 4, True), ({"kind": "range", "lo": 2, "hi": 4}, 5, False), ({"kind": "max", "hi": 4}, 4, True), ({"kind": "max", "hi": 4}, 5, False), ({"kind": "min", "lo": 4}, 4, True), ({"kind": "min", "lo": 4}, 3, False), ({"kind": "in", "values": [2, 4]}, 4, True), ({"kind": "in", "values": [2, 4]}, 3, False), ] for prediction, observed, expected in cases: self.assertEqual(voter_model.prediction_hit(prediction, observed), expected) self.assertFalse(voter_model.prediction_hit(prediction, None)) def test_numeric_metric_path_and_prediction_output(self): summary = {"sweep": {"by_a": {"10": {"win_share_a": 0.1}}}} self.assertEqual(voter_model.metric_value(summary, "sweep.by_a.10.win_share_a"), 0.1) source = {"predictions": [{"id": "P", "text": "x", "metric": "sweep.by_a.10.win_share_a", "kind": "eq", "value": 0.1}]} self.assertEqual(voter_model.check_predictions(summary, source)["hits"], 1) def test_small_summary_is_deterministic_except_runtime_fields(self): first = voter_model.build_summary(list(range(1, 101))) second = voter_model.build_summary(list(range(1, 101))) for value in (first, second): value.pop("generated_at"); value.pop("runtime_s") self.assertEqual(first, second) class AcceptTests(unittest.TestCase): def setUp(self): self.result = voter_model.run_experiment(8, [4], [7, 8], 10000) def test_accepts_matching_count(self): scene = {"data": {"summary": {"n": 2}}} self.assertEqual(accept.inspect(self.result, [scene])["verdict"], "受入") def test_rejects_count_mismatch(self): scene = {"data": {"summary": {"n": 1}}} self.assertEqual(accept.inspect(self.result, [scene])["verdict"], "不受入") def test_rejects_banned_word(self): scene = {"text": accept.BANNED[0], "data": {"summary": {"n": 2}}} self.assertIn("禁止語", accept.inspect(self.result, [scene])["bad"]) def test_rejects_email(self): scene = {"text": "person" + "@" + "example" + ".com", "data": {"summary": {"n": 2}}} self.assertIn("PII候補", accept.inspect(self.result, [scene])["bad"]) if __name__ == "__main__": unittest.main()