From f739b5b4c8886ed7297c398137669cfa2639b388 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sun, 14 Jun 2026 19:31:16 -0500 Subject: [PATCH] test(coverage): add 138 unit tests for least-covered modules (#78) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four test files covering the modules with zero or near-zero test coverage: - scripts/qc-flag/test_lean_qc_flagger.py (55 tests) Covers all pure helpers, code-line detection, QCIssue/FileResult classes, and all 5 check functions (structural, naming, Q16, proof quality, deps). - 4-Infrastructure/shim/test_braid_diat_codec.py (22 tests) Covers Q0_2 encode/decode, ChiralityDIAT roundtrips and verify_b, MountainPacked dict↔bytes, BraidResidualPacked bracket↔bytes. - 5-Applications/tools-scripts/llm/test_emitter_utils.py (33 tests) Covers sha256_bytes, canonical_json_bytes, repo_relative, safe_slug, timestamps, context bundles, message building, extract_answer, auth guard, and verify_receipt schema rules. - 4-Infrastructure/shim/test_validate_rrc_predictions.py (28 tests) Covers require_path, parse_equation, build_proof_metrics, build_receipt with all domain mappings, and MATH_* constant sets. Build: all 138 tests pass (pytest), py_compile clean Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: Allaun Silverfox --- .../shim/test_braid_diat_codec.py | 224 ++++++++++ .../shim/test_validate_rrc_predictions.py | 258 +++++++++++ .../tools-scripts/llm/test_emitter_utils.py | 299 +++++++++++++ scripts/qc-flag/test_lean_qc_flagger.py | 411 ++++++++++++++++++ 4 files changed, 1192 insertions(+) create mode 100644 4-Infrastructure/shim/test_braid_diat_codec.py create mode 100644 4-Infrastructure/shim/test_validate_rrc_predictions.py create mode 100644 5-Applications/tools-scripts/llm/test_emitter_utils.py create mode 100644 scripts/qc-flag/test_lean_qc_flagger.py diff --git a/4-Infrastructure/shim/test_braid_diat_codec.py b/4-Infrastructure/shim/test_braid_diat_codec.py new file mode 100644 index 00000000..b44b6f60 --- /dev/null +++ b/4-Infrastructure/shim/test_braid_diat_codec.py @@ -0,0 +1,224 @@ +#!/usr/bin/env python3 +"""Unit tests for braid_diat_codec.py — encode/decode roundtrip and layer logic.""" + +import importlib.util +import struct +import sys +import unittest +from pathlib import Path + +MODULE_PATH = Path(__file__).with_name("braid_diat_codec.py") +SPEC = importlib.util.spec_from_file_location("braid_diat_codec", MODULE_PATH) +assert SPEC is not None and SPEC.loader is not None +codec = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = codec +SPEC.loader.exec_module(codec) + + +# --------------------------------------------------------------------------- +# Q0_2 encoding +# --------------------------------------------------------------------------- + +class TestQ02Encoding(unittest.TestCase): + def test_encode_all_states(self): + self.assertEqual(codec.encode_q02(0), 0) + self.assertEqual(codec.encode_q02(16384), 1) + self.assertEqual(codec.encode_q02(32768), 2) + self.assertEqual(codec.encode_q02(49152), 3) + + def test_decode_all_states(self): + self.assertEqual(codec.decode_q02(0), 0) + self.assertEqual(codec.decode_q02(1), 16384) + self.assertEqual(codec.decode_q02(2), 32768) + self.assertEqual(codec.decode_q02(3), 49152) + + def test_roundtrip_all_states(self): + for v in codec.Q0_2_STATES: + self.assertEqual(codec.decode_q02(codec.encode_q02(v)), v) + + def test_decode_masks_to_2_bits(self): + self.assertEqual(codec.decode_q02(4), 0) + self.assertEqual(codec.decode_q02(5), 16384) + + +# --------------------------------------------------------------------------- +# Layer 1: ChiralityDIAT +# --------------------------------------------------------------------------- + +class TestChiralityDIAT(unittest.TestCase): + def test_encode_decode_roundtrip(self): + original = codec.ChiralityDIAT(codec.Chirality.POSITIVE, 5, 100, 200, 42) + raw = original.to_bytes() + self.assertEqual(len(raw), 8) + restored = codec.ChiralityDIAT.from_bytes(raw) + self.assertEqual(restored.chirality, codec.Chirality.POSITIVE) + self.assertEqual(restored.shell, 5) + self.assertEqual(restored.offset_a, 100) + self.assertEqual(restored.offset_b, 200) + self.assertEqual(restored.prod_msb, 42) + + def test_all_chiralities(self): + for ch in codec.Chirality: + obj = codec.ChiralityDIAT(ch, 0, 0, 0, 0) + restored = codec.ChiralityDIAT.from_bytes(obj.to_bytes()) + self.assertEqual(restored.chirality, ch) + + def test_decode_from_n(self): + slot = codec.ChiralityDIAT.decode(codec.Chirality.LEFT, 25) + self.assertIsNotNone(slot) + self.assertEqual(slot.chirality, codec.Chirality.LEFT) + self.assertEqual(slot.shell, 5) + self.assertEqual(slot.offset_a, 0) + self.assertEqual(slot.offset_b, 11) + + def test_decode_n_too_large_returns_none(self): + result = codec.ChiralityDIAT.decode(codec.Chirality.NONE, 0x400000) + self.assertIsNone(result) + + def test_to_n_roundtrip(self): + for n in [0, 1, 4, 9, 16, 25, 100, 1000]: + slot = codec.ChiralityDIAT.decode(codec.Chirality.POSITIVE, n) + self.assertIsNotNone(slot) + self.assertEqual(slot.to_n(), n) + + def test_verify_b(self): + slot = codec.ChiralityDIAT.decode(codec.Chirality.ACHIRAL, 17) + self.assertIsNotNone(slot) + self.assertTrue(slot.verify_b()) + + def test_verify_b_corrupted(self): + slot = codec.ChiralityDIAT.decode(codec.Chirality.ACHIRAL, 17) + slot.offset_b = 999 + self.assertFalse(slot.verify_b()) + + def test_bytes_roundtrip_preserves_all_fields(self): + for n in [0, 7, 50, 255]: + slot = codec.ChiralityDIAT.decode(codec.Chirality.POSITIVE, n) + restored = codec.ChiralityDIAT.from_bytes(slot.to_bytes()) + self.assertEqual(slot.shell, restored.shell) + self.assertEqual(slot.offset_a, restored.offset_a) + self.assertEqual(slot.offset_b, restored.offset_b) + self.assertEqual(slot.prod_msb, restored.prod_msb) + + +# --------------------------------------------------------------------------- +# Layer 2: MountainPacked +# --------------------------------------------------------------------------- + +class TestMountainPacked(unittest.TestCase): + def _sample_mountain_dict(self): + return { + "height": 3, + "apex": [100, 200, 300], + "base": [[10, 20, 30], [40, 50, 60]], + "inner": {}, + } + + def test_from_mountain_roundtrip(self): + m = self._sample_mountain_dict() + packed = codec.MountainPacked.from_mountain(m) + self.assertEqual(packed.height, 3) + self.assertEqual(packed.apex_x, 100) + self.assertEqual(packed.apex_y, 200) + self.assertEqual(packed.apex_z, 300) + self.assertEqual(packed.base_count, 2) + result = packed.to_mountain() + self.assertEqual(result["height"], 3) + self.assertEqual(result["apex"], [100, 200, 300]) + self.assertEqual(result["base"], [[10, 20, 30], [40, 50, 60]]) + + def test_bytes_roundtrip(self): + m = self._sample_mountain_dict() + packed = codec.MountainPacked.from_mountain(m) + raw = packed.to_bytes() + restored = codec.MountainPacked.from_bytes(raw) + self.assertEqual(restored.height, packed.height) + self.assertEqual(restored.apex_x, packed.apex_x) + self.assertEqual(restored.apex_y, packed.apex_y) + self.assertEqual(restored.apex_z, packed.apex_z) + self.assertEqual(restored.base_count, packed.base_count) + self.assertEqual(restored.bases, packed.bases) + + def test_empty_bases(self): + m = {"height": 1, "apex": [0, 0, 0], "base": [], "inner": {}} + packed = codec.MountainPacked.from_mountain(m) + self.assertEqual(packed.base_count, 0) + self.assertEqual(packed.bases, []) + result = packed.to_mountain() + self.assertEqual(result["base"], []) + + def test_negative_coords(self): + m = {"height": 2, "apex": [-100, -200, -300], "base": [[-1, -2, -3]], "inner": {}} + packed = codec.MountainPacked.from_mountain(m) + raw = packed.to_bytes() + restored = codec.MountainPacked.from_bytes(raw) + result = restored.to_mountain() + self.assertEqual(result["apex"], [-100, -200, -300]) + self.assertEqual(result["base"], [[-1, -2, -3]]) + + +# --------------------------------------------------------------------------- +# Layer 3: BraidResidualPacked +# --------------------------------------------------------------------------- + +class TestBraidResidualPacked(unittest.TestCase): + def _sample_bracket(self): + return { + "lower": 0, + "upper": 16384, + "gap": 32768, + "kappa": 49152, + "phi": 0, + "admissible": True, + } + + def test_from_bracket_roundtrip(self): + br = self._sample_bracket() + packed = codec.BraidResidualPacked.from_bracket(br) + result = packed.to_bracket() + self.assertEqual(result, br) + + def test_bytes_roundtrip(self): + br = self._sample_bracket() + packed = codec.BraidResidualPacked.from_bracket(br) + raw = packed.to_bytes() + self.assertEqual(len(raw), 8) + restored = codec.BraidResidualPacked.from_bytes(raw) + self.assertEqual(restored.to_bracket(), br) + + def test_all_q02_states_roundtrip(self): + for v in codec.Q0_2_STATES: + br = {"lower": v, "upper": v, "gap": v, "kappa": v, "phi": v, "admissible": False} + packed = codec.BraidResidualPacked.from_bracket(br) + result = codec.BraidResidualPacked.from_bytes(packed.to_bytes()).to_bracket() + self.assertEqual(result, br) + + def test_admissible_bit(self): + br_true = self._sample_bracket() + br_true["admissible"] = True + br_false = dict(br_true) + br_false["admissible"] = False + packed_t = codec.BraidResidualPacked.from_bracket(br_true) + packed_f = codec.BraidResidualPacked.from_bracket(br_false) + self.assertTrue(codec.BraidResidualPacked.from_bytes(packed_t.to_bytes()).admissible) + self.assertFalse(codec.BraidResidualPacked.from_bytes(packed_f.to_bytes()).admissible) + + +# --------------------------------------------------------------------------- +# Chirality enum +# --------------------------------------------------------------------------- + +class TestChiralityEnum(unittest.TestCase): + def test_values(self): + self.assertEqual(int(codec.Chirality.NONE), 0) + self.assertEqual(int(codec.Chirality.POSITIVE), 1) + self.assertEqual(int(codec.Chirality.LEFT), 2) + self.assertEqual(int(codec.Chirality.ACHIRAL), 3) + + def test_all_fit_in_2_bits(self): + for ch in codec.Chirality: + self.assertLessEqual(int(ch), 3) + + +if __name__ == "__main__": + unittest.main() diff --git a/4-Infrastructure/shim/test_validate_rrc_predictions.py b/4-Infrastructure/shim/test_validate_rrc_predictions.py new file mode 100644 index 00000000..8e3db8ee --- /dev/null +++ b/4-Infrastructure/shim/test_validate_rrc_predictions.py @@ -0,0 +1,258 @@ +#!/usr/bin/env python3 +"""Unit tests for pure functions in validate_rrc_predictions.py. + +The module uses Python 3.12+ f-string syntax, so we extract testable functions +via targeted exec rather than a full module import. +""" + +import math +import os +import re +import sys +import unittest +from collections import Counter +from pathlib import Path + +# --------------------------------------------------------------------------- +# Extract testable functions from source without full module load +# --------------------------------------------------------------------------- + +_SRC = Path(__file__).with_name("validate_rrc_predictions.py").read_text() + +_NAMESPACE: dict = {} +# Execute only the pure-function portion (up to the classify function) +# by extracting individual function blocks. + +exec( + compile( + """ +import json, os, re, subprocess, sys, tempfile, math +from collections import Counter, defaultdict + +""" + + re.search( + r"(def require_path.*?)(?=\nPIST_DECOMPOSE)", + _SRC, + re.DOTALL, + ).group(1) + + "\n" + + re.search( + r"(MATH_OPERATORS\s*=\s*\{.*?\})", + _SRC, + re.DOTALL, + ).group(1) + + "\n" + + re.search( + r"(MATH_VARIABLES\s*=\s*\{.*?\})", + _SRC, + re.DOTALL, + ).group(1) + + "\n" + + re.search( + r"(MATH_CONSTANTS\s*=\s*\{.*?\})", + _SRC, + re.DOTALL, + ).group(1) + + "\n" + + re.search( + r"(def parse_equation.*?)(?=\ndef build_proof_metrics)", + _SRC, + re.DOTALL, + ).group(1) + + "\n" + + re.search( + r"(def build_proof_metrics.*?)(?=\ndef build_receipt)", + _SRC, + re.DOTALL, + ).group(1) + + "\n" + + re.search( + r"(def build_receipt.*?)(?=\ndef classify)", + _SRC, + re.DOTALL, + ).group(1), + "", + "exec", + ), + _NAMESPACE, +) + +require_path = _NAMESPACE["require_path"] +parse_equation = _NAMESPACE["parse_equation"] +build_proof_metrics = _NAMESPACE["build_proof_metrics"] +build_receipt = _NAMESPACE["build_receipt"] +MATH_OPERATORS = _NAMESPACE["MATH_OPERATORS"] +MATH_VARIABLES = _NAMESPACE["MATH_VARIABLES"] +MATH_CONSTANTS = _NAMESPACE["MATH_CONSTANTS"] + + +# --------------------------------------------------------------------------- +# require_path +# --------------------------------------------------------------------------- + +class TestRequirePath(unittest.TestCase): + def test_single_key(self): + self.assertEqual(require_path({"a": 1}, "a"), 1) + + def test_nested_keys(self): + self.assertEqual(require_path({"a": {"b": {"c": 42}}}, "a", "b", "c"), 42) + + def test_missing_key_raises(self): + with self.assertRaises((KeyError, TypeError)): + require_path({"a": 1}, "a", "b") + + def test_missing_top_key_raises(self): + with self.assertRaises(KeyError): + require_path({}, "missing") + + +# --------------------------------------------------------------------------- +# parse_equation +# --------------------------------------------------------------------------- + +class TestParseEquation(unittest.TestCase): + def test_basic_equation(self): + result = parse_equation("cognitive_load_field") + self.assertEqual(result["equation_text"], "cognitive_load_field") + # "cognitive", "load", "field" are all in MATH_OPERATORS + self.assertIn("cognitive", result["operators"]) + + def test_operator_extraction(self): + result = parse_equation("adjusted-overflow-gate") + self.assertIn("adjusted", result["operators"]) + self.assertIn("overflow", result["operators"]) + self.assertIn("gate", result["operators"]) + + def test_constant_extraction(self): + result = parse_equation("phi-delta-mapping") + self.assertIn("phi", result["constants"]) + self.assertIn("delta", result["constants"]) + + def test_normalized_equation_sorted(self): + result = parse_equation("z-a-m") + self.assertEqual(result["normalized_equation"], "a_m_z") + + def test_ast_metrics_present(self): + result = parse_equation("heat-loss-equations") + self.assertIn("node_count", result["ast_metrics"]) + self.assertIn("depth", result["ast_metrics"]) + self.assertIn("branching_factor", result["ast_metrics"]) + self.assertIn("symbol_entropy", result["ast_metrics"]) + + def test_depth_clamped_at_16(self): + long_eq = "-".join(["x"] * 20) + result = parse_equation(long_eq) + self.assertEqual(result["ast_metrics"]["depth"], 16) + + def test_single_part_zero_entropy(self): + result = parse_equation("single") + self.assertEqual(result["ast_metrics"]["symbol_entropy"], 0.0) + + def test_operator_counts(self): + result = parse_equation("adjusted-adjusted-gate-loss") + self.assertEqual(result["operator_counts"]["adjusted"], 2) + self.assertEqual(result["operator_counts"]["gate"], 1) + self.assertEqual(result["operator_counts"]["loss"], 1) + self.assertEqual(result["operator_counts"]["overflow"], 0) + + +# --------------------------------------------------------------------------- +# build_proof_metrics +# --------------------------------------------------------------------------- + +class TestBuildProofMetrics(unittest.TestCase): + def test_cognitive_load_field(self): + m = build_proof_metrics("CognitiveLoadField") + self.assertEqual(m["tactic_count"], 3) + self.assertEqual(m["dependency_count"], 2) + + def test_signal_shaped(self): + m = build_proof_metrics("SignalShapedRouteCompiler") + self.assertEqual(m["tactic_count"], 4) + + def test_projectable_geometry(self): + m = build_proof_metrics("ProjectableGeometryTopology") + self.assertEqual(m["tactic_count"], 5) + + def test_cad_force(self): + m = build_proof_metrics("CadForceProbeReceipt") + self.assertEqual(m["tactic_count"], 6) + + def test_logogram(self): + m = build_proof_metrics("LogogramProjection") + self.assertEqual(m["tactic_count"], 8) + + def test_unknown_shape_defaults(self): + m = build_proof_metrics("SomeUnknownShape") + self.assertEqual(m["tactic_count"], 2) + self.assertEqual(m["dependency_count"], 1) + + +# --------------------------------------------------------------------------- +# build_receipt +# --------------------------------------------------------------------------- + +class TestBuildReceipt(unittest.TestCase): + def test_receipt_structure(self): + eq = {"equation": "heat-loss", "shape": "CognitiveLoadField", "status": "verified"} + r = build_receipt(eq) + self.assertEqual(r["receipt_version"], "rrc-proof-receipt-v2") + self.assertEqual(r["theorem_name"], "heat-loss") + self.assertEqual(r["domain"], "analysis") + self.assertEqual(r["status"], "verified") + self.assertTrue(r["kernel_checked"]) + self.assertIn("proof_metrics", r) + self.assertIn("metrics", r) + + def test_domain_mapping_topology(self): + eq = {"equation": "x", "shape": "SignalShapedRouteCompiler", "status": "ok"} + r = build_receipt(eq) + self.assertEqual(r["domain"], "topology") + + def test_domain_mapping_geometry(self): + eq = {"equation": "x", "shape": "ProjectableGeometryTopology", "status": "ok"} + r = build_receipt(eq) + self.assertEqual(r["domain"], "geometry") + + def test_domain_mapping_physics(self): + eq = {"equation": "x", "shape": "CadForceProbeReceipt", "status": "ok"} + r = build_receipt(eq) + self.assertEqual(r["domain"], "physics") + + def test_domain_mapping_symbolic(self): + eq = {"equation": "x", "shape": "LogogramProjection", "status": "ok"} + r = build_receipt(eq) + self.assertEqual(r["domain"], "symbolic") + + def test_domain_mapping_unknown(self): + eq = {"equation": "x", "shape": "HoldForUnlawfulOrUnderspecifiedShape", "status": "ok"} + r = build_receipt(eq) + self.assertEqual(r["domain"], "unknown") + + def test_metrics_present(self): + eq = {"equation": "phi-mapping", "shape": "CognitiveLoadField", "status": "ok"} + r = build_receipt(eq) + self.assertIn("statement_chars", r["metrics"]) + self.assertIn("proof_chars", r["metrics"]) + self.assertIn("tactic_count", r["metrics"]) + + +# --------------------------------------------------------------------------- +# MATH sets +# --------------------------------------------------------------------------- + +class TestMathSets(unittest.TestCase): + def test_operators_non_empty(self): + self.assertTrue(len(MATH_OPERATORS) > 0) + + def test_variables_non_empty(self): + self.assertTrue(len(MATH_VARIABLES) > 0) + + def test_constants_contain_known(self): + self.assertIn("pi", MATH_CONSTANTS) + self.assertIn("e", MATH_CONSTANTS) + self.assertIn("phi", MATH_CONSTANTS) + + +if __name__ == "__main__": + unittest.main() diff --git a/5-Applications/tools-scripts/llm/test_emitter_utils.py b/5-Applications/tools-scripts/llm/test_emitter_utils.py new file mode 100644 index 00000000..629a258e --- /dev/null +++ b/5-Applications/tools-scripts/llm/test_emitter_utils.py @@ -0,0 +1,299 @@ +#!/usr/bin/env python3 +"""Unit tests for pure utility functions in ollama_deepseek_review_emitter.py.""" + +import importlib.util +import json +import sys +import tempfile +import unittest +from pathlib import Path + +MODULE_PATH = Path(__file__).with_name("ollama_deepseek_review_emitter.py") +SPEC = importlib.util.spec_from_file_location("ollama_deepseek_review_emitter", MODULE_PATH) +assert SPEC is not None and SPEC.loader is not None +emitter = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = emitter +SPEC.loader.exec_module(emitter) + + +# --------------------------------------------------------------------------- +# sha256_bytes +# --------------------------------------------------------------------------- + +class TestSha256Bytes(unittest.TestCase): + def test_deterministic(self): + h1 = emitter.sha256_bytes(b"hello") + h2 = emitter.sha256_bytes(b"hello") + self.assertEqual(h1, h2) + + def test_prefix(self): + h = emitter.sha256_bytes(b"test") + self.assertTrue(h.startswith("sha256:")) + + def test_different_inputs(self): + self.assertNotEqual(emitter.sha256_bytes(b"a"), emitter.sha256_bytes(b"b")) + + def test_empty_input(self): + h = emitter.sha256_bytes(b"") + self.assertTrue(h.startswith("sha256:")) + self.assertEqual(len(h), len("sha256:") + 64) + + +# --------------------------------------------------------------------------- +# canonical_json_bytes +# --------------------------------------------------------------------------- + +class TestCanonicalJsonBytes(unittest.TestCase): + def test_sorted_keys(self): + data = {"z": 1, "a": 2} + result = emitter.canonical_json_bytes(data) + parsed = json.loads(result) + self.assertEqual(list(parsed.keys()), ["a", "z"]) + + def test_compact_separators(self): + data = {"key": "value"} + result = emitter.canonical_json_bytes(data).decode("utf-8") + self.assertNotIn(" ", result) + self.assertIn('{"key":"value"}', result) + + def test_utf8_encoding(self): + data = {"text": "café"} + result = emitter.canonical_json_bytes(data) + self.assertIn("café".encode("utf-8"), result) + + def test_deterministic(self): + data = {"b": 2, "a": 1} + r1 = emitter.canonical_json_bytes(data) + r2 = emitter.canonical_json_bytes(data) + self.assertEqual(r1, r2) + + +# --------------------------------------------------------------------------- +# repo_relative +# --------------------------------------------------------------------------- + +class TestRepoRelative(unittest.TestCase): + def test_subpath(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + child = root / "sub" / "file.txt" + child.parent.mkdir(parents=True, exist_ok=True) + child.touch() + result = emitter.repo_relative(child, root) + self.assertEqual(result, "sub/file.txt") + + def test_outside_repo_returns_absolute(self): + result = emitter.repo_relative(Path("/outside/path.txt"), Path("/repo")) + self.assertIn("outside", result) + + +# --------------------------------------------------------------------------- +# safe_slug +# --------------------------------------------------------------------------- + +class TestSafeSlug(unittest.TestCase): + def test_basic(self): + self.assertEqual(emitter.safe_slug("hello world"), "hello_world") + + def test_special_chars(self): + self.assertEqual(emitter.safe_slug("foo@bar#baz"), "foo_bar_baz") + + def test_dots_and_dashes_preserved(self): + self.assertEqual(emitter.safe_slug("v3.2-model"), "v3.2-model") + + def test_empty_string(self): + self.assertEqual(emitter.safe_slug(""), "review") + + def test_only_special_chars(self): + self.assertEqual(emitter.safe_slug("@#$"), "review") + + def test_leading_trailing_special(self): + self.assertEqual(emitter.safe_slug(" @hello@ "), "hello") + + +# --------------------------------------------------------------------------- +# utc_timestamp / iso_from_stamp +# --------------------------------------------------------------------------- + +class TestTimestamps(unittest.TestCase): + def test_utc_timestamp_format(self): + ts = emitter.utc_timestamp() + self.assertTrue(ts.endswith("Z")) + self.assertEqual(len(ts), 16) + + def test_iso_from_stamp_roundtrip(self): + stamp = "20260512T120000Z" + iso = emitter.iso_from_stamp(stamp) + self.assertIn("2026-05-12", iso) + self.assertIn("12:00:00", iso) + self.assertIn("+00:00", iso) + + +# --------------------------------------------------------------------------- +# read_context_bundle +# --------------------------------------------------------------------------- + +class TestReadContextBundle(unittest.TestCase): + def test_bundles_files(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + f1 = root / "a.md" + f2 = root / "b.md" + f1.write_text("content A", encoding="utf-8") + f2.write_text("content B", encoding="utf-8") + text, files = emitter.read_context_bundle([f1, f2], root) + self.assertIn("content A", text) + self.assertIn("content B", text) + self.assertEqual(files, ["a.md", "b.md"]) + + def test_empty_list(self): + with tempfile.TemporaryDirectory() as tmp: + text, files = emitter.read_context_bundle([], Path(tmp)) + self.assertEqual(text, "") + self.assertEqual(files, []) + + +# --------------------------------------------------------------------------- +# build_messages +# --------------------------------------------------------------------------- + +class TestBuildMessages(unittest.TestCase): + def test_primary_message(self): + msgs = emitter.build_messages("Review this.", "context text", None) + self.assertEqual(len(msgs), 2) + self.assertEqual(msgs[0]["role"], "system") + self.assertIn("context text", msgs[1]["content"]) + self.assertIn("REVIEW REQUEST", msgs[1]["content"]) + + def test_continuation_message(self): + with tempfile.NamedTemporaryFile(mode="w", suffix=".md", delete=False) as f: + f.write("previous answer content") + f.flush() + msgs = emitter.build_messages("Continue.", "", Path(f.name)) + self.assertIn("CONTINUATION REQUEST", msgs[1]["content"]) + self.assertIn("previous answer content", msgs[1]["content"]) + import os + os.unlink(f.name) + + +# --------------------------------------------------------------------------- +# build_request_body +# --------------------------------------------------------------------------- + +class TestBuildRequestBody(unittest.TestCase): + def test_structure(self): + msgs = [{"role": "user", "content": "hello"}] + body = emitter.build_request_body("deepseek-v3", msgs, 1000, 0.5) + self.assertEqual(body["model"], "deepseek-v3") + self.assertEqual(body["messages"], msgs) + self.assertEqual(body["max_tokens"], 1000) + self.assertEqual(body["temperature"], 0.5) + self.assertFalse(body["stream"]) + + +# --------------------------------------------------------------------------- +# extract_answer +# --------------------------------------------------------------------------- + +class TestExtractAnswer(unittest.TestCase): + def test_basic_extraction(self): + response = { + "choices": [{"message": {"role": "assistant", "content": "the answer"}}], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, + } + text, usage, keys = emitter.extract_answer(response) + self.assertEqual(text, "the answer") + self.assertEqual(usage["prompt_tokens"], 10) + self.assertEqual(usage["completion_tokens"], 5) + self.assertEqual(usage["total_tokens"], 15) + + def test_missing_usage_defaults_to_zero(self): + response = {"choices": [{"message": {"role": "assistant", "content": "text"}}]} + text, usage, keys = emitter.extract_answer(response) + self.assertEqual(usage["prompt_tokens"], 0) + self.assertEqual(usage["completion_tokens"], 0) + + def test_total_tokens_computed_if_zero(self): + response = { + "choices": [{"message": {"role": "assistant", "content": "text"}}], + "usage": {"prompt_tokens": 3, "completion_tokens": 7, "total_tokens": 0}, + } + _, usage, _ = emitter.extract_answer(response) + self.assertEqual(usage["total_tokens"], 10) + + def test_no_choices_raises(self): + with self.assertRaises(ValueError): + emitter.extract_answer({"choices": []}) + + def test_no_choices_key_raises(self): + with self.assertRaises(ValueError): + emitter.extract_answer({}) + + def test_message_keys_returned(self): + response = { + "choices": [{"message": {"role": "assistant", "content": "x", "reasoning": "y"}}], + } + _, _, keys = emitter.extract_answer(response) + self.assertIn("reasoning", keys) + + +# --------------------------------------------------------------------------- +# call_ollama_chat: auth guard +# --------------------------------------------------------------------------- + +class TestCallOllamaChatAuth(unittest.TestCase): + def test_remote_without_key_raises(self): + with self.assertRaises(RuntimeError): + emitter.call_ollama_chat("https://remote.api/v1/chat", None, b"{}") + + +# --------------------------------------------------------------------------- +# verify_receipt schema validation +# --------------------------------------------------------------------------- + +class TestVerifyReceiptSchemaRules(unittest.TestCase): + def test_primary_rejects_previous_answer_path(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + answer = root / "answer.md" + answer.write_text("content", encoding="utf-8") + receipt = { + "schema": emitter.PRIMARY_SCHEMA, + "answer_path": "answer.md", + "answer_sha256": emitter.sha256_bytes(answer.read_bytes()), + "context_files": ["ctx.md"], + "previous_answer_path": "old.md", + } + rp = root / "test.receipt.json" + rp.write_text(json.dumps(receipt), encoding="utf-8") + with self.assertRaises(ValueError): + emitter.verify_receipt(rp, root) + + def test_continuation_rejects_context_files(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + answer = root / "answer.md" + answer.write_text("content", encoding="utf-8") + receipt = { + "schema": emitter.CONTINUATION_SCHEMA, + "answer_path": "answer.md", + "answer_sha256": emitter.sha256_bytes(answer.read_bytes()), + "previous_answer_path": "prev.md", + "context_files": ["ctx.md"], + } + rp = root / "test.receipt.json" + rp.write_text(json.dumps(receipt), encoding="utf-8") + with self.assertRaises(ValueError): + emitter.verify_receipt(rp, root) + + def test_unknown_schema_raises(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + rp = root / "bad.receipt.json" + rp.write_text(json.dumps({"schema": "unknown_v99"}), encoding="utf-8") + with self.assertRaises(ValueError): + emitter.verify_receipt(rp, root) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/qc-flag/test_lean_qc_flagger.py b/scripts/qc-flag/test_lean_qc_flagger.py new file mode 100644 index 00000000..67552b91 --- /dev/null +++ b/scripts/qc-flag/test_lean_qc_flagger.py @@ -0,0 +1,411 @@ +#!/usr/bin/env python3 +"""Unit tests for lean_qc_flagger.py — QC inspection helpers.""" + +import importlib.util +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path + +MODULE_PATH = Path(__file__).with_name("lean_qc_flagger.py") +SPEC = importlib.util.spec_from_file_location("lean_qc_flagger", MODULE_PATH) +assert SPEC is not None and SPEC.loader is not None +flagger = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = flagger +SPEC.loader.exec_module(flagger) + + +# --------------------------------------------------------------------------- +# Helper utilities +# --------------------------------------------------------------------------- + +class TestGetLine(unittest.TestCase): + def test_first_line(self): + self.assertEqual(flagger._get_line("abc\ndef\n", 0), 1) + + def test_second_line(self): + self.assertEqual(flagger._get_line("abc\ndef\n", 4), 2) + + def test_end_of_content(self): + self.assertEqual(flagger._get_line("abc\ndef\nghi", 10), 3) + + +class TestDefNames(unittest.TestCase): + def test_extracts_defs(self): + src = "def foo := 1\ndef bar := 2" + self.assertEqual(flagger._def_names(src), {"foo", "bar"}) + + def test_no_defs(self): + self.assertEqual(flagger._def_names("theorem t : True := trivial"), set()) + + +class TestTheoremNames(unittest.TestCase): + def test_extracts_theorems(self): + src = "theorem thmA : True := trivial\ntheorem thmB : Nat = Nat := rfl" + self.assertEqual(flagger._theorem_names(src), {"thmA", "thmB"}) + + +class TestPrivateDefNames(unittest.TestCase): + def test_extracts_private(self): + src = "private def helper := 42\ndef public := 1" + self.assertEqual(flagger._private_def_names(src), {"helper"}) + + +class TestGetImportsOpens(unittest.TestCase): + def test_get_imports(self): + src = "import Mathlib.Data.Nat\nimport Semantics.RRC\ndef foo := 1" + self.assertEqual(flagger._get_imports(src), ["Mathlib.Data.Nat", "Semantics.RRC"]) + + def test_get_opens(self): + src = "open Nat\nopen List\ntheorem t := by decide" + self.assertEqual(flagger._get_opens(src), ["Nat", "List"]) + + +class TestNormalizePathSep(unittest.TestCase): + def test_backslash(self): + self.assertEqual(flagger._normalize_path_sep("a\\b\\c"), "a/b/c") + + def test_forward_slash_unchanged(self): + self.assertEqual(flagger._normalize_path_sep("a/b/c"), "a/b/c") + + +# --------------------------------------------------------------------------- +# Code-line detection (block comment filtering) +# --------------------------------------------------------------------------- + +class TestGetCodeLines(unittest.TestCase): + def test_plain_code(self): + src = "def foo := 1\ntheorem t := by decide" + result = flagger._get_code_lines(src) + self.assertTrue(all(v is True for v in result)) + + def test_line_comments(self): + src = "-- comment\ndef foo := 1" + result = flagger._get_code_lines(src) + self.assertFalse(result[0]) + self.assertTrue(result[1]) + + def test_block_comment(self): + src = "/- block\n comment -/\ndef foo := 1" + result = flagger._get_code_lines(src) + self.assertFalse(result[0]) + self.assertFalse(result[1]) + self.assertTrue(result[2]) + + def test_empty_lines(self): + src = "\n\ndef foo := 1" + result = flagger._get_code_lines(src) + self.assertFalse(result[0]) + self.assertFalse(result[1]) + self.assertTrue(result[2]) + + def test_single_line_block_comment(self): + src = "/- inline -/\ndef foo := 1" + result = flagger._get_code_lines(src) + self.assertFalse(result[0]) + self.assertTrue(result[1]) + + +class TestPosIsCode(unittest.TestCase): + def test_code_position(self): + src = "def foo := 1" + self.assertTrue(flagger._pos_is_code(src, 0)) + + def test_comment_position(self): + src = "-- comment\ndef foo := 1" + self.assertFalse(flagger._pos_is_code(src, 3)) + + +class TestIsDataDef(unittest.TestCase): + def test_struct_def(self): + src = "def myConfig : Config := {\n field1 := 1\n}" + self.assertTrue(flagger._is_data_def(src, "myConfig")) + + def test_numeric_def(self): + src = "def maxSize : Nat := 42" + self.assertTrue(flagger._is_data_def(src, "maxSize")) + + def test_function_def(self): + src = "def addOne (n : Nat) : Nat := n + 1" + self.assertFalse(flagger._is_data_def(src, "addOne")) + + +# --------------------------------------------------------------------------- +# QCIssue / FileResult +# --------------------------------------------------------------------------- + +class TestQCIssue(unittest.TestCase): + def test_to_dict(self): + issue = flagger.QCIssue("structural_health", "test msg", 10, "ERROR") + d = issue.to_dict() + self.assertEqual(d["check"], "structural_health") + self.assertEqual(d["message"], "test msg") + self.assertEqual(d["line"], 10) + self.assertEqual(d["severity"], "ERROR") + + +class TestFileResult(unittest.TestCase): + def test_empty_result_passes(self): + r = flagger.FileResult("test.lean") + self.assertTrue(r.passed) + self.assertEqual(r.issues, []) + + def test_warning_keeps_pass(self): + r = flagger.FileResult("test.lean") + r.add_issue(flagger.QCIssue("check", "msg", 1, "WARNING")) + self.assertTrue(r.passed) + + def test_error_fails(self): + r = flagger.FileResult("test.lean") + r.add_issue(flagger.QCIssue("check", "msg", 1, "ERROR")) + self.assertFalse(r.passed) + + def test_to_dict_shape(self): + r = flagger.FileResult("Test.lean") + r.structural = {"theorems": 2} + r.add_issue(flagger.QCIssue("check", "msg", 5, "WARNING")) + d = r.to_dict() + self.assertEqual(d["path"], "Test.lean") + self.assertTrue(d["passed"]) + self.assertEqual(d["issue_count"], 1) + self.assertEqual(d["structural"]["theorems"], 2) + + +# --------------------------------------------------------------------------- +# Check functions +# --------------------------------------------------------------------------- + +class TestCheckStructuralHealth(unittest.TestCase): + def test_counts_theorems_and_defs(self): + src = "theorem t1 : True := trivial\ndef f := 1\ndef g := 2" + r = flagger.FileResult("Test.lean") + flagger.check_structural_health(src, r) + self.assertEqual(r.structural["theorems"], 1) + self.assertEqual(r.structural["defs"], 2) + + def test_sorry_flagged_as_error(self): + src = "theorem t : True := by sorry" + r = flagger.FileResult("Test.lean") + flagger.check_structural_health(src, r) + self.assertFalse(r.passed) + self.assertTrue(any(i.check == "structural_health" and "sorry" in i.message for i in r.issues)) + + def test_sorry_in_comment_not_flagged(self): + src = "-- sorry is in comment\ntheorem t : True := trivial" + r = flagger.FileResult("Test.lean") + flagger.check_structural_health(src, r) + self.assertEqual(r.structural["sorries"], 0) + + def test_empty_theorem_detected(self): + src = "theorem t : True := by\ntheorem t2 : False := by decide" + r = flagger.FileResult("Test.lean") + flagger.check_structural_health(src, r) + self.assertEqual(r.structural["empty_theorems"], 1) + + def test_eval_and_eval_bang_counted(self): + src = "#eval 1 + 1\n#eval! IO.println" + r = flagger.FileResult("Test.lean") + flagger.check_structural_health(src, r) + self.assertEqual(r.structural["eval"], 1) + self.assertEqual(r.structural["eval_bang"], 1) + + def test_native_decide_counted(self): + src = "theorem t := by native_decide" + r = flagger.FileResult("Test.lean") + flagger.check_structural_health(src, r) + self.assertEqual(r.structural["native_decide"], 1) + + +class TestCheckNamingConventions(unittest.TestCase): + def test_pascal_case_file_passes(self): + src = "def foo := 1" + r = flagger.FileResult("MyModule.lean") + flagger.check_naming_conventions(src, r, "MyModule.lean") + naming_issues = [i for i in r.issues if "File name" in i.message] + self.assertEqual(naming_issues, []) + + def test_snake_case_file_flagged(self): + src = "def foo := 1" + r = flagger.FileResult("my_module.lean") + flagger.check_naming_conventions(src, r, "my_module.lean") + self.assertTrue(any("snake_case" in i.message for i in r.issues)) + + def test_non_pascal_type_flagged(self): + src = "inductive myType where | a | b" + r = flagger.FileResult("Test.lean") + flagger.check_naming_conventions(src, r, "Test.lean") + self.assertTrue(any("myType" in i.message and "PascalCase" in i.message for i in r.issues)) + + def test_non_camel_def_flagged(self): + src = "def MyFunc := 1" + r = flagger.FileResult("Test.lean") + flagger.check_naming_conventions(src, r, "Test.lean") + self.assertTrue(any("MyFunc" in i.message for i in r.issues)) + + def test_banned_prefix_flagged(self): + src = "def getItems := []" + r = flagger.FileResult("Test.lean") + flagger.check_naming_conventions(src, r, "Test.lean") + self.assertTrue(any("Banned prefix" in i.message for i in r.issues)) + + def test_banned_suffix_flagged(self): + src = "def helper_v2 := 1" + r = flagger.FileResult("Test.lean") + flagger.check_naming_conventions(src, r, "Test.lean") + self.assertTrue(any("Banned suffix" in i.message for i in r.issues)) + + +class TestCheckQ16Compliance(unittest.TestCase): + def test_float_in_code_flagged(self): + src = "def x : Float := 1.0" + r = flagger.FileResult("Test.lean") + flagger.check_q16_compliance(src, r) + self.assertTrue(any("Float" in i.message for i in r.issues)) + + def test_float_in_comment_not_flagged(self): + src = "-- Float is fine in comments\ndef x : Nat := 1" + r = flagger.FileResult("Test.lean") + flagger.check_q16_compliance(src, r) + q16_issues = [i for i in r.issues if i.check == "q16_compliance"] + self.assertEqual(q16_issues, []) + + +class TestCheckProofQuality(unittest.TestCase): + def test_def_without_companion_flagged(self): + src = "def orphanFunc (n : Nat) : Nat := n + 1" + r = flagger.FileResult("Test.lean") + flagger.check_proof_quality(src, r) + self.assertTrue(any("orphanFunc" in i.message and "companion" in i.message for i in r.issues)) + + def test_def_with_companion_theorem_ok(self): + src = "def addOne (n : Nat) : Nat := n + 1\ntheorem addOneSpec : addOne 0 = 1 := rfl" + r = flagger.FileResult("Test.lean") + flagger.check_proof_quality(src, r) + orphan = [i for i in r.issues if "addOne" in i.message and "companion" in i.message] + self.assertEqual(orphan, []) + + def test_private_def_not_flagged(self): + src = "private def helper := 42" + r = flagger.FileResult("Test.lean") + flagger.check_proof_quality(src, r) + orphan = [i for i in r.issues if "helper" in i.message and "companion" in i.message] + self.assertEqual(orphan, []) + + def test_data_def_not_flagged(self): + src = "def maxSize : Nat := 42" + r = flagger.FileResult("Test.lean") + flagger.check_proof_quality(src, r) + orphan = [i for i in r.issues if "maxSize" in i.message and "companion" in i.message] + self.assertEqual(orphan, []) + + +class TestCheckDependencyAnalysis(unittest.TestCase): + def test_unused_import_flagged(self): + src = "import Semantics.Unused\ndef foo := 1" + r = flagger.FileResult("Test.lean") + flagger.check_dependency_analysis(src, r, "Test.lean") + self.assertTrue(any("Unused import" in i.message for i in r.issues)) + + def test_used_import_not_flagged(self): + src = "import Semantics.RRC\ndef foo := RRC.classify x" + r = flagger.FileResult("Test.lean") + flagger.check_dependency_analysis(src, r, "Test.lean") + unused = [i for i in r.issues if "Unused import" in i.message] + self.assertEqual(unused, []) + + def test_opened_import_not_flagged(self): + src = "import Semantics.Types\nopen Types\ndef foo := mkValue 1" + r = flagger.FileResult("Test.lean") + flagger.check_dependency_analysis(src, r, "Test.lean") + unused = [i for i in r.issues if "Unused import" in i.message] + self.assertEqual(unused, []) + + +# --------------------------------------------------------------------------- +# scan_file integration +# --------------------------------------------------------------------------- + +class TestScanFile(unittest.TestCase): + def test_scan_clean_file(self): + with tempfile.NamedTemporaryFile(suffix=".lean", mode="w", delete=False) as f: + f.write("def addOne (n : Nat) : Nat := n + 1\n#eval addOne 0\n") + f.flush() + try: + result = flagger.scan_file(f.name) + self.assertTrue(result.passed) + finally: + os.unlink(f.name) + + def test_scan_file_with_sorry(self): + with tempfile.NamedTemporaryFile(suffix=".lean", mode="w", delete=False) as f: + f.write("theorem bogus : True := by sorry\n") + f.flush() + try: + result = flagger.scan_file(f.name) + self.assertFalse(result.passed) + finally: + os.unlink(f.name) + + def test_scan_nonexistent_file(self): + result = flagger.scan_file("/tmp/nonexistent_file_12345.lean") + self.assertFalse(result.passed) + + +# --------------------------------------------------------------------------- +# find_lean_files / gather_imports +# --------------------------------------------------------------------------- + +class TestFindLeanFiles(unittest.TestCase): + def test_single_file(self): + with tempfile.NamedTemporaryFile(suffix=".lean", delete=False) as f: + try: + files = flagger.find_lean_files(f.name) + self.assertEqual(len(files), 1) + finally: + os.unlink(f.name) + + def test_directory(self): + with tempfile.TemporaryDirectory() as td: + (Path(td) / "A.lean").write_text("def a := 1") + (Path(td) / "B.lean").write_text("def b := 2") + (Path(td) / "not_lean.py").write_text("pass") + files = flagger.find_lean_files(td) + self.assertEqual(len(files), 2) + + def test_nonexistent_path(self): + files = flagger.find_lean_files("/tmp/nonexistent_dir_12345") + self.assertEqual(files, []) + + +class TestGatherImports(unittest.TestCase): + def test_gathers_imports_from_files(self): + with tempfile.TemporaryDirectory() as td: + a = Path(td) / "A.lean" + a.write_text("import Semantics.RRC\ndef foo := 1") + b = Path(td) / "B.lean" + b.write_text("import Semantics.Types\nimport Semantics.AVM\ndef bar := 2") + imports = flagger.gather_imports([str(a), str(b)]) + self.assertEqual(imports[str(a)], ["Semantics.RRC"]) + self.assertEqual(imports[str(b)], ["Semantics.Types", "Semantics.AVM"]) + + +# --------------------------------------------------------------------------- +# Markdown report generation +# --------------------------------------------------------------------------- + +class TestGenerateMarkdownReport(unittest.TestCase): + def test_report_structure(self): + r = flagger.FileResult("Test.lean") + r.structural = {"theorems": 1, "defs": 0} + r.add_issue(flagger.QCIssue("check", "test issue", 5, "WARNING")) + md = flagger.generate_markdown_report([r], "/path") + self.assertIn("QC Flag Report", md) + self.assertIn("Test.lean", md) + self.assertIn("PASS", md) + self.assertIn("test issue", md) + + +if __name__ == "__main__": + unittest.main()