import math import random import unittest from unittest import mock import selection_quality_toolkit as q class TestHelpers(unittest.TestCase): def test_words_lowercases_and_splits(self): self.assertEqual(q.words("Fix THE bug_1, now!"), ["fix", "the", "bug_1", "now"]) def test_stable_hash_is_stable_and_bits_matter(self): self.assertEqual(q.stable_hash("abc"), q.stable_hash("abc")) self.assertNotEqual(q.stable_hash("abc"), q.stable_hash("abd")) self.assertLess(q.stable_hash("abc", 32), 2 ** 32) self.assertGreater(max(q.stable_hash(str(i)) for i in range(50)), 2 ** 40) 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]), 2.0) self.assertEqual(q.stdev([]), 0.0) self.assertEqual(q.stdev([5]), 0.0) def test_sigmoid_both_signs(self): self.assertAlmostEqual(q.sigmoid(0), 0.5) self.assertAlmostEqual(q.sigmoid(2), 1 / (1 + math.exp(-2))) self.assertAlmostEqual(q.sigmoid(-2), 1 / (1 + math.exp(2))) 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_logit_inverts_sigmoid_and_clips(self): self.assertAlmostEqual(q.logit(q.sigmoid(1.3)), 1.3) self.assertTrue(math.isfinite(q.logit(0.0))) self.assertTrue(math.isfinite(q.logit(1.0))) self.assertLess(q.logit(0.0), -10) self.assertGreater(q.logit(1.0), 10) def test_dot_norm_cosine_euclid(self): self.assertEqual(q.dot([1, 2, 3], [4, 5, 6]), 32) self.assertEqual(q.norm([3, 4]), 5.0) self.assertAlmostEqual(q.cosine([1, 0], [1, 1]), 1 / math.sqrt(2)) self.assertEqual(q.cosine([0, 0], [1, 1]), 0.0) self.assertEqual(q.cosine([1, 1], [0, 0]), 0.0) self.assertAlmostEqual(q.cosine([1, 2], [-1, -2]), -1.0) self.assertEqual(q.euclid([0, 0], [3, 4]), 5.0) def test_overlap_is_a_share_of_the_first(self): self.assertEqual(q.overlap([1, 2, 3, 4], [3, 4, 5]), 0.5) self.assertEqual(q.overlap([], [1]), 0.0) self.assertEqual(q.overlap([1, 2], [1, 2, 3, 4]), 1.0) class TestInformativeness(unittest.TestCase): def test_entropy(self): self.assertAlmostEqual(q.entropy([0.5, 0.5]), 1.0) self.assertAlmostEqual(q.entropy([0.25] * 4), 2.0) self.assertEqual(q.entropy([1.0, 0.0]), 0.0) self.assertAlmostEqual(q.entropy([0.9, 0.1]), 0.4690, places=3) def test_least_confidence_and_margin(self): self.assertAlmostEqual(q.least_confidence([0.7, 0.2, 0.1]), 0.3) self.assertAlmostEqual(q.margin([0.7, 0.2, 0.1]), 0.5) self.assertAlmostEqual(q.margin([0.4, 0.45, 0.15]), 0.05) self.assertEqual(q.margin([1.0]), 1.0) def test_vote_entropy_and_disagreement(self): self.assertEqual(q.vote_entropy(["a", "a", "a"]), 0.0) self.assertAlmostEqual(q.vote_entropy(["a", "b"]), 1.0) self.assertAlmostEqual(q.vote_entropy(["a", "a", "b", "b"]), 1.0) self.assertEqual(q.vote_entropy([]), 0.0) self.assertEqual(q.disagreement(["a", "a", "a"]), 0.0) self.assertAlmostEqual(q.disagreement(["a", "a", "b"]), 1 / 3) self.assertAlmostEqual(q.disagreement(["a", "b", "c", "d"]), 0.75) self.assertEqual(q.disagreement([]), 0.0) def test_pass_rate_and_variance(self): self.assertAlmostEqual(q.pass_rate([True, False, True, True]), 0.75) self.assertEqual(q.pass_rate([]), 0.0) self.assertEqual(q.pass_variance(0.5), 0.25) self.assertEqual(q.pass_variance(0.0), 0.0) self.assertEqual(q.pass_variance(1.0), 0.0) self.assertAlmostEqual(q.pass_variance(0.2), 0.16) def test_difficulty_and_sometimes_passes(self): self.assertEqual(q.difficulty_from_passes(2, 8), 0.75) self.assertEqual(q.difficulty_from_passes(0, 4), 1.0) self.assertFalse(q.sometimes_passes(0, 8)) self.assertFalse(q.sometimes_passes(8, 8)) self.assertTrue(q.sometimes_passes(1, 8)) self.assertTrue(q.sometimes_passes(7, 8)) self.assertFalse(q.sometimes_passes(1, 1)) def test_rank_ids_orders_and_breaks_ties_by_id(self): s = {5: 0.9, 2: 0.9, 7: 0.1, 3: 0.5} self.assertEqual(q.rank_ids(s), [2, 5, 3, 7]) self.assertEqual(q.rank_ids(s, 2), [2, 5]) self.assertEqual(q.rank_ids(s, largest=False), [7, 3, 2, 5]) self.assertEqual(q.rank_ids(s, 1, largest=False), [7]) self.assertEqual(q.rank_ids(s, 0), []) def test_cap_per_group_keeps_first_in_order(self): items = [("a", 1), ("a", 2), ("b", 3), ("a", 4), ("a", 5), ("b", 6)] self.assertEqual(q.cap_per_group(items, lambda t: t[0], 2), [("a", 1), ("a", 2), ("b", 3), ("b", 6)]) self.assertEqual(q.cap_per_group(items, lambda t: t[0], 0), []) self.assertEqual(len(q.cap_per_group(items, lambda t: t[0], 9)), 6) def test_task_pool_is_seeded_and_has_the_right_shape(self): a = q.make_task_pool(30, ability=0.5, seed=3) b = q.make_task_pool(30, ability=0.5, seed=3) c = q.make_task_pool(30, ability=0.5, seed=4) self.assertEqual(a, b) self.assertNotEqual(a["difficulty"], c["difficulty"]) self.assertEqual(len(a["difficulty"]), 30) self.assertEqual(a["ability"], 0.5) def test_pass_probability_uses_ability_minus_difficulty(self): t = {"ability": 1.0, "difficulty": {0: 1.0, 1: -1.0, 2: 3.0}} self.assertAlmostEqual(q.pass_probability(t, 0), 0.5) self.assertAlmostEqual(q.pass_probability(t, 1), q.sigmoid(2.0)) self.assertAlmostEqual(q.pass_probability(t, 2), q.sigmoid(-2.0)) def test_sample_passes_range_seed_and_extremes(self): t = {"ability": 0.0, "difficulty": {0: -30.0, 1: 30.0, 2: 0.0}} p = q.sample_passes(t, 20, seed=1) self.assertEqual(p[0], 20) self.assertEqual(p[1], 0) self.assertTrue(0 <= p[2] <= 20) self.assertEqual(p, q.sample_passes(t, 20, seed=1)) many = {"ability": 0.0, "difficulty": {i: 0.0 for i in range(40)}} self.assertNotEqual(q.sample_passes(many, 10, seed=1), q.sample_passes(many, 10, seed=2)) self.assertEqual(sorted(p), [0, 1, 2]) def test_sample_passes_rate_matches_probability(self): t = {"ability": 0.0, "difficulty": {i: 0.0 for i in range(200)}} p = q.sample_passes(t, 50, seed=5) self.assertAlmostEqual(sum(p.values()) / (200 * 50), 0.5, delta=0.03) def test_true_signal_peaks_where_p_is_half(self): t = {"ability": 0.0, "difficulty": {0: 0.0, 1: 4.0, 2: -4.0}} s = q.true_signal(t) self.assertAlmostEqual(s[0], 0.25) self.assertLess(s[1], 0.05) self.assertAlmostEqual(s[1], s[2]) def test_demo_more_attempts_find_more_of_the_true_top(self): rows = q.informativeness_demo(n_tasks=200, top=30, ks=(1, 64), seed=1) self.assertEqual(rows[0]["share_0_lt_passes_lt_k"], 0.0) self.assertGreater(rows[1]["share_0_lt_passes_lt_k"], 0.95) self.assertGreater(rows[1]["overlap_with_truth"], rows[0]["overlap_with_truth"] + 0.3) self.assertEqual([r["k"] for r in rows], [1, 64]) class TestDiversity(unittest.TestCase): def line(self, n): return {i: (float(i), 0.0) for i in range(n)} def test_k_center_starts_at_smallest_id_then_goes_far(self): self.assertEqual(q.k_center_greedy(self.line(10), 3), [0, 9, 4]) self.assertEqual(q.k_center_greedy(self.line(10), 1), [0]) self.assertEqual(q.k_center_greedy(self.line(10), 2), [0, 9]) def test_k_center_start_argument_and_limits(self): self.assertEqual(q.k_center_greedy(self.line(10), 2, start=4)[0], 4) self.assertEqual(q.k_center_greedy(self.line(5), 9), [0, 1, 2, 3, 4]) self.assertEqual(q.k_center_greedy(self.line(5), 0), []) self.assertEqual(q.k_center_greedy(self.line(5), 5), [0, 1, 2, 3, 4]) def test_k_center_beats_random_on_clustered_points(self): rng = random.Random(0) pts = {} for c in [(0, 0), (10, 0), (0, 10), (10, 10)]: for _ in range(30 if c == (0, 0) else 5): pts[len(pts)] = (c[0] + rng.gauss(0, .5), c[1] + rng.gauss(0, .5)) kc = q.covering_radius(pts, q.k_center_greedy(pts, 4)) rr = [q.covering_radius(pts, random.Random(s).sample(sorted(pts), 4)) for s in range(20)] self.assertLess(kc, min(rr)) def test_covering_radius(self): pts = self.line(5) self.assertEqual(q.covering_radius(pts, [0]), 4.0) self.assertEqual(q.covering_radius(pts, [2]), 2.0) self.assertEqual(q.covering_radius(pts, [0, 4]), 2.0) self.assertEqual(q.covering_radius(pts, list(pts)), 0.0) def test_score_first_diverse(self): items = [(0, 0.9, (0, 0)), (1, 0.8, (0.1, 0)), (2, 0.7, (5, 0)), (3, 0.6, (5.05, 0)), (4, 0.5, (10, 0))] self.assertEqual(q.score_first_diverse(items, 10, 1.0), [0, 2, 4]) self.assertEqual(q.score_first_diverse(items, 2, 1.0), [0, 2]) self.assertEqual(q.score_first_diverse(items, 10, 0.0), [0, 1, 2, 3, 4]) self.assertEqual(q.score_first_diverse(items, 0, 0.0), []) self.assertEqual(q.score_first_diverse(items, 10, 0.1), [0, 1, 2, 4]) def test_score_first_diverse_distance_is_inclusive_and_ties_go_to_small_id(self): items = [(1, 0.5, (0, 0)), (0, 0.5, (1, 0)), (2, 0.1, (2, 0))] self.assertEqual(q.score_first_diverse(items, 3, 1.0), [0, 1, 2]) self.assertEqual(q.score_first_diverse(items, 3, 1.01), [0]) def test_shingles_and_jaccard(self): self.assertEqual(q.shingles("a b c d", 3), {"a b c", "b c d"}) self.assertEqual(q.shingles("A b", 3), {"a b"}) self.assertEqual(q.shingles("", 3), set()) self.assertEqual(q.shingles("a b c d", 2), {"a b", "b c", "c d"}) self.assertEqual(q.jaccard({1, 2}, {2, 3}), 1 / 3) self.assertEqual(q.jaccard(set(), set()), 1.0) self.assertEqual(q.jaccard({1}, set()), 0.0) def test_minhash_shape_and_determinism(self): sh = q.shingles("one two three four five six seven", 3) a = q.minhash(sh, 32, 0) self.assertEqual(len(a), 32) self.assertEqual(a, q.minhash(sh, 32, 0)) self.assertNotEqual(a, q.minhash(sh, 32, 1)) self.assertEqual(q.minhash(set(), 4, 0), (q._PRIME,) * 4) self.assertEqual(q.minhash(sh, 32, 0), q.minhash(set(sh), 32, 0)) def test_minhash_agreement_estimates_jaccard(self): a = {f"s{i}" for i in range(100)} b = {f"s{i}" for i in range(50, 150)} sa, sb = q.minhash(a, 400, 1), q.minhash(b, 400, 1) est = sum(x == y for x, y in zip(sa, sb)) / 400 self.assertAlmostEqual(est, 1 / 3, delta=0.07) TEXTS = { "a": "the quick brown fox jumps over the lazy dog near the river bank today", "b": "the quick brown fox jumps over the lazy dog near the river bank tonight", "c": "completely different words about cooking pasta with garlic and olive oil", "d": "the quick brown fox jumps over the lazy dog near the river bank today", } def test_lsh_candidates_include_true_pairs_and_are_sorted(self): c = q.lsh_candidates(self.TEXTS) self.assertIn(("a", "d"), c) self.assertIn(("a", "b"), c) self.assertNotIn(("a", "c"), c) self.assertEqual(c, sorted(c)) def test_lsh_requires_bands_that_divide(self): with self.assertRaises(ValueError): q.lsh_candidates(self.TEXTS, num_perm=64, bands=10) def test_near_duplicates_threshold(self): hi = q.near_duplicates(self.TEXTS, 0.99) self.assertEqual([(a, b) for a, b, _ in hi], [("a", "d")]) lo = q.near_duplicates(self.TEXTS, 0.5) self.assertIn(("a", "b"), [(a, b) for a, b, _ in lo]) self.assertEqual(hi[0][2], 1.0) j = dict(((a, b), j) for a, b, j in lo)[("a", "b")] self.assertAlmostEqual(j, q.jaccard(q.shingles(self.TEXTS["a"]), q.shingles(self.TEXTS["b"])), places=4) def test_threshold_is_inclusive(self): t = {"x": "a b c d e f", "y": "a b c d e f"} self.assertEqual(len(q.near_duplicates(t, 1.0)), 1) def test_duplicate_groups_and_dedupe(self): self.assertEqual(q.duplicate_groups(self.TEXTS, 0.5), [["a", "b", "d"]]) kept, dropped = q.dedupe_keep_first(self.TEXTS, 0.5) self.assertEqual(kept, ["a", "c"]) self.assertEqual(dropped, {"b": "a", "d": "a"}) kept, dropped = q.dedupe_keep_first(self.TEXTS, 0.99) self.assertEqual(kept, ["a", "b", "c"]) self.assertEqual(dropped, {"d": "a"}) def test_duplicate_groups_chain_through_union(self): t = {1: "a b c d e f g h", 2: "a b c d e f g h", 3: "a b c d e f g h", 9: "x y z w v u t s"} self.assertEqual(q.duplicate_groups(t, 0.9), [[1, 2, 3]]) kept, dropped = q.dedupe_keep_first(t, 0.9) self.assertEqual(kept, [1, 9]) self.assertEqual(dropped, {2: 1, 3: 1}) def test_union_find_keeps_smallest_id_as_root(self): t = {5: "a b c d e f g h", 2: "a b c d e f g h", 8: "a b c d e f g h"} self.assertEqual(q.duplicate_groups(t, 0.9), [[2, 5, 8]]) self.assertEqual(q.dedupe_keep_first(t, 0.9)[1], {5: 2, 8: 2}) def test_hashed_features_count_unigrams_and_bigrams(self): f = q.hashed_features("a b a", 1000) self.assertEqual(sum(f.values()), 5) self.assertEqual(f[q.stable_hash("a") % 1000], 2 + (1 if q.stable_hash("a b") % 1000 == q.stable_hash("a") % 1000 else 0) + (1 if q.stable_hash("b a") % 1000 == q.stable_hash("a") % 1000 else 0)) self.assertTrue(all(0 <= b < 1000 for b in f)) self.assertEqual(q.hashed_features("", 10), {}) def test_bucket_probs_sum_to_one_and_smooth(self): p = q.bucket_probs(["a b c", "a b"], 16, 1.0) self.assertAlmostEqual(sum(p), 1.0) self.assertTrue(all(x > 0 for x in p)) self.assertEqual(len(p), 16) p0 = q.bucket_probs(["a"], 4, 0.0) self.assertEqual(sorted(p0)[-1], 1.0) pa = q.bucket_probs(["a"], 4, 1.0) self.assertGreater(min(pa), 0.0) def test_dsir_prefers_documents_like_the_target(self): docs = q.make_topic_docs(40, seed=1) target = [t for _tp, t in q.make_topic_docs(10, seed=2, topics=["code"]).values()] lw = q.dsir_log_weights({i: d[1] for i, d in docs.items()}, target) code = [lw[i] for i in docs if docs[i][0] == "code"] other = [lw[i] for i in docs if docs[i][0] != "code"] self.assertGreater(min(code), max(other) - 5) self.assertGreater(q.mean(code), q.mean(other) + 5) top = q.rank_ids(lw, 40) self.assertGreater(q.selection_share(docs, top, "code"), 0.9) def test_gumbel_top_k(self): lw = {i: 0.0 for i in range(100)} lw[7] = 50.0 self.assertIn(7, q.gumbel_top_k(lw, 3, 0)) self.assertEqual(len(set(q.gumbel_top_k(lw, 10, 0))), 10) self.assertEqual(q.gumbel_top_k(lw, 10, 3), q.gumbel_top_k(lw, 10, 3)) self.assertNotEqual(q.gumbel_top_k(lw, 10, 3), q.gumbel_top_k(lw, 10, 4)) self.assertEqual(q.gumbel_top_k(lw, 0, 0), []) def test_gumbel_top_k_follows_the_weights(self): lw = {0: math.log(9.0), 1: math.log(1.0)} first = sum(q.gumbel_top_k(lw, 1, s)[0] == 0 for s in range(400)) / 400 self.assertAlmostEqual(first, 0.9, delta=0.05) def test_topic_docs(self): d = q.make_topic_docs(5, doc_len=9, seed=0) self.assertEqual(len(d), 15) self.assertEqual({t for t, _ in d.values()}, {"code", "food", "sport"}) self.assertTrue(all(len(x.split()) == 9 for _t, x in d.values())) self.assertEqual(d, q.make_topic_docs(5, doc_len=9, seed=0)) self.assertEqual(len(q.make_topic_docs(5, topics=["code"])), 5) self.assertEqual(q.selection_share({0: ("code", ""), 1: ("food", "")}, [0, 1], "code"), 0.5) class TestGradientSimilarity(unittest.TestCase): def test_logistic_p_and_grad(self): self.assertAlmostEqual(q.logistic_p([0, 0], [1, 2]), 0.5) self.assertAlmostEqual(q.logistic_p([1, 1], [1, 2]), q.sigmoid(3)) g = q.example_grad([0, 0], [2, -4], 1) self.assertEqual(g, [-1.0, 2.0]) g0 = q.example_grad([0, 0], [2, -4], 0) self.assertEqual(g0, [1.0, -2.0]) def test_train_logistic_learns_a_separable_rule(self): rng = random.Random(0) data = [] for _ in range(100): x = [rng.gauss(0, 1), rng.gauss(0, 1)] data.append((x, 1 if x[0] > 0 else 0)) w = q.train_logistic(data, steps=300, lr=1.0) self.assertGreater(q.accuracy(w, data), 0.95) self.assertGreater(w[0], 1.0) self.assertLess(abs(w[1]), abs(w[0])) def test_train_logistic_starts_from_w0_and_zero_steps_returns_it(self): self.assertEqual(q.train_logistic([([1.0, 0.0], 1)], steps=0, w0=[0.3, 0.4]), [0.3, 0.4]) self.assertEqual(q.train_logistic([([1.0, 0.0], 1)], steps=0), [0.0, 0.0]) w = q.train_logistic([([1.0, 0.0], 1)], steps=1, lr=1.0) self.assertAlmostEqual(w[0], 0.5) w2 = q.train_logistic([([1.0, 0.0], 1)], steps=1, lr=1.0, w0=[2.0, 0.0]) self.assertAlmostEqual(w2[0], 2.0 + (1 - q.sigmoid(2.0))) def test_train_logistic_divides_by_n(self): one = q.train_logistic([([1.0], 1)], steps=1, lr=1.0) two = q.train_logistic([([1.0], 1), ([1.0], 1)], steps=1, lr=1.0) self.assertAlmostEqual(one[0], two[0]) def test_accuracy_threshold(self): self.assertEqual(q.accuracy([0.0], [([1.0], 1), ([1.0], 0)]), 0.5) self.assertEqual(q.accuracy([1.0], [([1.0], 1), ([-1.0], 0)]), 1.0) self.assertEqual(q.accuracy([-1.0], [([1.0], 1), ([-1.0], 0)]), 0.0) def test_world_shape_and_labels(self): w = q.make_less_world(n_per_task=20, dim=5, seed=1, label_noise=0.0, n_support=4, n_test=30) self.assertEqual(len(w["pool"]), 60) self.assertEqual([e["id"] for e in w["pool"]], list(range(60))) self.assertEqual({e["task"] for e in w["pool"]}, {"A", "B", "C"}) self.assertEqual(len(w["support"]), 4) self.assertEqual(len(w["test"]), 30) for e in w["pool"]: self.assertEqual(e["y"], 1 if q.dot(w["dirs"][e["task"]], e["x"]) > 0 else 0) for x, y in w["test"] + w["support"]: self.assertEqual(y, 1 if q.dot(w["dirs"]["A"], x) > 0 else 0) self.assertEqual(w, q.make_less_world(n_per_task=20, dim=5, seed=1, label_noise=0.0, n_support=4, n_test=30)) def test_world_directions_and_norms(self): w = q.make_less_world(n_per_task=200, dim=6, seed=2, big_frac=0.5, big_scale=5.0) d = w["dirs"] self.assertAlmostEqual(q.norm(d["A"]), 1.0) self.assertAlmostEqual(q.norm(d["C"]), 1.0) self.assertLess(q.cosine(d["A"], d["C"]), -0.4) big = [q.norm(e["x"]) for e in w["pool"] if e["big"]] small = [q.norm(e["x"]) for e in w["pool"] if not e["big"]] self.assertGreater(q.mean(big), 3 * q.mean(small)) self.assertAlmostEqual(sum(e["big"] for e in w["pool"]) / 600, 0.5, delta=0.08) def test_label_noise_flips_some_labels(self): w = q.make_less_world(n_per_task=300, dim=4, seed=3, label_noise=0.2) wrong = sum(e["y"] != (1 if q.dot(w["dirs"][e["task"]], e["x"]) > 0 else 0) for e in w["pool"]) / 900 self.assertAlmostEqual(wrong, 0.2, delta=0.05) def test_projection_keeps_cosines_roughly(self): rng = random.Random(0) m = q.make_projection(50, 400, seed=1) self.assertEqual(len(m), 400) self.assertEqual(len(m[0]), 50) self.assertAlmostEqual(abs(m[0][0]), 1 / math.sqrt(400)) a = [rng.gauss(0, 1) for _ in range(50)] b = [x + rng.gauss(0, 1) for x in a] self.assertAlmostEqual(q.cosine(q.project(m, a), q.project(m, b)), q.cosine(a, b), delta=0.12) self.assertEqual(m, q.make_projection(50, 400, seed=1)) self.assertNotEqual(m, q.make_projection(50, 400, seed=2)) signs = [v > 0 for row in m for v in row] self.assertAlmostEqual(sum(signs) / len(signs), 0.5, delta=0.03) def test_less_scores_prefer_aligned_gradients(self): w = [0.0, 0.0] support = [([1.0, 0.0], 1), ([2.0, 0.0], 1)] pool = [{"id": 0, "x": [1.0, 0.0], "y": 1}, {"id": 1, "x": [1.0, 0.0], "y": 0}, {"id": 2, "x": [0.0, 1.0], "y": 1}, {"id": 3, "x": [5.0, 0.0], "y": 1}] s = q.less_scores(pool, support, w) self.assertAlmostEqual(s[0], 1.0) self.assertAlmostEqual(s[1], -1.0) self.assertAlmostEqual(s[2], 0.0) self.assertAlmostEqual(s[3], 1.0) d = q.less_scores(pool, support, w, use_cosine=False) self.assertGreater(d[3], d[0]) self.assertAlmostEqual(d[0], 0.5 * 0.75) self.assertAlmostEqual(d[1], -0.5 * 0.75) def test_less_scores_with_projection(self): w = [0.0] * 4 support = [([1.0, 1.0, 0.0, 0.0], 1)] pool = [{"id": 0, "x": [1.0, 1.0, 0.0, 0.0], "y": 1}, {"id": 1, "x": [-1.0, -1.0, 0.0, 0.0], "y": 1}] s = q.less_scores(pool, support, w, proj=q.make_projection(4, 64, 0)) self.assertGreater(s[0], 0.95) self.assertLess(s[1], -0.95) def test_less_scores_use_mean_target_gradient(self): w = [0.0, 0.0] pool = [{"id": 0, "x": [1.0, 0.0], "y": 1}] mixed = [([1.0, 0.0], 1), ([-1.0, 0.0], 1)] s = q.less_scores(pool, mixed, w, use_cosine=False) self.assertAlmostEqual(s[0], 0.0) def test_less_demo_selects_more_target_examples_than_random(self): rows = [q.less_demo(sd) for sd in range(4)] sel = q.mean(r["less_cosine"]["from_A"] for r in rows) rnd = q.mean(r["random"]["from_A"] for r in rows) self.assertGreater(sel, rnd + 10) self.assertGreater(q.mean(r["less_cosine"]["acc_A"] for r in rows), q.mean(r["random"]["acc_A"] for r in rows)) for r in rows: self.assertEqual(r["random"]["n"], 90) self.assertEqual(r["less_cosine"]["n"], 90) self.assertEqual(r["full_pool"]["n"], 450) self.assertEqual(r["full_pool"]["from_A"], 150) def test_dot_product_prefers_large_norm_examples(self): rows = [q.less_demo(sd) for sd in range(6)] self.assertGreater(q.mean(r["dot_product"]["big"] for r in rows), q.mean(r["less_cosine"]["big"] for r in rows) + 3) def test_less_demo_with_projection_runs(self): r = q.less_demo(1, proj_dim=6) self.assertEqual(r["less_cosine"]["n"], 90) self.assertEqual(r["less_cosine"]["from_A"] + r["less_cosine"]["from_C"] <= 90, True) def test_less_demo_fraction(self): r = q.less_demo(0, frac=0.1) self.assertEqual(r["random"]["n"], 45) class TestLabelNoise(unittest.TestCase): def test_blobs(self): xs, ys = q.make_blobs(10, 3, 4, 5.0, seed=0) self.assertEqual(len(xs), 30) self.assertEqual(ys, [0] * 10 + [1] * 10 + [2] * 10) self.assertEqual(len(xs[0]), 4) c0 = [q.mean(x[j] for x in xs[:10]) for j in range(2)] self.assertAlmostEqual(c0[0], 5.0, delta=1.5) xs2, _ = q.make_blobs(10, 3, 4, 5.0, seed=0) self.assertEqual(xs, xs2) def test_blob_spreads(self): xs, _ = q.make_blobs(300, 2, 2, 0.0, seed=1, spreads=[0.1, 3.0]) self.assertLess(q.stdev([x[0] for x in xs[:300]]), 0.3) self.assertGreater(q.stdev([x[0] for x in xs[300:]]), 2.0) def test_flip_labels(self): labels = [0, 1, 2] * 100 noisy, flipped = q.flip_labels(labels, 0.2, 3, seed=0) self.assertEqual(flipped, [i for i in range(300) if noisy[i] != labels[i]]) self.assertTrue(all(noisy[i] == (labels[i] + 1) % 3 for i in flipped)) self.assertAlmostEqual(len(flipped) / 300, 0.2, delta=0.06) self.assertEqual(q.flip_labels(labels, 0.0, 3)[1], []) self.assertEqual(len(q.flip_labels(labels, 1.0, 3)[1]), 300) self.assertEqual(q.flip_labels(labels, 0.2, 3, seed=0), (noisy, flipped)) def test_centroid_probs(self): xs = [[0.0], [0.2], [10.0], [10.2]] ys = [0, 0, 1, 1] p = q.centroid_probs(xs, ys, [[0.1], [10.1], [6.1]], 2, tau=1.5) self.assertGreater(p[0][0], 0.99) self.assertGreater(p[1][1], 0.99) self.assertAlmostEqual(p[2][0] + p[2][1], 1.0) self.assertLess(p[2][0], 0.5) self.assertGreater(q.centroid_probs(xs, ys, [[4.0]], 2)[0][0], 0.5) wide = q.centroid_probs(xs, ys, [[2.0]], 2, tau=10.0) narrow = q.centroid_probs(xs, ys, [[2.0]], 2, tau=0.5) self.assertLess(wide[0][0], narrow[0][0]) def test_centroid_probs_missing_class(self): p = q.centroid_probs([[1.0], [2.0]], [0, 0], [[1.5]], 2) self.assertAlmostEqual(sum(p[0]), 1.0) def test_cross_val_probs_are_out_of_sample(self): xs = [[0.0], [0.0], [0.0], [0.0], [0.0], [10.0], [10.0], [10.0], [10.0], [10.0]] ys = [0, 0, 0, 0, 0, 1, 1, 1, 1, 0] p = q.cross_val_probs(xs, ys, 2, folds=5, seed=0) self.assertEqual(len(p), 10) self.assertTrue(all(abs(sum(v) - 1) < 1e-9 for v in p)) self.assertLess(p[9][0], 0.2) self.assertEqual(p, q.cross_val_probs(xs, ys, 2, folds=5, seed=0)) self.assertNotEqual(p, q.cross_val_probs(xs, ys, 2, folds=5, seed=1)) def test_class_thresholds(self): probs = [[0.9, 0.1], [0.7, 0.3], [0.2, 0.8], [0.4, 0.6]] t = q.class_thresholds(probs, [0, 0, 1, 1], 2) self.assertAlmostEqual(t[0], 0.8) self.assertAlmostEqual(t[1], 0.7) def test_confident_joint_by_hand(self): probs = [[0.9, 0.1], [0.8, 0.2], [0.1, 0.9], [0.2, 0.8], [0.15, 0.85], [0.6, 0.4]] labels = [0, 0, 1, 1, 0, 1] C, assigned = q.confident_joint(probs, labels, 2) t = q.class_thresholds(probs, labels, 2) self.assertAlmostEqual(t[0], (0.9 + 0.8 + 0.15) / 3) self.assertAlmostEqual(t[1], (0.9 + 0.8 + 0.4) / 3) self.assertEqual(assigned[4], 1) self.assertEqual(C[0][1], 1) self.assertEqual(assigned[0], 0) self.assertEqual(C[0][0], 2) self.assertEqual(C[1][1], 2) self.assertNotIn(5, [i for i in assigned if assigned[i] == 1 and labels[i] == 0]) self.assertEqual(sum(map(sum, C)), len(assigned)) def test_confident_joint_skips_unconfident_examples(self): probs = [[0.5, 0.5, 0.0], [0.9, 0.05, 0.05], [0.05, 0.9, 0.05], [0.05, 0.05, 0.9], [0.3, 0.3, 0.4]] labels = [0, 0, 1, 2, 0] _C, assigned = q.confident_joint(probs, labels, 3) t = q.class_thresholds(probs, labels, 3) for i, p in enumerate(probs): ok = [j for j in range(3) if p[j] >= t[j]] self.assertEqual(i in assigned, bool(ok)) if ok: self.assertEqual(assigned[i], max(ok, key=lambda j: p[j])) def test_confident_joint_prefers_the_larger_probability_and_lower_class_on_ties(self): probs = [[0.5, 0.5], [0.5, 0.5]] _C, assigned = q.confident_joint(probs, [0, 1], 2) self.assertEqual(assigned, {0: 0, 1: 0}) def test_find_label_issues_orders_by_self_confidence(self): probs = [[0.9, 0.1], [0.05, 0.95], [0.2, 0.8], [0.9, 0.1], [0.1, 0.9], [0.85, 0.15]] labels = [0, 0, 0, 0, 1, 1] bad = q.find_label_issues(probs, labels, 2) self.assertEqual(bad, [1, 5, 2]) self.assertLess(probs[1][0], probs[5][1]) self.assertLess(probs[5][1], probs[2][0]) def test_self_confidence_rank(self): probs = [[0.9, 0.1], [0.4, 0.6], [0.4, 0.6]] self.assertEqual(q.self_confidence_rank(probs, [0, 0, 1]), [1, 2, 0]) self.assertEqual(q.self_confidence_rank(probs, [0, 1, 1]), [1, 2, 0]) self.assertEqual(q.self_confidence_rank(probs, [1, 0, 0]), [0, 1, 2]) def test_precision_recall(self): self.assertEqual(q.precision_recall([1, 2, 3], [2, 3, 4, 5]), (2 / 3, 0.5)) self.assertEqual(q.precision_recall([], [1]), (0.0, 0.0)) self.assertEqual(q.precision_recall([1], []), (0.0, 0.0)) def test_noise_demo_finds_most_flips(self): r = q.label_noise_demo(seed=0) self.assertEqual(r["n"], 300) self.assertGreater(r["cl_recall"], 0.8) self.assertGreater(r["cl_precision"], 0.6) self.assertGreater(r["unfixed_error_rate"], 0.05) self.assertEqual(r["cl_flagged"], round(r["cl_flagged"])) def test_noise_study_averages(self): s = q.label_noise_study(seeds=3) rows = [q.label_noise_demo(seed=i) for i in range(3)] self.assertAlmostEqual(s["cl_precision"], round(q.mean(r["cl_precision"] for r in rows), 3)) self.assertAlmostEqual(s["flipped"], round(q.mean(r["flipped"] for r in rows), 3)) self.assertEqual(set(s), {"flipped", "cl_flagged", "cl_precision", "cl_recall", "base_precision", "base_recall"}) def test_training_dynamics(self): d = q.training_dynamics([0.2, 0.4, 0.6], [False, False, True]) self.assertAlmostEqual(d["confidence"], 0.4) self.assertAlmostEqual(d["variability"], q.stdev([0.2, 0.4, 0.6])) self.assertAlmostEqual(d["correctness"], 1 / 3) self.assertEqual(q.training_dynamics([0.5, 0.5], [True, True])["variability"], 0.0) def test_history_shapes_and_learning(self): rng = random.Random(0) data = [([rng.gauss(0, 1), 1.0], 0) for _ in range(30)] data = [(x, 1 if x[0] > 0 else 0) for x, _ in data] ph, ch = q.train_logistic_history(data, epochs=6, lr=0.2, seed=0) self.assertEqual(len(ph), 30) self.assertEqual(len(ph[0]), 6) self.assertEqual(len(ch[0]), 6) self.assertGreater(q.mean(p[-1] for p in ph), q.mean(p[0] for p in ph)) self.assertTrue(all(0 <= v <= 1 for p in ph for v in p)) self.assertEqual(ph, q.train_logistic_history(data, epochs=6, lr=0.2, seed=0)[0]) self.assertNotEqual(ph, q.train_logistic_history(data, epochs=6, lr=0.2, seed=1)[0]) def test_history_probability_is_of_the_given_label(self): data = [([5.0, 1.0], 1)] * 10 + [([5.0, 1.0], 0)] * 10 ph, ch = q.train_logistic_history(data, epochs=3, lr=0.1, seed=0) for i in range(10): self.assertAlmostEqual(ph[i][-1] + ph[i + 10][-1], 1.0) self.assertNotEqual(ch[i][-1], ch[i + 10][-1]) def test_data_map_regions(self): stats = {i: {"confidence": i / 10, "variability": 0.0} for i in range(10)} stats[4]["variability"] = 0.9 stats[7]["variability"] = 0.8 r = q.data_map_regions(stats, 0.2) self.assertEqual(r["ambiguous"], [4, 7]) self.assertEqual(r["hard"], [0, 1]) self.assertEqual(r["easy"], [8, 9]) r3 = q.data_map_regions(stats, 0.3) self.assertEqual(len(r3["ambiguous"]), 3) self.assertEqual(r3["ambiguous"], [0, 4, 7]) self.assertEqual(r3["hard"], [1, 2, 3]) self.assertEqual(len(set(r3["hard"]) & set(r3["ambiguous"])), 0) self.assertEqual(len(set(r3["easy"]) & set(r3["hard"])), 0) self.assertEqual(q.data_map_regions({0: stats[0]}, 0.1)["ambiguous"], [0]) def test_data_map_hard_region_is_full_of_flipped_labels(self): rows = [q.data_map_demo(seed=s) for s in range(4)] self.assertGreater(q.mean(r["share_flipped_in_hard"] for r in rows), 3 * q.mean(r["base_rate"] for r in rows)) self.assertLess(q.mean(r["share_flipped_in_easy"] for r in rows), 0.05) self.assertEqual(rows[0]["n"], 300) class TestAgreement(unittest.TestCase): def test_percent_agreement(self): self.assertEqual(q.percent_agreement([1, 1, 0, 0], [1, 0, 0, 0]), 0.75) def test_cohen_kappa_by_hand(self): self.assertAlmostEqual(q.cohen_kappa([1, 1, 0, 0], [1, 0, 0, 0]), 0.5) self.assertEqual(q.cohen_kappa([1, 1], [1, 1]), 1.0) self.assertAlmostEqual(q.cohen_kappa([0, 1, 0, 1], [1, 0, 1, 0]), -1.0) self.assertEqual(q.cohen_kappa(["a", "b", "a", "b"], ["a", "b", "a", "b"]), 1.0) def test_scott_pi_by_hand(self): self.assertAlmostEqual(q.scott_pi([1, 1, 0, 0], [1, 0, 0, 0]), 0.46667, places=4) self.assertEqual(q.scott_pi([1, 1], [1, 1]), 1.0) def test_weighted_kappa_by_hand(self): d = lambda x, y: abs(x - y) / 2 self.assertAlmostEqual(q.weighted_kappa([0, 1], [0, 2], d), 0.5) self.assertEqual(q.weighted_kappa([1, 1], [1, 1], d), 1.0) self.assertAlmostEqual(q.weighted_kappa([0, 1, 1, 0], [0, 1, 1, 0], d), 1.0) def test_weighted_with_zero_one_distance_is_cohen(self): a, b = [1, 1, 0, 0, 1, 0], [1, 0, 0, 0, 1, 1] self.assertAlmostEqual(q.weighted_kappa(a, b, q.nominal_distance), q.cohen_kappa(a, b)) def test_artstein_and_poesio_numbers(self): a, b = q.artstein_example() self.assertEqual(len(a), 100) self.assertAlmostEqual(q.percent_agreement(a, b), 0.88) self.assertAlmostEqual(q.scott_pi(a, b), 0.7995, places=4) self.assertAlmostEqual(q.cohen_kappa(a, b), 0.8013, places=4) self.assertAlmostEqual(q.weighted_kappa(a, b, q.artstein_distance), 0.8163, places=4) units = [[x, y] for x, y in zip(a, b)] self.assertAlmostEqual(q.krippendorff_alpha(units, q.artstein_distance), 0.8156, places=4) self.assertAlmostEqual(q.krippendorff_alpha(units), q.scott_pi(a, b), delta=0.002) self.assertGreater(q.krippendorff_alpha(units), q.scott_pi(a, b)) def test_artstein_distance(self): self.assertEqual(q.artstein_distance("stat", "stat"), 0.0) self.assertEqual(q.artstein_distance("stat", "chck"), 0.5) self.assertEqual(q.artstein_distance("chck", "ireq"), 0.5) self.assertEqual(q.artstein_distance("stat", "ireq"), 1.0) def test_fleiss_by_hand(self): r = [["a", "a", "a"], ["a", "a", "b"], ["b", "b", "b"], ["a", "b", "b"]] self.assertAlmostEqual(q.fleiss_multi_pi(r), 1 / 3) self.assertEqual(q.fleiss_multi_pi([["a", "a"], ["a", "a"]]), 1.0) def test_fleiss_equals_scott_for_two_raters(self): a, b = [1, 1, 0, 0, 1, 0, 1], [1, 0, 0, 0, 1, 1, 1] self.assertAlmostEqual(q.fleiss_multi_pi([[x, y] for x, y in zip(a, b)]), q.scott_pi(a, b)) def test_fleiss_validates_rater_counts(self): with self.assertRaises(ValueError): q.fleiss_multi_pi([["a", "b"], ["a"]]) with self.assertRaises(ValueError): q.fleiss_multi_pi([["a"], ["b"]]) def test_alpha_by_hand(self): self.assertAlmostEqual(q.krippendorff_alpha([["a", "a"], ["b", "b"], ["a", "b"]]), 4 / 9) self.assertAlmostEqual(q.krippendorff_alpha([["a", "a", None], ["b", "b"], ["a", "b"]]), 4 / 9) self.assertAlmostEqual(q.krippendorff_alpha([["a", "a", None], ["b", "b"], ["a", "b"], ["a"], [None, "b"]]), 4 / 9) self.assertEqual(q.krippendorff_alpha([["a", "a"], ["a", "a"]]), 1.0) self.assertEqual(q.krippendorff_alpha([["a"], ["b"]]), 1.0) self.assertAlmostEqual(q.krippendorff_alpha([[1, 2], [3, 4]], q.squared_distance), 0.7) def test_alpha_is_symmetric_in_rater_order(self): u = [["a", "b"], ["b", "b"], ["a", "a"], ["b", "a"]] self.assertAlmostEqual(q.krippendorff_alpha(u), q.krippendorff_alpha([x[::-1] for x in u])) self.assertAlmostEqual(q.krippendorff_alpha([["a", "b"], ["b", "a"]]), -0.5) def test_alpha_with_unequal_numbers_of_raters(self): u = [["a", "a", "a"], ["a", "b"], ["b", "b", "b", "b"]] a = q.krippendorff_alpha(u) self.assertTrue(0 < a < 1) def test_distances(self): self.assertEqual(q.nominal_distance("a", "a"), 0.0) self.assertEqual(q.nominal_distance("a", "b"), 1.0) self.assertEqual(q.squared_distance(1, 4), 9.0) def test_prevalence_demo(self): r = q.prevalence_demo() self.assertEqual(r["agreement"], 0.9) self.assertLess(r["kappa"], 0.0) self.assertLess(r["alpha"], 0.0) self.assertAlmostEqual(r["kappa"], -0.053, places=3) def test_ensembles(self): self.assertEqual(q.ensemble_max([0, 3, 1]), 3) self.assertEqual(q.ensemble_median([0, 3, 1]), 1) self.assertEqual(q.ensemble_median([3, 3, 0]), 3) self.assertEqual(q.ensemble_median([2, 0, 0]), 0) def test_keep_by_severity(self): ann = {0: [[0, 1, 0], [1, 1, 1]], 1: [[0, 2, 0], [0, 0, 0]], 2: [[1, 1, 1], [2, 2, 0]], 3: [[0, 0, 0], [0, 0, 3]]} self.assertEqual(q.keep_by_severity(ann), [0]) self.assertEqual(q.keep_by_severity(ann, q.ensemble_median), [0, 1, 3]) self.assertEqual(q.keep_by_severity(ann, q.ensemble_max, 3), [0, 1, 2]) self.assertEqual(q.keep_by_severity(ann, q.ensemble_max, 4), [0, 1, 2, 3]) def test_simulate_annotation(self): ann, ok = q.simulate_annotation(300, 3, 1.0, seed=2) self.assertEqual(len(ann), 300) self.assertEqual(len(ann[0]), 2) self.assertEqual(len(ann[0][0]), 3) for i, crits in ann.items(): for labs in crits: self.assertEqual(len(set(labs)), 1) self.assertEqual(q.keep_by_severity(ann, q.ensemble_max), ok) self.assertEqual(ann, q.simulate_annotation(300, 3, 1.0, seed=2)[0]) ann2, ok2 = q.simulate_annotation(300, 5, 0.5, seed=2) self.assertEqual(len(ann2[0][0]), 5) self.assertTrue(all(0 <= v <= 3 for c in ann2.values() for labs in c for v in labs)) self.assertTrue(all(abs(v - w) <= 2 for c in ann2.values() for labs in c for v in labs for w in labs)) def test_max_rule_trades_recall_for_precision(self): ann, ok = q.simulate_annotation(2000, 3, 0.7, seed=0) mx = q.selection_quality(q.keep_by_severity(ann, q.ensemble_max), ok, 2000) md = q.selection_quality(q.keep_by_severity(ann, q.ensemble_median), ok, 2000) self.assertGreater(mx["precision"], md["precision"]) self.assertLess(mx["recall"], md["recall"] - 0.1) self.assertLess(mx["kept_share"], md["kept_share"]) self.assertGreater(mx["precision"], 0.99) def test_selection_quality(self): r = q.selection_quality([1, 2, 3], [2, 3, 4, 5], 10) self.assertEqual(r, {"kept_share": 0.3, "precision": 0.667, "recall": 0.5}) e = q.selection_quality([], [1], 10) self.assertEqual((e["precision"], e["recall"]), (0.0, 0.0)) self.assertEqual(q.selection_quality([1], [], 5)["recall"], 0.0) class TestIRT(unittest.TestCase): def test_kendall_tau_b(self): self.assertEqual(q.kendall_tau_b([1, 2, 3, 4], [1, 2, 3, 4]), 1.0) self.assertEqual(q.kendall_tau_b([1, 2, 3, 4], [4, 3, 2, 1]), -1.0) self.assertAlmostEqual(q.kendall_tau_b([1, 2, 3], [1, 3, 2]), 1 / 3) self.assertAlmostEqual(q.kendall_tau_b([1, 1, 2], [1, 2, 3]), 2 / math.sqrt(6)) self.assertAlmostEqual(q.kendall_tau_b([1, 2, 3], [1, 1, 2]), 2 / math.sqrt(6)) self.assertEqual(q.kendall_tau_b([1, 1, 1], [1, 2, 3]), 0.0) self.assertEqual(q.kendall_tau_b([1, 1, 2, 2], [1, 1, 2, 2]), 1.0) 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) self.assertAlmostEqual(q.pearson([1, 2, 3, 4], [1, 3, 2, 4]), 0.8) def test_world_shapes(self): w = q.make_irt_world(12, 30, seed=1) self.assertEqual(len(w["resp"]), 12) self.assertEqual(len(w["resp"][0]), 30) self.assertEqual(len(w["theta"][0]), 1) self.assertEqual(len(w["alpha"][0]), 1) self.assertEqual(len(w["beta"]), 30) self.assertTrue(all(v in (0, 1) for r in w["resp"] for v in r)) self.assertTrue(all(a[0] > 0 for a in w["alpha"])) self.assertEqual(w, q.make_irt_world(12, 30, seed=1)) self.assertNotEqual(w["resp"], q.make_irt_world(12, 30, seed=2)["resp"]) def test_world_two_dimensions(self): w = q.make_irt_world(400, 50, seed=1, dims=2) self.assertEqual(len(w["theta"][0]), 2) self.assertEqual(len(w["alpha"][0]), 2) self.assertTrue(all(a >= 0 for al in w["alpha"] for a in al)) r = q.pearson([t[0] for t in w["theta"]], [t[1] for t in w["theta"]]) self.assertAlmostEqual(r, 0.5, delta=0.12) self.assertAlmostEqual(q.stdev([t[1] for t in w["theta"]]), 1.0, delta=0.12) def test_stronger_models_get_more_right(self): w = q.make_irt_world(60, 100, seed=3) acc = [q.mean(r) for r in w["resp"]] self.assertGreater(q.pearson(acc, [t[0] for t in w["theta"]]), 0.9) def test_fit_recovers_abilities_and_difficulties(self): w = q.make_irt_world(40, 120, seed=0) th, al, be = q.fit_irt(w["resp"], "2pl", epochs=120) self.assertGreater(q.pearson(th, [t[0] for t in w["theta"]]), 0.9) self.assertGreater(q.pearson(be, w["beta"]), 0.8) self.assertTrue(all(0.05 <= a <= 5.0 for a in al)) self.assertGreater(q.pearson(al, [a[0] for a in w["alpha"]]), 0.2) self.assertEqual(len(th), 40) self.assertEqual(len(be), 120) def test_one_parameter_model_fixes_alpha(self): w = q.make_irt_world(30, 60, seed=0) th, al, be = q.fit_irt(w["resp"], "1pl", epochs=60) self.assertEqual(al, [1.0] * 60) self.assertGreater(q.pearson(th, [t[0] for t in w["theta"]]), 0.85) def test_fit_orders_models_and_items_by_accuracy(self): w = q.make_irt_world(30, 80, seed=4) th, al, be = q.fit_irt(w["resp"], "2pl", epochs=80) self.assertGreater(q.pearson(th, [q.mean(r) for r in w["resp"]]), 0.95) item_acc = [q.mean(r[i] for r in w["resp"]) for i in range(80)] self.assertLess(q.pearson(be, item_acc), -0.9) def test_fit_theta(self): a, b = [1.0] * 6, [0.0] * 6 up = q.fit_theta(a, b, range(6), [1] * 6) down = q.fit_theta(a, b, range(6), [0] * 6) self.assertGreater(up, 0.5) self.assertAlmostEqual(up, -down) mid = q.fit_theta(a, b, range(6), [1, 1, 1, 0, 0, 0]) self.assertAlmostEqual(mid, 0.0, places=6) g = -up + sum(1.0 * (1 - q.sigmoid(up)) for _ in range(6)) self.assertAlmostEqual(g, 0.0, places=6) def test_fit_theta_uses_only_the_given_items(self): a, b = [1.0] * 4, [0.0] * 4 y = [1, 1, 0, 0] self.assertGreater(q.fit_theta(a, b, [0, 1], y), 0.3) self.assertLess(q.fit_theta(a, b, [2, 3], y), -0.3) def test_fit_theta_uses_alpha_and_beta(self): t1 = q.fit_theta([1.0], [2.0], [0], [1]) t0 = q.fit_theta([1.0], [0.0], [0], [1]) self.assertGreater(t1, t0) ta = q.fit_theta([3.0], [0.0], [0], [1]) self.assertNotAlmostEqual(ta, t0) def test_estimate_pirt(self): a, b = [1.0] * 10, [-1.0, -0.5, 0, 0.5, 1.0] * 2 none = q.estimate_pirt(a, b, [0, 1, 2], [0] * 10) allr = q.estimate_pirt(a, b, [0, 1, 2], [1] * 10) self.assertTrue(0 < none < allr < 1) self.assertAlmostEqual(q.estimate_pirt(a, b, [], [0] * 10), q.mean(q.sigmoid(-x) for x in b)) def test_fisher_information(self): self.assertAlmostEqual(q.fisher_information(2.0, 0.0, 0.0), 1.0) self.assertAlmostEqual(q.fisher_information(1.0, 1.0, 1.0), 0.25) self.assertLess(q.fisher_information(1.0, 0.0, 5.0), 0.01) self.assertAlmostEqual(q.fisher_information(1.0, 1.0, 2.0), q.sigmoid(1) * (1 - q.sigmoid(1))) def test_select_random(self): s = q.select_random(50, 10, seed=1) self.assertEqual(s, sorted(s)) self.assertEqual(len(set(s)), 10) self.assertEqual(s, q.select_random(50, 10, seed=1)) self.assertNotEqual(s, q.select_random(50, 10, seed=2)) def test_select_stratified(self): acc = [0.9, 0.1, 0.5, 0.7, 0.3, 0.2, 0.8, 0.6, 0.4, 0.0] s = q.select_stratified(acc, 2) order = sorted(range(10), key=lambda i: acc[i]) self.assertEqual(s, sorted([order[2], order[7]])) self.assertEqual(q.select_stratified(acc, 10), list(range(10))) def test_select_irt_grid(self): s = q.select_irt_grid([1.0] * 4, [-2.0, -1.0, 1.0, 2.0], [-2.0, -1.0, 1.0, 2.0], 2) self.assertEqual(s, [1, 3]) s4 = q.select_irt_grid([1.0] * 4, [-2.0, -1.0, 1.0, 2.0], [-2.0, -1.0, 1.0, 2.0], 4) self.assertEqual(s4, [0, 1, 2, 3]) hi = q.select_irt_grid([0.2, 3.0, 0.2], [0.0, 0.0, 0.0], [0.0, 0.0, 0.0], 1) self.assertEqual(hi, [1]) def test_irt_grid_does_not_repeat_items(self): s = q.select_irt_grid([1.0] * 5, [0.0] * 5, [0.0] * 20, 5) self.assertEqual(s, [0, 1, 2, 3, 4]) def test_select_anchors_by_hand(self): resp = [[1, 1, 0, 0], [1, 1, 0, 1], [1, 1, 0, 0]] anchors, w = q.select_anchors(resp, 2) self.assertEqual(anchors, [3, 0]) self.assertEqual(w, [0.5, 0.5]) a1, w1 = q.select_anchors(resp, 1) self.assertEqual((a1, w1), ([3], [1.0])) a4, w4 = q.select_anchors(resp, 4) self.assertAlmostEqual(sum(w4), 1.0) def test_weighted_accuracy(self): self.assertAlmostEqual(q.weighted_accuracy([0, 2], [0.75, 0.25], [1, 0, 0]), 0.75) self.assertAlmostEqual(q.weighted_accuracy([0, 1], [2, 2], [1, 0]), 0.5) def test_study_structure_and_determinism(self): r = q.tiny_benchmark_study(0, n_train=20, n_test=10, n_items=60, k=10, epochs=40) self.assertEqual(len(r), 6) self.assertTrue(all(-1 <= v["tau"] <= 1 and 0 <= v["mae"] <= 1 for v in r.values())) self.assertEqual(r, q.tiny_benchmark_study(0, n_train=20, n_test=10, n_items=60, k=10, epochs=40)) self.assertNotEqual(r, q.tiny_benchmark_study(1, n_train=20, n_test=10, n_items=60, k=10, epochs=40)) self.assertIn("IRT-chosen, 2PL ability", r) def test_full_benchmark_selection_is_exact(self): r = q.tiny_benchmark_study(0, n_train=20, n_test=10, n_items=30, k=30, epochs=30) self.assertEqual(r["random, mean"], {"tau": 1.0, "mae": 0.0}) self.assertEqual(r["stratified by difficulty, mean"], {"tau": 1.0, "mae": 0.0}) def test_average_matches_the_mean_of_studies(self): kw = dict(n_train=20, n_test=10, n_items=40, k=8, epochs=30) avg = q.tiny_benchmark_average(3, **kw) rows = [q.tiny_benchmark_study(s, **kw) for s in range(3)] name = "random, mean" self.assertAlmostEqual(avg[name]["tau"], round(q.mean(r[name]["tau"] for r in rows), 3)) self.assertAlmostEqual(avg[name]["tau_sd"], round(q.stdev([r[name]["tau"] for r in rows]), 3)) self.assertAlmostEqual(avg[name]["mae"], round(q.mean(r[name]["mae"] for r in rows), 4)) def test_irt_selection_beats_random_on_error(self): kw = dict(n_train=40, n_test=20, n_items=120, k=15, epochs=60) avg = q.tiny_benchmark_average(6, **kw) self.assertLess(avg["IRT-chosen, 2PL ability"]["mae"], avg["random, mean"]["mae"]) self.assertGreater(avg["IRT-chosen, 2PL ability"]["tau"], 0.5) class TestEvalGen(unittest.TestCase): def A(self, name, crit, fn): return q.Assertion(name, crit, fn) def test_coverage_ffr_alignment_by_hand(self): fails = [True, True, False, False, True] good = [False, False, False, True, True] self.assertAlmostEqual(q.coverage(fails, good), 2 / 3) self.assertAlmostEqual(q.false_failure_rate(fails, good), 0.5) self.assertAlmostEqual(q.alignment(2 / 3, 0.5), 4 / 7) self.assertEqual(q.alignment(0.0, 1.0), 0.0) self.assertEqual(q.alignment(1.0, 0.0), 1.0) def test_coverage_and_ffr_edge_cases(self): self.assertEqual(q.coverage([True, False], [True, True]), 0.0) self.assertEqual(q.false_failure_rate([True, False], [False, False]), 0.0) self.assertEqual(q.coverage([True, True], [False, False]), 1.0) self.assertEqual(q.false_failure_rate([True, True], [True, True]), 1.0) def test_paper_table_one(self): self.assertEqual(q.paper_table_alignment(), {"medical EvalGen and SPADE": 48.29, "product EvalGen": 66.46, "product SPADE": 54.35}) def test_set_fails_is_any(self): a = self.A("a", "c1", lambda o: o["x"] > 1) b = self.A("b", "c2", lambda o: o["x"] < 5) outs = [{"x": 0}, {"x": 3}, {"x": 9}] self.assertEqual(q.set_fails([a], outs), [True, False, False]) self.assertEqual(q.set_fails([a, b], outs), [True, False, True]) self.assertEqual(q.set_fails([], outs), [False, False, False]) def test_selectivity_and_failure_score(self): a = self.A("a", "c", lambda o: o["x"] <= 1) b = self.A("b", "c", lambda o: o["x"] <= 4) outs = [{"id": i, "x": i} for i in range(6)] self.assertAlmostEqual(q.selectivity(a, outs), 2 / 6) sel = {"a": 2 / 6, "b": 5 / 6} self.assertEqual(q.failure_score(outs[0], [a, b], sel), 0.0) self.assertAlmostEqual(q.failure_score(outs[3], [a, b], sel), 2 / 6) self.assertAlmostEqual(q.failure_score(outs[5], [a, b], sel), 7 / 6) def test_grading_order(self): a = self.A("a", "c", lambda o: o["x"] <= 1) b = self.A("b", "c", lambda o: o["x"] <= 4) outs = [{"id": i, "x": i} for i in range(6)] self.assertEqual(q.grading_order(outs, [a, b], "highest"), [5, 2, 3, 4, 0, 1]) self.assertEqual(q.grading_order(outs, [a, b], "lowest"), [0, 1, 2, 3, 4, 5]) self.assertEqual(q.grading_order(outs, [a, b], "alternating"), [5, 0, 2, 1, 3, 4]) r = q.grading_order(outs, [a, b], "random", seed=3) self.assertEqual(sorted(r), list(range(6))) self.assertEqual(r, q.grading_order(outs, [a, b], "random", seed=3)) self.assertNotEqual(q.grading_order(outs, [a, b], "random", seed=3), q.grading_order(outs, [a, b], "random", seed=4)) with self.assertRaises(ValueError): q.grading_order(outs, [a, b], "sideways") def test_alternating_has_no_repeats(self): outs = q.make_outputs(50, 1) o = q.grading_order(outs, q.candidate_assertions(), "alternating") self.assertEqual(sorted(o), list(range(50))) def test_select_per_criterion(self): outs = [{"id": i, "x": i} for i in range(10)] good = {i: i < 5 for i in range(10)} cands = [self.A("le4", "c", lambda o: o["x"] <= 4), self.A("le6", "c", lambda o: o["x"] <= 6), self.A("le2", "c", lambda o: o["x"] <= 2), self.A("other", "d", lambda o: True)] pick = q.select_per_criterion(cands, outs, set(range(10)), good) self.assertEqual(pick["c"].name, "le4") self.assertEqual(pick["d"].name, "other") self.assertEqual(sorted(pick), ["c", "d"]) none = q.select_per_criterion(cands[2:3], outs, set(range(10)), good, ffr_max=0.1) self.assertIsNone(none["c"]) sub = q.select_per_criterion(cands, outs, {0, 1, 8, 9}, good) self.assertIn(sub["c"].name, ("le4", "le2")) self.assertEqual(q.select_per_criterion(cands[:3], outs, set(range(10)), good, ffr_max=0.6)["c"].name, "le4") def test_select_per_criterion_ties_go_to_the_name_that_sorts_first(self): outs = [{"id": 0, "x": 0}, {"id": 1, "x": 1}] good = {0: True, 1: False} a = self.A("zeta", "c", lambda o: o["x"] == 0) b = self.A("alpha", "c", lambda o: o["x"] == 0) self.assertEqual(q.select_per_criterion([a, b], outs, {0, 1}, good)["c"].name, "alpha") def test_ffr_limit_is_inclusive(self): outs = [{"id": i, "x": i} for i in range(5)] good = {0: True, 1: True, 2: True, 3: True, 4: False} a = self.A("a", "c", lambda o: o["x"] != 0) self.assertIsNotNone(q.select_per_criterion([a], outs, set(range(5)), good, ffr_max=0.25)["c"]) self.assertIsNone(q.select_per_criterion([a], outs, set(range(5)), good, ffr_max=0.2499)["c"]) def test_outputs_and_rules(self): o = q.make_outputs(300, 1) self.assertEqual(len(o), 300) self.assertEqual([x["id"] for x in o], list(range(300))) self.assertTrue(all(20 <= x["words"] <= 90 for x in o)) self.assertTrue(all(x["proper_frac"] in (0.0, 1 / 3, 0.5, 2 / 3, 1.0) for x in o)) self.assertAlmostEqual(q.mean(x["hashtag"] for x in o), 0.3, delta=0.08) self.assertEqual(o, q.make_outputs(300, 1)) self.assertNotEqual(o, q.make_outputs(300, 2)) self.assertTrue(q.user_grade_strict({"proper_frac": 1.0, "words": 60})) self.assertFalse(q.user_grade_strict({"proper_frac": 1.0, "words": 61})) self.assertFalse(q.user_grade_strict({"proper_frac": 2 / 3, "words": 10})) self.assertTrue(q.user_grade_relaxed({"proper_frac": 0.5, "words": 60})) self.assertFalse(q.user_grade_relaxed({"proper_frac": 1 / 3, "words": 10})) self.assertFalse(q.user_grade_relaxed({"proper_frac": 1.0, "words": 61})) def test_candidates(self): c = q.candidate_assertions() self.assertEqual(len(c), 9) self.assertEqual(len({a.name for a in c}), 9) self.assertEqual({a.criterion for a in c}, {"proper_nouns", "length", "hashtags"}) by = {a.name: a for a in c} self.assertTrue(by["proper>=0.6"].fn({"proper_frac": 2 / 3})) self.assertFalse(by["proper>=0.6"].fn({"proper_frac": 0.5})) self.assertTrue(by["proper>=1.0"].fn({"proper_frac": 1.0})) self.assertFalse(by["proper>=1.0"].fn({"proper_frac": 2 / 3})) self.assertTrue(by["proper>=0.3"].fn({"proper_frac": 1 / 3})) self.assertTrue(by["words<=40"].fn({"words": 40})) self.assertFalse(by["words<=40"].fn({"words": 41})) self.assertTrue(by["no_hashtag"].fn({"hashtag": False})) self.assertFalse(by["no_hashtag"].fn({"hashtag": True})) def test_set_alignment(self): outs = [{"id": 0, "x": 0}, {"id": 1, "x": 1}, {"id": 2, "x": 2}, {"id": 3, "x": 3}] a = self.A("a", "c", lambda o: o["x"] <= 1) good = [True, True, False, True] self.assertAlmostEqual(q.set_alignment([a], outs, good), q.alignment(1.0, 1 / 3)) def test_drift_demo(self): rows = [q.criteria_drift_demo(seed=s) for s in range(4)] for r in rows: self.assertEqual(r["graded"], 20) self.assertLess(r["early_set_vs_relaxed_grades"], r["early_set_vs_strict_grades"]) self.assertGreater(r["regraded_set_vs_relaxed_grades"], r["early_set_vs_relaxed_grades"]) self.assertIn("proper>=1.0", rows[1]["picked_early"]) self.assertIn("proper>=0.6", rows[1]["picked_after_drift"]) self.assertEqual(q.criteria_drift_demo(seed=1), rows[1]) self.assertEqual(q.criteria_drift_demo(seed=1, n_grade=12)["graded"], 12) def test_drift_demo_uses_the_policy(self): a = q.criteria_drift_demo(seed=0, policy="alternating") b = q.criteria_drift_demo(seed=0, policy="random") self.assertNotEqual(a, b) def test_sampling_study(self): r = q.sampling_study(worlds=6) self.assertEqual(set(r), {"random", "highest", "lowest", "alternating"}) self.assertTrue(all(0 <= v["mean"] <= 1 and v["sd"] >= 0 for v in r.values())) self.assertGreater(r["random"]["sd"], r["alternating"]["sd"]) self.assertEqual(r, q.sampling_study(worlds=6)) self.assertEqual(set(q.sampling_study(("random",), worlds=2)), {"random"}) self.assertGreater(r["random"]["mean"], r["alternating"]["mean"]) class TestPipeline(unittest.TestCase): def T(self, id, task="t0", prompt="p", actions=None, passed=True, consent=True, defect="", repo="r"): return q.Trace(id, task, repo, prompt, actions if actions is not None else ["view a.py", "submit"], passed, consent, defect) def test_trace_defaults(self): t = q.Trace(1, "t", "r", "p", ["submit"], True) self.assertTrue(t.consent_train) self.assertEqual(t.defect, "") def test_make_prompt(self): a = q.make_prompt("libx", "f", "breaks.") self.assertEqual(a, q.make_prompt("libx", "f", "breaks.")) self.assertTrue(a.startswith("In libx the function f breaks. Example:")) self.assertTrue(a.endswith("Fix the bug and make the failing tests pass.")) rng = random.Random(1) b, c = q.make_prompt("libx", "f", "breaks.", rng), q.make_prompt("libx", "f", "breaks.", rng) self.assertNotEqual(b, c) self.assertIn(" of ", a) def test_reword_changes_one_word(self): p = q.make_prompt("libx", "f", "breaks.") for s in range(20): r = q._reword(p, random.Random(s)) self.assertEqual(len(r.split()), len(p.split())) self.assertEqual(sum(a != b for a, b in zip(r.split(), p.split())), 1) self.assertGreaterEqual(q.jaccard(q.shingles(r), q.shingles(p)), 0.6) def test_eval_prompts(self): e = q.make_eval_prompts(12, 1) self.assertEqual(len(set(e)), 12) self.assertEqual(e, q.make_eval_prompts(12, 1)) self.assertNotEqual(e, q.make_eval_prompts(12, 2)) def test_pool_has_every_defect_and_is_seeded(self): pool = q.make_trace_pool(80, 0) d = {t.defect for t in pool} self.assertTrue({"", "loop", "tamper", "pii", "truncated", "near_dup", "leak", "weak_tests"} <= d) self.assertEqual([t.id for t in pool], list(range(len(pool)))) self.assertEqual(pool, q.make_trace_pool(80, 0)) self.assertNotEqual(pool, q.make_trace_pool(80, 1)) self.assertEqual(len({t.task_id for t in pool}), 80) def test_defect_traces_look_like_their_defect(self): pool = q.make_trace_pool(120, 2) for t in pool: if t.defect == "loop": self.assertGreaterEqual(q.longest_run(t.actions), 12) elif t.defect == "tamper": self.assertTrue(any(a.startswith("edit tests/") for a in t.actions)) self.assertTrue(t.passed) elif t.defect == "pii": self.assertTrue(q.has_pii(t.prompt)) elif t.defect == "truncated": self.assertNotEqual(t.actions[-1], "submit") elif t.defect == "weak_tests": self.assertTrue(t.passed) elif t.defect == "": self.assertLess(q.longest_run(t.actions), 10) self.assertEqual(t.actions[-1], "submit") self.assertFalse(q.has_pii(t.prompt)) def test_pool_eval_prompts_and_leaks(self): ev = q.make_eval_prompts(12, 1) pool = q.make_trace_pool(120, 3, ev) leaks = [t for t in pool if t.defect == "leak"] self.assertTrue(leaks) for t in leaks: self.assertTrue(any(q.jaccard(q.shingles(t.prompt), q.shingles(e)) >= 0.6 for e in ev)) clean = [t for t in pool if t.defect == ""] for t in clean[:40]: self.assertFalse(any(q.jaccard(q.shingles(t.prompt), q.shingles(e)) >= 0.6 for e in ev)) def test_consent_rate(self): pool = q.make_trace_pool(150, 0) self.assertAlmostEqual(q.mean(not t.consent_train for t in pool), 0.12, delta=0.04) def test_longest_run(self): self.assertEqual(q.longest_run(["view a", "view b", "grep c", "view d"]), 2) self.assertEqual(q.longest_run([]), 0) self.assertEqual(q.longest_run(["view a"] * 5), 5) self.assertEqual(q.longest_run(["a x", "b", "b", "b", "a"]), 3) self.assertEqual(q.longest_run(["", "", "x"]), 2) def test_has_pii(self): self.assertTrue(q.has_pii("mail me at a.b+c@site.org now")) self.assertTrue(q.has_pii("call 555-0100 today")) self.assertTrue(q.has_pii("call 555 0100 today")) self.assertFalse(q.has_pii("version 12-345 is fine")) self.assertFalse(q.has_pii("nothing here @ all")) def test_gates_one_by_one(self): ts = [self.T(0), self.T(1, consent=False), self.T(2, passed=False), self.T(3, actions=["view a"]), self.T(4, actions=["view a"] * 10 + ["submit"]), self.T(5, actions=["edit tests/t.py", "submit"]), self.T(6, actions=["run sys.exit(0)", "submit"]), self.T(7, actions=["edit conftest.py", "submit"]), self.T(8, prompt="mail x@y.com")] ids = lambda xs: [t.id for t in xs] self.assertEqual(ids(q.gate_consent(ts)), [0, 2, 3, 4, 5, 6, 7, 8]) self.assertEqual(ids(q.gate_resolved(ts)), [0, 1, 3, 4, 5, 6, 7, 8]) self.assertEqual(ids(q.gate_complete(ts)), [0, 1, 2, 4, 5, 6, 7, 8]) self.assertEqual(ids(q.gate_loops(ts)), [0, 1, 2, 3, 5, 6, 7, 8]) self.assertEqual(ids(q.gate_loops(ts, max_run=11)), list(range(9))) self.assertEqual(ids(q.gate_tamper(ts)), [0, 1, 2, 3, 4, 8]) self.assertEqual(ids(q.gate_pii(ts)), [0, 1, 2, 3, 4, 5, 6, 7]) self.assertEqual(q.gate_complete([self.T(0, actions=[])]), []) def test_gate_leak(self): ev = [q.make_prompt("libx", "f", "breaks.", random.Random(5))] near = q._reword(ev[0], random.Random(1)) ts = [self.T(0, prompt=near), self.T(1, prompt=ev[0]), self.T(2, prompt=q.make_prompt("liby", "g", "stalls.", random.Random(9)))] self.assertEqual([t.id for t in q.gate_leak(ts, eval_prompts=ev)], [2]) self.assertEqual([t.id for t in q.gate_leak(ts, eval_prompts=[])], [0, 1, 2]) self.assertEqual([t.id for t in q.gate_leak(ts, eval_prompts=ev, leak_threshold=1.0)], [0, 2]) def test_gate_leak_ignores_copies_among_traces(self): p = q.make_prompt("libx", "f", "breaks.", random.Random(5)) ts = [self.T(0, "a", p), self.T(1, "b", p)] self.assertEqual(len(q.gate_leak(ts, eval_prompts=[])), 2) def test_gate_dedupe_tasks(self): p = q.make_prompt("libx", "f", "breaks.", random.Random(5)) p2 = q._reword(p, random.Random(1)) q3 = q.make_prompt("liby", "g", "stalls.", random.Random(9)) ts = [self.T(0, "a", p), self.T(1, "a", p), self.T(2, "b", p2), self.T(3, "c", q3), self.T(4, "b", p2)] self.assertEqual([t.id for t in q.gate_dedupe_tasks(ts)], [0, 1, 3]) self.assertEqual([t.id for t in q.gate_dedupe_tasks(ts, dup_threshold=0.99)], [0, 1, 2, 3, 4]) def test_gate_dedupe_uses_the_first_trace_by_id(self): p = q.make_prompt("libx", "f", "breaks.", random.Random(5)) p2 = q._reword(p, random.Random(1)) ts = [self.T(5, "b", p2), self.T(1, "a", p)] self.assertEqual([t.id for t in q.gate_dedupe_tasks(ts)], [1]) def test_gate_not_trivial(self): ts = [self.T(0, "a"), self.T(1, "b"), self.T(2, "c")] rates = {"a": 1.0, "b": 0.9, "c": 0.5} self.assertEqual([t.id for t in q.gate_not_trivial(ts, rates=rates)], [1, 2]) self.assertEqual([t.id for t in q.gate_not_trivial(ts, rates=rates, max_pass_rate=0.95)], [1, 2]) self.assertEqual([t.id for t in q.gate_not_trivial(ts, rates=rates, max_pass_rate=1.0)], [0, 1, 2]) self.assertEqual(len(q.gate_not_trivial(ts, rates=None)), 3) def test_gate_cap(self): ts = [self.T(i, "a" if i < 5 else "b") for i in (7, 1, 2, 3, 4, 5, 6)] self.assertEqual([t.id for t in q.gate_cap(ts, cap=3)], [1, 2, 3, 5, 6, 7]) self.assertEqual(len(q.gate_cap(ts, cap=1)), 2) def test_task_pass_rates(self): ts = [self.T(0, "a"), self.T(1, "a", passed=False), self.T(2, "b")] self.assertEqual(q.task_pass_rates(ts), {"a": 0.5, "b": 1.0}) def test_run_pipeline_funnel(self): ev = q.make_eval_prompts(12, 1) pool = q.make_trace_pool(100, 0, ev) r = q.run_pipeline(pool, ev) f = r["funnel"] self.assertEqual(f[0], ("pool", len(pool), len(pool))) self.assertEqual([g for g, _, _ in f[1:]], [n for n, _ in q.GATES]) for (_g, n_in, n_out), (_h, m_in, _m) in zip(f[1:], f[2:]): self.assertLessEqual(n_out, n_in) for i in range(1, len(f) - 1): self.assertEqual(f[i][2], f[i + 1][1]) self.assertEqual(f[-1][2], len(r["kept"])) self.assertTrue(all(t.passed and t.consent_train for t in r["kept"])) task_counts = {} for t in r["kept"]: task_counts[t.task_id] = task_counts.get(t.task_id, 0) + 1 self.assertLessEqual(max(task_counts.values()), 3) def test_run_pipeline_skip_and_cap_option(self): ev = q.make_eval_prompts(12, 1) pool = q.make_trace_pool(100, 0, ev) full = q.run_pipeline(pool, ev) skip = q.run_pipeline(pool, ev, skip=("tamper",)) self.assertNotIn("tamper", [g for g, _, _ in skip["funnel"]]) self.assertGreater(len(skip["kept"]), len(full["kept"])) self.assertTrue(any(t.defect == "tamper" for t in skip["kept"])) capped = q.run_pipeline(pool, ev, cap=1) tasks = [t.task_id for t in capped["kept"]] self.assertEqual(len(tasks), len(set(tasks))) def test_gates_catch_what_they_target(self): ev = q.make_eval_prompts(12, 1) pool = q.make_trace_pool(150, 0, ev) rep = q.defect_report(pool, q.run_pipeline(pool, ev)["kept"]) for d in ("leak", "loop", "pii", "tamper", "truncated"): self.assertGreater(rep[d]["in_pool"], 10) self.assertEqual(rep[d]["kept"], 0) self.assertGreater(rep["weak_tests"]["kept"], 0) self.assertLess(rep["clean_passing"]["kept"], rep["clean_passing"]["in_pool"]) def test_defect_report_counts(self): ts = [self.T(0, defect="loop"), self.T(1, defect="loop"), self.T(2), self.T(3, passed=False), self.T(4, consent=False)] rep = q.defect_report(ts, [ts[0], ts[2]]) self.assertEqual(rep["loop"], {"in_pool": 2, "kept": 1}) self.assertEqual(rep["clean_passing"], {"in_pool": 1, "kept": 1}) self.assertEqual(sorted(rep), ["clean_passing", "loop"]) def test_wilson_interval(self): lo, hi = q.wilson_interval(0, 40) self.assertEqual(lo, 0.0) self.assertAlmostEqual(hi, 0.0881, places=3) lo, hi = q.wilson_interval(20, 40) self.assertAlmostEqual(lo, 0.3519, places=3) self.assertAlmostEqual(hi, 0.6481, places=3) self.assertEqual(q.wilson_interval(0, 0), (0.0, 1.0)) self.assertEqual(q.wilson_interval(40, 40)[1], 1.0) def test_audit_sample_covers_repos(self): ts = [self.T(i, repo="r%d" % (i % 3)) for i in range(30)] s = q.audit_sample(ts, 6, seed=0) self.assertEqual(len(s), 6) self.assertEqual({t.repo for t in s}, {"r0", "r1", "r2"}) self.assertEqual([t.id for t in s], sorted(t.id for t in s)) self.assertEqual(s, q.audit_sample(ts, 6, seed=0)) self.assertNotEqual(s, q.audit_sample(ts, 6, seed=1)) self.assertEqual(len(q.audit_sample(ts, 100)), 30) self.assertEqual(len({t.id for t in q.audit_sample(ts, 100)}), 30) self.assertEqual(q.audit_sample([], 5), []) def test_dataset_card(self): ts = [self.T(0, "a", repo="r1"), self.T(1, "a", repo="r2"), self.T(2, "b", repo="r1")] c = q.dataset_card(ts, [("pool", 5, 5), ("consent", 5, 3)], ["e1", "e2"]) self.assertEqual(c["traces"], 3) self.assertEqual(c["tasks"], 2) self.assertEqual(c["repos"], ["r1", "r2"]) self.assertEqual(c["gates"], ["consent"]) self.assertEqual(c["eval_prompts_checked"], 2) self.assertTrue(c["known_gaps"]) class TestDemo(unittest.TestCase): def test_main_prints_nine_sections(self): import io import contextlib buf = io.StringIO() old = q.tiny_benchmark_average q.tiny_benchmark_average = lambda seeds, **kw: old(2, n_train=20, n_test=10, n_items=40, k=8, epochs=20, dims=kw["dims"]) try: with contextlib.redirect_stdout(buf): q.main() finally: q.tiny_benchmark_average = old out = buf.getvalue() for n in range(1, 10): self.assertIn("== %d. " % n, out) self.assertIn("Artstein and Poesio", out) class TestAddedAfterMutationRun(unittest.TestCase): """Tests written after a mutation run showed that earlier tests missed some changes. Some pin exact values from the made-up worlds (regression pins), and some check the meaning of the code.""" # ---- MinHash banding, importance weights, Gumbel draw def _pair_texts(self, n_pairs, size, inter): texts = {} for p in range(n_pairs): common = [f"c{p}w{j}" for j in range(inter)] texts[2 * p] = " ".join(common + [f"a{p}w{j}" for j in range(size - inter)]) texts[2 * p + 1] = " ".join(common + [f"b{p}w{j}" for j in range(size - inter)]) return texts def test_banding_rate_matches_the_formula(self): n = 400 # word sets of size 30 sharing 20 words: Jaccard 20/40 = 0.5, expected rate 1 - (1 - 0.5^4)^16 = 0.644 c = {a for a, b in q.lsh_candidates(self._pair_texts(n, 30, 20), k=1, seed=3) if b == a + 1 and a % 2 == 0} self.assertGreater(len(c) / n, 0.54) self.assertLess(len(c) / n, 0.74) # Jaccard 4/40 = 0.1, expected rate 0.0016 c = {a for a, b in q.lsh_candidates(self._pair_texts(n, 22, 4), k=1, seed=3) if b == a + 1 and a % 2 == 0} self.assertLess(len(c) / n, 0.03) def test_dsir_weight_counts_repeated_features(self): raw = {0: "alpha beta alpha alpha", 1: "gamma delta", 2: "alpha gamma"} target = ["alpha alpha beta", "alpha beta"] w = q.dsir_log_weights(raw, target, buckets=64) pt, pr = q.bucket_probs(target, 64), q.bucket_probs(list(raw.values()), 64) for i, t in raw.items(): expect = sum(c * math.log(pt[b] / pr[b]) for b, c in q.hashed_features(t, 64).items()) self.assertAlmostEqual(w[i], expect) def test_gumbel_top_one_follows_softmax_of_log_weights(self): lw = {0: 0.0, 1: 1.0, 2: 2.0} n = 3000 wins = [0, 0, 0] for s in range(n): wins[q.gumbel_top_k(lw, 1, seed=s)[0]] += 1 z = sum(math.exp(v) for v in lw.values()) for i in range(3): self.assertAlmostEqual(wins[i] / n, math.exp(lw[i]) / z, delta=0.04) def test_gumbel_draw_keeps_a_low_weight_item_rare_but_not_never(self): lw = {0: 0.0, 1: 0.0, 2: 0.0, 3: -3.0} n = 6000 hits = sum(q.gumbel_top_k(lw, 1, seed=s)[0] == 3 for s in range(n)) self.assertAlmostEqual(hits / n, math.exp(-3) / (3 + math.exp(-3)), delta=0.008) def test_drift_demo_uses_exactly_n_grade_graded_outputs(self): d = q.criteria_drift_demo(n_grade=8, seed=1) self.assertEqual(d["picked_early"], ["no_hashtag", "proper>=1.0", "words<=60"]) self.assertEqual((d["early_set_vs_strict_grades"], d["early_set_vs_relaxed_grades"], d["regraded_set_vs_relaxed_grades"]), (0.783, 0.491, 0.683)) d = q.criteria_drift_demo(n_grade=10, seed=2) self.assertEqual(d["picked_after_drift"], ["no_hashtag", "proper>=0.5", "words<=60"]) self.assertEqual((d["early_set_vs_strict_grades"], d["early_set_vs_relaxed_grades"], d["regraded_set_vs_relaxed_grades"]), (0.656, 0.388, 0.863)) # ---- the toy world and the gradient-similarity demo def test_accuracy_counts_a_tie_as_class_one(self): self.assertEqual(q.accuracy([0.0, 0.0], [([1.0, 1.0], 1), ([2.0, 0.0], 1)]), 1.0) self.assertEqual(q.accuracy([0.0, 0.0], [([1.0, 1.0], 0)]), 0.0) def test_less_world_noise_and_big_norms_only_in_the_pool(self): w = q.make_less_world(n_per_task=20, big_frac=1.0, big_scale=100.0, label_noise=1.0, n_support=30, n_test=30) self.assertTrue(all(e["big"] for e in w["pool"])) self.assertTrue(all(q.norm(e["x"]) > 50 for e in w["pool"])) for x, y in w["support"] + w["test"]: self.assertLess(q.norm(x), 30) self.assertEqual(y, 1 if q.dot(w["dirs"]["A"], x) > 0 else 0) for e in w["pool"]: if e["task"] == "A": self.assertEqual(e["y"], 0 if q.dot(w["dirs"]["A"], e["x"]) > 0 else 1) def test_less_demo_pinned_values_for_seed_zero(self): r = q.less_demo(seed=0) self.assertEqual({k: v["acc_A"] for k, v in r.items()}, {"random": 0.493, "less_cosine": 0.737, "dot_product": 0.713, "full_pool": 0.437}) self.assertEqual({k: v["from_A"] for k, v in r.items()}, {"random": 30, "less_cosine": 56, "dot_product": 56, "full_pool": 150}) # ---- centroid classifier, label noise, data maps def test_centroid_probs_uses_tau_squared_and_is_stable(self): p = q.centroid_probs([[0.0, 0.0], [2.0, 0.0]], [0, 1], [[0.0, 0.0]], 2, tau=2.0)[0] self.assertAlmostEqual(p[0], 1 / (1 + math.exp(-0.5))) far = q.centroid_probs([[0.0, 0.0], [1000.0, 0.0]], [0, 1], [[0.0, 0.0]], 2, tau=1.0)[0] self.assertAlmostEqual(far[0], 1.0) self.assertAlmostEqual(sum(far), 1.0) def test_centroid_probs_gives_a_missing_class_the_origin(self): p = q.centroid_probs([[3.0, 0.0], [3.0, 0.0]], [0, 0], [[0.0, 0.0]], 2, tau=1.0)[0] self.assertGreater(p[1], 0.9) def test_label_noise_demo_and_study_pinned_values(self): r = q.label_noise_demo(seed=0) self.assertEqual((r["flipped"], r["cl_flagged"], r["cl_precision"], r["base_precision"], r["base_recall"]), (34, 36, 0.833, 0.833, 0.882)) s = q.label_noise_study(seeds=3) self.assertEqual((s["flipped"], s["cl_precision"], s["base_precision"], s["base_recall"]), (32.667, 0.812, 0.783, 0.848)) def test_baseline_flags_as_many_as_confident_learning(self): xs, ys = q.make_blobs(60, 3, 2, 3.0, 0, [0.6, 0.9, 2.2]) noisy, flipped = q.flip_labels(ys, 0.1, 3, 1) probs = q.cross_val_probs(xs, noisy, 3, seed=2) cl = q.find_label_issues(probs, noisy, 3) r = q.label_noise_demo(n_per_class=60, seed=0) # precision times flagged is the number of true hits, which is a whole number hits = r["base_precision"] * r["cl_flagged"] self.assertAlmostEqual(hits, round(hits), delta=0.05) self.assertEqual(len(q.self_confidence_rank(probs, noisy)[:len(cl)]), len(cl)) def test_history_correctness_agrees_with_the_given_label_probability(self): rng = random.Random(5) data = [([rng.gauss(0, 1), rng.gauss(0, 1), 1.0], rng.randint(0, 1)) for _ in range(40)] ph, ch = q.train_logistic_history(data, epochs=6, lr=0.1, seed=1) for i in range(40): for e in range(6): self.assertEqual(ch[i][e], ph[i][e] > 0.5) def test_easy_region_never_overlaps_the_hard_region(self): stats = {i: {"variability": v, "confidence": c} for i, (v, c) in enumerate([(0.9, 0.5), (0.8, 0.4), (0.7, 0.6), (0.1, 0.1), (0.2, 0.2), (0.3, 0.9)])} reg = q.data_map_regions(stats, fraction=0.5) self.assertEqual(reg["ambiguous"], [0, 1, 2]) self.assertEqual(reg["hard"], [3, 4, 5]) self.assertEqual(reg["easy"], []) # ---- agreement and rank correlation def test_tau_b_with_ties_on_both_sides(self): # pairs: four concordant, one tied only in x, one tied only in y, so tau = 4 / sqrt(5 * 5) self.assertAlmostEqual(q.kendall_tau_b([1, 1, 2, 3], [1, 2, 2, 3]), 0.8) def test_email_pattern_accepts_a_plus_or_dash_before_the_at_sign(self): self.assertTrue(q.has_pii("write to jo+@example.com")) self.assertTrue(q.has_pii("write to jo-@example.com")) def test_alpha_weights_units_with_three_ratings(self): # by hand: o00 = 1, o01 = o10 = 2, n = 5, Do = 4/5, De = 12/20, alpha = 1 - 0.8 / 0.6 self.assertAlmostEqual(q.krippendorff_alpha([[0, 0, 1], [0, 1]]), -1 / 3) # ---- the IRT fit and the tiny-benchmark recipes def test_irt_world_in_two_dimensions_loads_items_on_both_skills(self): w = q.make_irt_world(5, 60, seed=1, dims=2) self.assertGreater(sum(abs(a[0] - a[1]) > 1e-6 for a in w["alpha"]), 40) self.assertTrue(all(a[0] >= 0 and a[1] >= 0 for a in w["alpha"])) def test_fit_starts_from_accuracy_logits(self): resp = [[1, 0, 1, 1], [0, 0, 1, 0], [1, 1, 1, 0]] th, al, be = q.fit_irt(resp, "2pl", epochs=0) for m, row in enumerate(resp): self.assertAlmostEqual(th[m], q.logit(q.mean(row) * 0.98 + 0.01) / 1.5) for i in range(4): acc = q.mean(resp[m][i] for m in range(3)) self.assertAlmostEqual(be[i], -q.logit(acc * 0.98 + 0.01)) self.assertEqual(al, [1.0] * 4) def _penalised_gradients(self, resp, th, al, be): M, N = len(resp), len(resp[0]) gt, ga, gb = [0.0] * M, [0.0] * N, [0.0] * N for m in range(M): for i in range(N): e = resp[m][i] - q.sigmoid(al[i] * th[m] - be[i]) gt[m] += al[i] * e ga[i] += th[m] * e gb[i] -= e return ([(gt[m] - th[m]) / N for m in range(M)], [(ga[i] - (al[i] - 1.0)) / M for i in range(N)], [(gb[i] - 0.1 * be[i]) / M for i in range(N)]) def test_fit_ends_where_the_penalised_gradient_is_zero(self): w = q.make_irt_world(30, 40, seed=2) th, al, be = q.fit_irt(w["resp"], "2pl", epochs=400) self.assertTrue(all(0.05 <= a <= 5.0 for a in al)) gt, ga, gb = self._penalised_gradients(w["resp"], [t for t in th], al, be) self.assertLess(max(abs(g) for g in gt), 2e-3) inside = [g for g, a in zip(ga, al) if 0.05 < a < 5.0] self.assertGreater(len(inside), 30) self.assertLess(max(abs(g) for g in inside), 2e-3) self.assertLess(max(abs(g) for g in gb), 2e-3) self.assertGreater(max(al), 1.3) def test_fit_theta_is_a_map_estimate(self): alpha, beta = [0.5, 2.0, 1.4, 0.8], [0.3, -0.5, 1.0, 0.0] items, y = [0, 1, 2, 3], {0: 1, 1: 0, 2: 1, 3: 1} t = q.fit_theta(alpha, beta, items, y) g = -t + sum(alpha[i] * (y[i] - q.sigmoid(alpha[i] * t - beta[i])) for i in items) self.assertLess(abs(g), 1e-10) def test_stratified_selection_takes_evenly_spaced_ranks(self): acc = [0.5, 0.1, 0.9, 0.3, 0.7, 0.2, 0.8, 0.4, 0.6, 0.0] self.assertEqual(q.select_stratified(acc, 3), [0, 1, 6]) def test_irt_grid_breaks_ties_toward_the_lower_id(self): self.assertEqual(q.select_irt_grid([1.0] * 4, [0.0] * 4, [0.0, 0.0], 2), [0, 1]) def test_anchor_points_on_a_small_response_matrix(self): cols = [(1, 1, 0, 0), (1, 1, 0, 0), (0, 0, 1, 1), (0, 0, 1, 1), (1, 0, 1, 0), (1, 1, 1, 1)] resp = [[cols[i][m] for i in range(6)] for m in range(4)] anchors, w = q.select_anchors(resp, 3) self.assertEqual(anchors, [0, 2, 4]) self.assertAlmostEqual(w[0], 0.5) self.assertAlmostEqual(w[1], 1 / 3) self.assertAlmostEqual(w[2], 1 / 6) def test_tiny_benchmark_study_pinned_values_for_seed_zero(self): r = q.tiny_benchmark_study(seed=0) taus = {k: v["tau"] for k, v in r.items()} self.assertEqual(taus, {"random, mean": 0.857, "stratified by difficulty, mean": 0.739, "anchor points, weighted mean": 0.608, "random, 2PL ability": 0.829, "IRT-chosen, 2PL ability": 0.785, "IRT-chosen, 1PL ability": 0.776}) self.assertEqual(r["IRT-chosen, 1PL ability"]["mae"], 0.0749) # ---- EvalGen part def test_drift_demo_and_sampling_study_pinned_values(self): d = q.criteria_drift_demo() self.assertEqual((d["picked_early"], d["picked_after_drift"]), (["no_hashtag", "proper>=0.6", "words<=50"], ["proper>=0.6", "words<=50"])) self.assertEqual((d["early_set_vs_strict_grades"], d["early_set_vs_relaxed_grades"], d["regraded_set_vs_relaxed_grades"]), (0.777, 0.627, 0.744)) s = q.sampling_study(worlds=3) self.assertEqual({k: v["mean"] for k, v in s.items()}, {"random": 0.815, "highest": 0.556, "lowest": 0.672, "alternating": 0.601}) # ---- trace pool and gates def test_prompt_has_the_worked_example_and_fixed_numbers_without_rng(self): p = q.make_prompt("r", "f", "s.") self.assertRegex(p, r"Example: f of \d{3} and \d{3} gives \d{3} but should give \d{3}\.") r = random.Random(0) nums = [r.randint(100, 999) for _ in range(4)] self.assertIn(f"f of {nums[0]} and {nums[1]} gives {nums[2]} but should give {nums[3]}.", p) self.assertEqual(p, q.make_prompt("r", "f", "s.")) def test_reword_changes_one_middle_word(self): base = q.make_prompt("libalpha", "parse_date", "returns an empty list for repeated keys.", random.Random(1)) w0 = base.split() for s in range(60): w1 = q._reword(base, random.Random(s)).split() diff = [i for i in range(len(w0)) if w0[i] != w1[i]] self.assertEqual(len(diff), 1) self.assertTrue(3 <= diff[0] < len(w0) - 6) def test_duplicate_prompts_are_redrawn(self): calls = [] def fake(repo, fn, symptom, rng=None): calls.append(1) tail = " ".join(f"w{j}" for j in range(30)) return f"same prompt {tail}" if len(calls) <= 2 else f"unique prompt number {len(calls)} {tail}" with mock.patch.object(q, "make_prompt", fake): ev = q.make_eval_prompts(3, seed=0) self.assertEqual(len(set(ev)), 3) calls.clear() with mock.patch.object(q, "make_prompt", fake): shared = "same prompt " + " ".join(f"w{j}" for j in range(30)) pool = q.make_trace_pool(n_tasks=6, seed=0, eval_prompts=[shared]) by_task = {} for t in pool: by_task.setdefault(t.task_id, t.prompt.replace(q.EMAIL_TAIL, "")) self.assertEqual(len(by_task), 6) self.assertEqual(len(set(by_task.values())), 6) self.assertNotIn(shared, by_task.values()) def _pool_tasks(self): ev = q.make_eval_prompts(12, 1) pool = q.make_trace_pool(n_tasks=600, seed=0, eval_prompts=ev) kinds, prompts = {}, {} for t in pool: prompts.setdefault(t.task_id, t.prompt.replace(q.EMAIL_TAIL, "")) kinds.setdefault(t.task_id, set()).add(t.defect) return ev, pool, kinds, prompts def test_near_duplicate_tasks_copy_a_plain_task_with_one_word_changed(self): ev, pool, kinds, prompts = self._pool_tasks() plain = [prompts[k] for k, v in kinds.items() if not (v & {"near_dup", "leak"})] dups = [prompts[k] for k, v in kinds.items() if "near_dup" in v] self.assertGreater(len(dups), 20) for p in dups: best = max(q.jaccard(q.shingles(p), q.shingles(x)) for x in plain) self.assertGreaterEqual(best, 0.8) self.assertLess(best, 1.0) def test_leak_tasks_are_reworded_eval_prompts(self): ev, pool, kinds, prompts = self._pool_tasks() leaks = [prompts[k] for k, v in kinds.items() if "leak" in v] self.assertGreater(len(leaks), 10) for p in leaks: self.assertNotIn(p, ev) self.assertGreaterEqual(max(q.jaccard(q.shingles(p), q.shingles(e)) for e in ev), 0.8) def test_pool_pass_rates_and_loop_outcomes(self): pool = q.make_trace_pool(n_tasks=300, seed=0) clean = [t for t in pool if not t.defect] self.assertGreater(q.mean(1.0 if t.passed else 0.0 for t in clean), 0.5) loops = [t.passed for t in pool if t.defect == "loop"] self.assertGreater(len(loops), 20) self.assertTrue(any(loops) and not all(loops)) def test_task_dedupe_uses_the_first_trace_by_id(self): base = q.make_prompt("libalpha", "parse_date", "returns an empty list for repeated keys.", random.Random(1)) near = q._reword(base, random.Random(2)) other = "something entirely different about a very other topic with many more words than before to avoid overlap" def tr(i, task, prompt): return q.Trace(i, task, "libalpha", prompt, ["view a.py", "submit"], True) a, b1, b2 = tr(0, "t1", base), tr(1, "t2", near), tr(2, "t2", other) for order in ([a, b1, b2], [b2, b1, a]): kept = q.gate_dedupe_tasks(order) self.assertEqual(sorted(t.id for t in kept), [0]) def test_pipeline_computes_pass_rates_before_any_gate(self): def tr(i, task, passed): return q.Trace(i, task, "libalpha", f"prompt for {task} number {i}", ["view a.py", "submit"], passed) traces = [tr(i, "easy", True) for i in range(5)] + [tr(5 + i, "mid", i < 2) for i in range(5)] kept = q.run_pipeline(traces)["kept"] self.assertEqual({t.task_id for t in kept}, {"mid"}) self.assertEqual(len(kept), 2) if __name__ == "__main__": unittest.main()