mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
f739b5b4c8
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 <bigdataiscoming+9i37y6j2@protonmail.com>
299 lines
11 KiB
Python
299 lines
11 KiB
Python
#!/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()
|