All courses · Course overview · Lesson sequence
Skip to lesson textproject/tests/test_methods.py
from dataclasses import replace
import json
import math
from pathlib import Path
import subprocess
import sys
import unittest
from methods import (attention_arithmetic, cosine, coverage, decode_bigram, demo,
fixed_vector_ranking, paragraph_chunks, reciprocal_rank_fusion,
renewal_rule, render_template, review_claims, select_context,
softmax, source_documents, train_bigram, word_windows, constraint_rerank)
from retrieval import Document, Hit, Index
def document(text, doc_id="a", source="synthetic://a", version="1"):
return Document(doc_id, doc_id, text, source, version, "current", "public")
class LanguageModelTests(unittest.TestCase):
def test_training_count_distribution_and_greedy_trace(self):
counts = train_bigram(["desk opens weekdays", "desk opens weekdays", "desk closes holidays"])
result = decode_bigram(counts, "desk")
self.assertEqual(result["tokens"], ["opens", "weekdays"])
self.assertEqual(result["stop_reason"], "end_token")
self.assertAlmostEqual(result["trace"][0]["distribution"]["opens"], 2 / 3)
self.assertEqual(decode_bigram(counts, "unknown")["stop_reason"], "unseen_history")
def test_sampling_replays_seed_and_loop_is_bounded(self):
counts = train_bigram(["desk opens weekdays", "desk closes holidays"])
self.assertEqual(decode_bigram(counts, "desk", temperature=2, seed=9),
decode_bigram(counts, "desk", temperature=2, seed=9))
cycle = {"a": {"a": 1}}
self.assertEqual(decode_bigram(cycle, "a", max_steps=3)["stop_reason"], "step_budget")
self.assertEqual(len(decode_bigram(cycle, "a", max_steps=3)["tokens"]), 3)
def test_temperature_and_stable_softmax(self):
cool = softmax([math.log(2), 0], 0.5)
warm = softmax([math.log(2), 0], 2)
self.assertAlmostEqual(cool[0], 0.8)
self.assertLess(warm[0], 2 / 3)
self.assertAlmostEqual(sum(softmax([10000, 10000])), 1)
for bad in (0, -1, math.inf, True):
with self.assertRaises(ValueError):
decode_bigram({}, temperature=bad)
def test_attention_is_a_weighted_average(self):
result = attention_arithmetic(1, [0, math.log(3)], [10, 20])
self.assertAlmostEqual(result["weights"][0], 0.25)
self.assertAlmostEqual(result["weights"][1], 0.75)
self.assertAlmostEqual(result["weighted_value"], 17.5)
with self.assertRaises(ValueError):
attention_arithmetic(1, [1], [1, 2])
class ChunkTests(unittest.TestCase):
def test_word_windows_have_exact_offsets_and_overlap(self):
source = document("one two\nthree four five six seven")
chunks = word_windows(source, 4, 1)
self.assertEqual([chunk.text for chunk in chunks], ["one two\nthree four", "four five six seven"])
for chunk in chunks:
self.assertEqual(source.text[chunk.start:chunk.end], chunk.text)
self.assertEqual(chunk.parent_id, source.id)
self.assertEqual(chunk.version, source.version)
def test_paragraphs_do_not_silently_split_an_oversized_condition(self):
source = document("A renewal lasts seven days if no hold exists.\n\nAsk early.")
chunks = paragraph_chunks(source, 20)
self.assertEqual(len(chunks), 2)
self.assertTrue(chunks[0].over_budget)
self.assertEqual(chunks[0].text, "A renewal lasts seven days if no hold exists.")
self.assertFalse(chunks[1].over_budget)
for chunk in chunks:
self.assertEqual(source.text[chunk.start:chunk.end], chunk.text)
def test_empty_input_and_impossible_window_are_handled(self):
self.assertEqual(word_windows(document(" "), 4, 1), [])
self.assertEqual(paragraph_chunks(document(" "), 10), [])
for size, overlap in ((0, 0), (4, 4), (4, -1), (4, True)):
with self.assertRaises(ValueError):
word_windows(document("red blue"), size, overlap)
class CandidateTests(unittest.TestCase):
def test_vector_geometry_is_calculated_not_text_encoded(self):
self.assertAlmostEqual(cosine([1, 1], [1, 0]), 1 / math.sqrt(2))
self.assertEqual(cosine([0, 0], [1, 0]), 0)
self.assertEqual(fixed_vector_ranking([0, 1], {"a": [1, 0], "b": [0, 1]})[0][0], "b")
with self.assertRaises(ValueError):
cosine([1], [1, 2])
with self.assertRaises(ValueError):
cosine([math.inf], [1])
def test_rank_fusion_matches_hand_arithmetic_and_dedup_contract(self):
result = reciprocal_rank_fusion([["a", "b"], ["c", "a"]])
self.assertEqual([item[0] for item in result], ["a", "c", "b"])
self.assertAlmostEqual(result[0][1], 1/61 + 1/62)
with self.assertRaises(ValueError):
reciprocal_rank_fusion([["a", "a"]])
def test_exact_constraint_filter_neither_invents_candidates_nor_understands_synonyms(self):
hit = Hit(document("Books are available."), 0.9)
self.assertEqual(constraint_rerank([hit], ["laptop"])["kept"], [])
self.assertEqual(constraint_rerank([hit], ["books"])["kept"], [hit])
self.assertEqual(constraint_rerank([hit], ["book"])["kept"], [])
class ContextAndAnswerTests(unittest.TestCase):
def test_context_deduplicates_and_counts_whole_quote_budget(self):
first = Hit(document("red blue"), 0.9)
duplicate = Hit(document("red blue", "b", "synthetic://b"), 0.8)
smaller = Hit(document("red", "c", "synthetic://c"), 0.7)
result = select_context([first, first, duplicate, smaller], 8)
self.assertEqual([entry["id"] for entry in result["packet"]["evidence"]], ["a"])
self.assertEqual(result["duplicate_text_ids"], ["b"])
self.assertEqual(result["budget_skipped_ids"], ["c"])
def test_conflicting_versions_block_both_instead_of_guessing_freshness(self):
a = document("open at nine", "a", "synthetic://hours", "1")
b = document("open at ten", "b", "synthetic://hours", "2")
result = select_context([Hit(a, 0.9), Hit(b, 0.8)])
self.assertEqual(result["packet"]["evidence"], [])
self.assertEqual(result["conflicting_source_versions"], {"synthetic://hours": ["1", "2"]})
def test_same_id_with_changed_text_and_ineligible_context_are_rejected(self):
a = document("red")
with self.assertRaises(ValueError):
select_context([Hit(a, 1), Hit(replace(a, text="blue"), 0.9)])
with self.assertRaises(ValueError):
select_context([Hit(replace(a, status="archived"), 1)])
def test_template_is_explicitly_authored_and_bound_to_exact_source(self):
sources = source_documents()
rule = renewal_rule(sources)
packet = select_context(Index(sources).search("renewed hold"))["packet"]
answer = render_template("q-renew", packet, rule)
self.assertEqual(answer["status"], "template_rendered")
self.assertFalse(answer["model_used"])
self.assertEqual(render_template("unknown", packet, rule)["status"], "abstain")
for field, change in (("quote", "changed"), ("version", "2"), ("source", "synthetic://elsewhere")):
changed = json.loads(json.dumps(packet))
changed["evidence"][0][field] = change
self.assertEqual(render_template("q-renew", changed, rule)["status"], "abstain")
self.assertEqual(render_template("q-renew", {"evidence": []}, rule)["status"], "abstain")
def test_citation_existence_and_manual_support_can_disagree(self):
claims = [{"id": "x", "text": "Wrong claim", "citation_ids": ["a"]}]
result = review_claims(claims, {"a"}, {"x": False}, {"x"})
self.assertEqual(result["citation_ids_valid_count"], 1)
self.assertEqual(result["manual_supported_claim_fraction"], 0)
self.assertEqual(result["required_supported_claim_coverage"], 0)
with self.assertRaises(ValueError):
review_claims(claims, {"a"}, {}, {"x"})
self.assertIsNone(review_claims([], set(), {}, set())["manual_supported_claim_fraction"])
self.assertEqual(coverage(["a"], ["a", "b"]), 0.5)
def test_every_taught_demo_runs_without_a_service(self):
for name in ("decoding", "chunking", "lexical", "hybrid", "context", "generation", "evaluation", "failures"):
with self.subTest(name=name):
json.dumps(demo(name))
def test_demo_cli_works_as_documented(self):
path = Path(__file__).resolve().parents[1] / "methods.py"
result = subprocess.run([sys.executable, "-B", str(path), "hybrid"], cwd="/tmp",
capture_output=True, text=True, check=True)
self.assertEqual(json.loads(result.stdout)["kind"], "manually_supplied_vector_simulation_no_encoder")
if __name__ == "__main__":
unittest.main()