#!/usr/bin/env python3 """Unit tests for readability_audit.py. Run with: python3 -m unittest scripts.readability_audit_test Tests use temporary fixture trees, temporary git repositories, and in-memory baselines to verify: - tracked inputs sorting and exclusion - language-aware file-kind classification and threshold boundaries - multi-language function identity (raw/multiline/comment/heredoc aware) - ratchet identity where level drops pass and new/increased identities fail - strict baseline schema validation - task read-set configuration failures, budget/reason, and total ratchet - CLI exit codes and rendered failure output - input mode: tracked vs worktree projection (modified/deleted/untracked) """ import json import os import shutil import subprocess import sys import tempfile import textwrap import unittest # Ensure the audit module is importable. SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, SCRIPTS_DIR) import readability_audit as ra AUDIT_SCRIPT = os.path.join(SCRIPTS_DIR, "readability_audit.py") class _RepoFixture(unittest.TestCase): """Base fixture: a temporary directory plus optional git repository.""" prefix = "ra-test-" def setUp(self): self.tmpdir = tempfile.mkdtemp(prefix=self.prefix) def tearDown(self): shutil.rmtree(self.tmpdir, ignore_errors=True) def _write(self, relpath: str, content: str) -> None: full = os.path.join(self.tmpdir, relpath) os.makedirs(os.path.dirname(full) or self.tmpdir, exist_ok=True) with open(full, "w") as f: f.write(content) def _git(self, *args: str) -> subprocess.CompletedProcess: return subprocess.run(["git", *args], check=True, cwd=self.tmpdir, capture_output=True) def _git_init(self) -> None: subprocess.run(["git", "init", "-q", self.tmpdir], check=True, capture_output=True) self._git("config", "user.email", "test@test") self._git("config", "user.name", "Test") self._git("add", "-A") self._git("commit", "-q", "-m", "init") # ============================================================ # TestTrackedInputsSortedAndExcluded # ============================================================ class TestTrackedInputsSortedAndExcluded(unittest.TestCase): """Input filtering: tracked languages only, excluded dirs removed, sorted.""" def test_tracked_languages_only_and_excluded(self) -> None: original = ra._run_git_ls_files_z ra._run_git_ls_files_z = lambda: [ "src/main.go", "src/util.py", "proto/gen/generate.pb.go", "apps/client/lib/gen/api.dart", "build/vendor/lib.go", "scripts/run.sh", "src/lib.kt", "src/view.swift", "src/app.dart", "src/data.json", ] try: tracked = [f for f in ra._run_git_ls_files_z() if not ra._should_exclude(f)] tracked_lang = sorted(f for f in tracked if ra._detect_language(f)) self.assertEqual(tracked_lang, [ "scripts/run.sh", "src/app.dart", "src/lib.kt", "src/main.go", "src/util.py", "src/view.swift", ]) self.assertNotIn("src/data.json", tracked_lang) finally: ra._run_git_ls_files_z = original def test_exclusion_patterns(self) -> None: self.assertTrue(ra._should_exclude("proto/gen/foo.go")) self.assertTrue(ra._should_exclude("apps/client/lib/gen/bar.dart")) self.assertTrue(ra._should_exclude("build/vendor/x.go")) self.assertTrue(ra._should_exclude("agent-ops/rules/common/rules-roadmap.md")) self.assertTrue(ra._should_exclude("agent-ops/skills/common/plan/SKILL.md")) self.assertFalse(ra._should_exclude("agent-ops/skills/project/e2e-smoke/SKILL.md")) self.assertFalse(ra._should_exclude("src/main.go")) self.assertFalse(ra._should_exclude("scripts/run.sh")) # ============================================================ # TestPolicyThresholdsByFileKind [REVIEW_REVIEW_REFACTOR-2] # ============================================================ class TestPolicyThresholdsByFileKind(_RepoFixture): """File-kind policy: language-aware test classification and boundaries.""" prefix = "ra-test-policy-" def test_classify_file(self) -> None: self.assertEqual(ra.classify_file("apps/edge/main.go"), "production") self.assertEqual(ra.classify_file("apps/edge/test/foo_test.go"), "test") self.assertEqual(ra.classify_file("apps/edge/foo_test.go"), "test") self.assertEqual(ra.classify_file("agent-ops/skills/my-skill/SKILL.md"), "skill_entrypoint") self.assertEqual(ra.classify_file("agent-ops/skills/other/SKILL.md"), "skill_entrypoint") def test_classify_file_language_aware_test_names(self) -> None: """Every instrumented language reports its own test naming convention.""" self.assertEqual(ra.classify_file("scripts/readability_audit_test.py"), "test") self.assertEqual(ra.classify_file("scripts/test_readability.py"), "test") self.assertEqual(ra.classify_file("apps/client/lib/widget_test.dart"), "test") self.assertEqual(ra.classify_file("android/app/src/RepoTest.kt"), "test") self.assertEqual(ra.classify_file("android/app/src/RepoTests.kt"), "test") self.assertEqual(ra.classify_file("ios/Runner/AppTests.swift"), "test") self.assertEqual(ra.classify_file("scripts/e2e_test.sh"), "test") self.assertEqual(ra.classify_file("scripts/tests/helper.sh"), "test") # production files that merely mention test must stay production self.assertEqual(ra.classify_file("apps/edge/testing.go"), "production") self.assertEqual(ra.classify_file("scripts/readability_audit.py"), "production") def test_classify_project_shell_test_entrypoints(self) -> None: """Real scripts/e2e-*.sh entrypoints are tests; other shell stays production.""" repo_root = os.path.dirname(SCRIPTS_DIR) for path in ("scripts/e2e-smoke.sh", "scripts/e2e-long-context-admission-smoke.sh"): self.assertTrue(os.path.isfile(os.path.join(repo_root, path)), path) self.assertEqual(ra.classify_file(path), "test", path) self.assertEqual(ra._file_threshold_for_kind(ra.classify_file(path)), ra.FILE_POLICY["test"], path) for path in ("scripts/dev/edge.sh", "agent-ops/bin/sync.sh"): self.assertTrue(os.path.isfile(os.path.join(repo_root, path)), path) self.assertEqual(ra.classify_file(path), "production", path) self.assertEqual(ra._file_threshold_for_kind(ra.classify_file(path)), ra.FILE_POLICY["production"], path) # the pattern is anchored: neither a bare name nor a nested copy matches self.assertEqual(ra.classify_file("scripts/e2e.sh"), "production") self.assertEqual(ra.classify_file("tools/scripts/e2e-smoke.sh"), "production") def test_shell_entrypoint_uses_test_threshold_end_to_end(self) -> None: """An 810-line e2e entrypoint reports test 'warning', not production's.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: body = "\n".join(f"echo {i}" for i in range(810)) + "\n" self._write("scripts/e2e-fixture-smoke.sh", body) self._write("scripts/dev/fixture.sh", body) result = ra.audit(["scripts/e2e-fixture-smoke.sh", "scripts/dev/fixture.sh"]) kinds = {e["path"]: e["kind"] for e in result["files"]} levels = {v["path"]: v["level"] for v in result["violations"] if v["metric"] == "file_loc"} self.assertEqual(kinds["scripts/e2e-fixture-smoke.sh"], "test") self.assertEqual(kinds["scripts/dev/fixture.sh"], "production") self.assertEqual(levels["scripts/e2e-fixture-smoke.sh"], "warning") self.assertEqual(levels["scripts/dev/fixture.sh"], "split_review") finally: os.chdir(old_cwd) def test_python_test_file_uses_test_policy(self) -> None: """A 700-line *_test.py is under the test warning threshold, not production's.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: body = "\n".join(f"# line {i}" for i in range(700)) + "\n" self._write("pkg/thing_test.py", body) self._write("pkg/thing.py", body) result = ra.audit(["pkg/thing_test.py", "pkg/thing.py"]) levels = {v["path"]: v["level"] for v in result["violations"] if v["metric"] == "file_loc"} self.assertNotIn("pkg/thing_test.py", levels) self.assertEqual(levels["pkg/thing.py"], "warning") finally: os.chdir(old_cwd) def test_production_file_threshold_exact_and_one_over(self) -> None: """production: warning=500, split_review=800, exception=1000.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: for count in (500, 501, 800, 801, 1000, 1001): lines = "\n".join(f"line {i}" for i in range(count)) + "\n" self._write(f"src/f_{count}.go", lines) files = [f"src/f_{c}.go" for c in (500, 501, 800, 801, 1000, 1001)] result = ra.audit(files) file_violations = [v for v in result["violations"] if v["metric"] == "file_loc"] self.assertEqual(len(file_violations), 5) levels_by_path = {v["path"]: v["level"] for v in file_violations} self.assertNotIn("src/f_500.go", levels_by_path) self.assertEqual(levels_by_path["src/f_501.go"], "warning") self.assertEqual(levels_by_path["src/f_800.go"], "warning") self.assertEqual(levels_by_path["src/f_801.go"], "split_review") self.assertEqual(levels_by_path["src/f_1000.go"], "split_review") self.assertEqual(levels_by_path["src/f_1001.go"], "exception") finally: os.chdir(old_cwd) def test_test_file_threshold_exact_and_one_over(self) -> None: """test: warning=800, split_review=1000.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: for count in (800, 801, 1000, 1001): lines = "\n".join(f"line {i}" for i in range(count)) + "\n" self._write(f"test/f_{count}_test.go", lines) files = [f"test/f_{c}_test.go" for c in (800, 801, 1000, 1001)] result = ra.audit(files) file_violations = [v for v in result["violations"] if v["metric"] == "file_loc"] self.assertEqual(len(file_violations), 3) levels_by_path = {v["path"]: v["level"] for v in file_violations} self.assertNotIn("test/f_800_test.go", levels_by_path) self.assertEqual(levels_by_path["test/f_801_test.go"], "warning") self.assertEqual(levels_by_path["test/f_1000_test.go"], "warning") self.assertEqual(levels_by_path["test/f_1001_test.go"], "split_review") finally: os.chdir(old_cwd) def test_skill_entrypoint_threshold(self) -> None: """skill_entrypoint: warning=300, split_review=500.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: for name, count in (("foo", 300), ("bar", 301), ("baz", 500), ("qux", 501)): lines = "\n".join(f"line {i}" for i in range(count)) + "\n" self._write(f"agent-ops/skills/{name}/SKILL.md", lines) files = [f"agent-ops/skills/{n}/SKILL.md" for n in ("foo", "bar", "baz", "qux")] result = ra.audit(files) file_violations = [v for v in result["violations"] if v["metric"] == "file_loc"] self.assertEqual(len(file_violations), 3) levels_by_path = {v["path"]: v["level"] for v in file_violations} self.assertNotIn("agent-ops/skills/foo/SKILL.md", levels_by_path) self.assertEqual(levels_by_path["agent-ops/skills/bar/SKILL.md"], "warning") self.assertEqual(levels_by_path["agent-ops/skills/baz/SKILL.md"], "warning") self.assertEqual(levels_by_path["agent-ops/skills/qux/SKILL.md"], "split_review") finally: os.chdir(old_cwd) def test_function_threshold(self) -> None: """function: warning=80, split_review=120.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: for count in (79, 81, 120, 121): body = "\n".join(f" line{i}" for i in range(count)) self._write(f"src/fn_{count}.go", f"package p\n\nfunc fn{count}() {{\n{body}\n}}\n") files = [f"src/fn_{c}.go" for c in (79, 81, 120, 121)] result = ra.audit(files) func_violations = [v for v in result["violations"] if v["metric"] == "function_loc"] self.assertEqual(len(func_violations), 4) levels_by_path = {v["path"]: v["level"] for v in func_violations} self.assertEqual(levels_by_path["src/fn_79.go"], "warning") self.assertEqual(levels_by_path["src/fn_81.go"], "warning") self.assertEqual(levels_by_path["src/fn_120.go"], "split_review") self.assertEqual(levels_by_path["src/fn_121.go"], "split_review") finally: os.chdir(old_cwd) # ============================================================ # TestMultiLanguageFunctionIdentity [REVIEW_REVIEW_REFACTOR-4] # ============================================================ class TestMultiLanguageFunctionIdentity(unittest.TestCase): """Raw/multiline/comment/heredoc aware measurement and qualified identity.""" def test_go_qualified_name_with_receiver(self) -> None: source = textwrap.dedent("""\ package main type MyType struct{} func (m *MyType) DoSomething() { x := 1 y := 2 } func Standalone() { z := 3 } """) names = {f["name"] for f in ra._extract_functions_go(source, "test.go")} self.assertIn("MyType.DoSomething", names) self.assertIn("Standalone", names) def test_go_brace_in_string_not_counted(self) -> None: source = textwrap.dedent("""\ package main func WithBraceInString() string { s := "hello } world" return s } """) funcs = ra._extract_functions_go(source, "test.go") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "WithBraceInString") self.assertEqual(funcs[0]["loc"], 4) def test_go_raw_string_does_not_end_function_early(self) -> None: """A backtick raw string with a brace must not terminate the function.""" source = textwrap.dedent("""\ package main func WithRawString() string { s := `line one } not a brace line three` return s } """) funcs = ra._extract_functions_go(source, "test.go") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "WithRawString") self.assertEqual(funcs[0]["loc"], 6) def test_go_block_comment_ignored(self) -> None: source = textwrap.dedent("""\ package main func WithBlockComment() { /* this is a { block comment } with braces */ x := 1 } """) funcs = ra._extract_functions_go(source, "test.go") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "WithBlockComment") self.assertEqual(funcs[0]["loc"], 4) def test_go_block_comment_does_not_nest(self) -> None: """Go block comments do not nest: the first */ closes the comment.""" source = textwrap.dedent("""\ package main func WithComment() { /* outer /* inner */ x := 1 } """) funcs = ra._extract_functions_go(source, "test.go") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "WithComment") self.assertEqual(funcs[0]["loc"], 4) def test_kotlin_nested_block_comment_ignored(self) -> None: """Kotlin comments nest, so an inner */ must not resume brace counting.""" source = textwrap.dedent("""\ class Repo { fun query(): String { /* outer /* inner } */ still comment } */ return "ok" } } """) funcs = ra._extract_functions_kotlin(source, "test.kt") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "Repo.query") self.assertEqual(funcs[0]["loc"], 4) def test_swift_nested_block_comment_ignored(self) -> None: source = textwrap.dedent("""\ extension Foo { func render() -> String { /* outer /* inner } */ still comment } */ return "ok" } } """) funcs = ra._extract_functions_swift(source, "test.swift") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "Foo.render") self.assertEqual(funcs[0]["loc"], 4) def test_shell_multiline_quote_holds_awk_body(self) -> None: """A quoted awk body spans newlines: its braces and callables are text.""" source = textwrap.dedent("""\ summarize() { awk ' function reset() { count = 0 } /^}/ { reset(); print } ' "$1" echo done } helper() { echo "hi" } """) funcs = ra._extract_functions_shell(source, "test.sh") names = {f["name"] for f in funcs} self.assertEqual(names, {"summarize", "helper"}) by_name = {f["name"]: f for f in funcs} self.assertEqual(by_name["summarize"]["loc"], 9) self.assertEqual(by_name["helper"]["loc"], 3) def test_shell_multiline_double_quote_preserves_lines(self) -> None: source = textwrap.dedent("""\ emit() { printf "%s } not a brace " "$1" echo done } """) funcs = ra._extract_functions_shell(source, "test.sh") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "emit") self.assertEqual(funcs[0]["loc"], 6) def test_dart_generic_async_function(self) -> None: source = textwrap.dedent("""\ import 'dart:async'; Future myAsyncFunc() async { await doSomething(); print('done'); } String simpleFunc() { return 'hello'; } """) names = {f["name"] for f in ra._extract_functions_dart(source, "test.dart")} self.assertIn("myAsyncFunc", names) self.assertIn("simpleFunc", names) def test_dart_string_braces_ignored(self) -> None: source = textwrap.dedent("""\ void main() { var s = "hello } world"; print(s); } """) funcs = ra._extract_functions_dart(source, "test.dart") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "main") def test_dart_multiline_string_braces_ignored(self) -> None: source = textwrap.dedent("""\ void withMultiline() { var s = ''' } not a brace '''; print(s); } """) funcs = ra._extract_functions_dart(source, "test.dart") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "withMultiline") self.assertEqual(funcs[0]["loc"], 6) def test_dart_class_scope_identity(self) -> None: source = textwrap.dedent("""\ class Repo { void save() { print('a'); } } class Cache { void save() { print('b'); } } """) names = {f["name"] for f in ra._extract_functions_dart(source, "test.dart")} self.assertEqual(names, {"Repo.save", "Cache.save"}) def test_dart_callback_type_is_not_a_function(self) -> None: """`void Function(...)` declares a callback type, never a callable named Function.""" source = textwrap.dedent("""\ class Button { void Function() onTap; void Function(String value)? onChanged; void register(void Function() cb) { onTap = cb; } } """) funcs = ra._extract_functions_dart(source, "test.dart") names = {f["name"] for f in funcs} self.assertEqual(names, {"Button.register"}) self.assertEqual(funcs[0]["loc"], 3) def test_dart_body_less_declaration_is_not_a_function(self) -> None: """An abstract declaration has no body to measure.""" source = textwrap.dedent("""\ abstract class Repo { void save(); Future load(); } class FileRepo extends Repo { void save() { print('saved'); } } """) names = {f["name"] for f in ra._extract_functions_dart(source, "test.dart")} self.assertEqual(names, {"FileRepo.save"}) def test_dart_expression_body_is_measured(self) -> None: """A `=>` body is a real callable and spans to its terminating `;`.""" source = textwrap.dedent("""\ class View { Widget build() => Column( children: [ Text('a'), ], ); } """) funcs = ra._extract_functions_dart(source, "test.dart") self.assertEqual([f["name"] for f in funcs], ["View.build"]) self.assertEqual(funcs[0]["loc"], 5) self.assertEqual((funcs[0]["start"], funcs[0]["end"]), (2, 6)) def test_shell_body_less_call_is_not_a_function(self) -> None: """`name()` with no body is not a declaration; the real one still is.""" source = textwrap.dedent("""\ #!/bin/sh usage() { echo "usage" } if [ -z "$1" ]; then usage fi """) names = {f["name"] for f in ra._extract_functions_shell(source, "test.sh")} self.assertEqual(names, {"usage"}) def test_shell_next_line_brace_body(self) -> None: source = textwrap.dedent("""\ run_all() { echo "a" } """) funcs = ra._extract_functions_shell(source, "test.sh") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "run_all") self.assertEqual(funcs[0]["loc"], 4) def test_next_line_type_brace_keeps_scope_identity(self) -> None: """A type whose body opener is on a later line still qualifies its methods.""" dart = textwrap.dedent("""\ class Repo extends Base { void save() { print('a'); } } """) self.assertEqual({f["name"] for f in ra._extract_functions_dart(dart, "t.dart")}, {"Repo.save"}) kotlin = textwrap.dedent("""\ class Repo( private val db: Db, ) { fun save(): Int { return 1 } } """) self.assertEqual({f["name"] for f in ra._extract_functions_kotlin(kotlin, "t.kt")}, {"Repo.save"}) swift = textwrap.dedent("""\ struct Repo { func save() { print("a") } } """) self.assertEqual({f["name"] for f in ra._extract_functions_swift(swift, "t.swift")}, {"Repo.save"}) def test_go_multiline_signature_with_receiver(self) -> None: """A receiver clause is not the header: the parameter list is.""" source = textwrap.dedent("""\ package main func (m *MyType) DoSomething( a int, b string, ) error { return nil } """) funcs = ra._extract_functions_go(source, "test.go") self.assertEqual([f["name"] for f in funcs], ["MyType.DoSomething"]) self.assertEqual(funcs[0]["loc"], 6) def test_kotlin_function_name_and_property_excluded(self) -> None: """Kotlin properties are not functions; only fun declarations count.""" source = textwrap.dedent("""\ package com.example fun calculate(a: Int, b: Int): Int { return a + b } val constant = 42 var mutable = 0 """) names = {f["name"] for f in ra._extract_functions_kotlin(source, "test.kt")} self.assertEqual(names, {"calculate"}) def test_kotlin_multiline_raw_string_and_scope_identity(self) -> None: source = textwrap.dedent("""\ class Repo { fun query(): String { val sql = \"\"\" SELECT } FROM t \"\"\" return sql } } """) funcs = ra._extract_functions_kotlin(source, "test.kt") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "Repo.query") self.assertEqual(funcs[0]["loc"], 6) def test_swift_function_name_and_property_excluded(self) -> None: source = textwrap.dedent("""\ func fetchData() { print("fetching") } var counter = 0 let limit = 10 """) names = {f["name"] for f in ra._extract_functions_swift(source, "test.swift")} self.assertEqual(names, {"fetchData"}) def test_swift_multiline_string_and_extension_identity(self) -> None: source = textwrap.dedent("""\ extension Foo { func render() -> String { let text = \"\"\" } not a brace \"\"\" return text } } """) funcs = ra._extract_functions_swift(source, "test.swift") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "Foo.render") self.assertEqual(funcs[0]["loc"], 6) def test_shell_function_name(self) -> None: source = textwrap.dedent("""\ #!/bin/sh my_function() { echo "hello" } another_function() { echo "world" } """) names = {f["name"] for f in ra._extract_functions_shell(source, "test.sh")} self.assertIn("my_function", names) self.assertIn("another_function", names) def test_shell_heredoc_braces_ignored(self) -> None: source = textwrap.dedent("""\ generate_config() { cat < None: """Colliding overloads split by signature; unique callables keep scope.name.""" source = textwrap.dedent("""\ class C { fun run(value: Int) { println(value) } fun run(value: String) { println(value) } fun once() { println(1) } } """) names = {f["name"] for f in ra._extract_functions_kotlin(source, "t.kt")} self.assertEqual(names, {"C.run(value: Int)", "C.run(value: String)", "C.once"}) def test_swift_overload_identity_is_signature_qualified(self) -> None: source = textwrap.dedent("""\ struct C { func run(value: Int) { print(value) } func run(value: String) { print(value) } } """) names = {f["name"] for f in ra._extract_functions_swift(source, "t.swift")} self.assertEqual(names, {"C.run(value: Int)", "C.run(value: String)"}) def test_dart_overload_identity_is_signature_qualified(self) -> None: source = textwrap.dedent("""\ class C { void run(int value) { print(value); } void run(String value) { print(value); } } """) names = {f["name"] for f in ra._extract_functions_dart(source, "t.dart")} self.assertEqual(names, {"C.run(int value)", "C.run(String value)"}) def test_go_receiver_overload_identity_excludes_receiver_clause(self) -> None: """Same method name on two receivers stays distinct without a signature.""" source = textwrap.dedent("""\ package main func (a *A) Run(x int) { _ = x } func (b *B) Run(x string) { _ = x } """) names = {f["name"] for f in ra._extract_functions_go(source, "t.go")} self.assertEqual(names, {"A.Run", "B.Run"}) def test_overload_identity_is_deterministic(self) -> None: source = textwrap.dedent("""\ class C { fun run(value: Int) { println(value) } fun run(value: String) { println(value) } } """) rendered = {json.dumps(ra._extract_functions_kotlin(source, "t.kt"), sort_keys=True) for _ in range(3)} self.assertEqual(len(rendered), 1) def test_python_property_and_setter_identity_is_split(self) -> None: """A property/setter pair shares scope.name and must not collide.""" source = textwrap.dedent("""\ class C: @property def value(self): return self._v @value.setter def value(self, v): self._v = v """) names = {f["name"] for f in ra._extract_functions_python(source, "t.py")} self.assertEqual(names, {"C.value(self)", "C.value(self, v)"}) def test_identical_signature_collision_gets_stable_ordinal(self) -> None: """When signatures tie too, position breaks the tie deterministically.""" source = textwrap.dedent("""\ class C: if FLAG: def run(self): return 1 else: def run(self): return 2 """) funcs = ra._extract_functions_python(source, "t.py") self.assertEqual(sorted(f["name"] for f in funcs), ["C.run(self)#1", "C.run(self)#2"]) first = next(f for f in funcs if f["name"] == "C.run(self)#1") second = next(f for f in funcs if f["name"] == "C.run(self)#2") self.assertLess(first["start"], second["start"]) def test_extracted_functions_expose_no_internal_keys(self) -> None: """The signature is an identity input, not part of the report schema.""" source = "class C {\n fun run(a: Int) {\n println(a)\n }\n}\n" for fn in ra._extract_functions_kotlin(source, "t.kt"): self.assertEqual(set(fn), {"name", "loc", "start", "end"}) def test_python_scope_qualified_identity(self) -> None: """Python classes are not functions and methods carry their class scope.""" source = textwrap.dedent("""\ class Foo: def bar(self): pass def baz(self): return 1 def standalone(): x = 1 y = 2 """) names = {f["name"] for f in ra._extract_functions_python(source, "test.py")} self.assertNotIn("Foo", names) self.assertEqual(names, {"Foo.bar", "Foo.baz", "standalone"}) def test_python_duplicate_method_names_stay_distinct(self) -> None: source = textwrap.dedent("""\ class A: def run(self): pass class B: def run(self): x = 1 return x def run(): pass """) names = {f["name"] for f in ra._extract_functions_python(source, "test.py")} self.assertEqual(names, {"A.run", "B.run", "run"}) def test_python_string_braces_ignored(self) -> None: source = textwrap.dedent("""\ def with_brace(): s = "hello } world" return s """) funcs = ra._extract_functions_python(source, "test.py") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "with_brace") def test_kotlin_multiline_expression_body_spans_full_expression(self) -> None: """An `=` body ends at the expression, not at the declaration line.""" source = textwrap.dedent("""\ class C { fun compute() = listOf( 1, 2, ) fun five() = 5 fun call() = compute() .size } """) by_name = {f["name"]: f for f in ra._extract_functions_kotlin(source, "test.kt")} self.assertEqual(set(by_name), {"C.compute", "C.five", "C.call"}) self.assertEqual(by_name["C.compute"]["start"], 2) self.assertEqual(by_name["C.compute"]["end"], 6) self.assertEqual(by_name["C.compute"]["loc"], 5) # A one-line expression body stays exactly one line. self.assertEqual(by_name["C.five"]["start"], 8) self.assertEqual(by_name["C.five"]["end"], 8) self.assertEqual(by_name["C.five"]["loc"], 1) # A leading chain line continues the declaration-line expression. self.assertEqual(by_name["C.call"]["start"], 10) self.assertEqual(by_name["C.call"]["end"], 11) self.assertEqual(by_name["C.call"]["loc"], 2) def test_kotlin_depth_zero_expression_continuations_span_full_expression(self) -> None: """Depth-0 chain and operator continuations span to the real end line.""" source = textwrap.dedent("""\ class C { fun call() = compute() .size fun total() = 1 + 2 fun done() = 3 } """) by_name = {f["name"]: f for f in ra._extract_functions_kotlin(source, "test.kt")} self.assertEqual(set(by_name), {"C.call", "C.total", "C.done"}) # Leading `.size` chain belongs to the declaration-line expression. self.assertEqual(by_name["C.call"]["start"], 2) self.assertEqual(by_name["C.call"]["end"], 3) self.assertEqual(by_name["C.call"]["loc"], 2) # A trailing binary operator keeps the expression open one more line. self.assertEqual(by_name["C.total"]["start"], 4) self.assertEqual(by_name["C.total"]["end"], 5) self.assertEqual(by_name["C.total"]["loc"], 2) # The sibling after a continuation is never swallowed. self.assertEqual(by_name["C.done"]["start"], 6) self.assertEqual(by_name["C.done"]["end"], 6) self.assertEqual(by_name["C.done"]["loc"], 1) def test_kotlin_depth_zero_operator_grammar_continuations_span_full_expression(self) -> None: """Comparison/word/range/infix and leading operators span to the end line.""" source = textwrap.dedent("""\ class C { infix fun Int.join(other: Int) = this + other infix fun String.then(other: Int) = other fun less() = 1 < 2 fun lessNoSpace() = 1< 2 fun greaterNoSpace() = 2> 1 fun contained() = 1 in listOf(1) fun typed(value: Any) = value is String fun cast(value: Any) = value as String fun range() = 1..< 3 fun custom() = 1 join 2 fun stringInfix() = "x" then 1 fun leadingLogical() = true && false fun leadingCast(value: Any) = value as String fun genericCast(value: Any) = value as List fun genericTyped(value: Any) = value is Map> fun plain(value: Any) = value fun done() = 3 } """) by_name = {f["name"]: f for f in ra._extract_functions_kotlin(source, "test.kt")} self.assertEqual(set(by_name), { "C.Int.join", "C.String.then", "C.less", "C.lessNoSpace", "C.greaterNoSpace", "C.contained", "C.typed", "C.cast", "C.range", "C.custom", "C.stringInfix", "C.leadingLogical", "C.leadingCast", "C.genericCast", "C.genericTyped", "C.plain", "C.done", }) # Trailing comparison (with or without lexical spacing), containment/ # type test/cast, open range, and custom infix with identifier or # string literal left operand, plus newline-before logical/cast # operators, each keep the expression open exactly one more line. expected_spans = { "C.less": (4, 5), "C.lessNoSpace": (6, 7), "C.greaterNoSpace": (8, 9), "C.contained": (10, 11), "C.typed": (12, 13), "C.cast": (14, 15), "C.range": (16, 17), "C.custom": (18, 19), "C.stringInfix": (20, 21), "C.leadingLogical": (22, 23), "C.leadingCast": (24, 25), } for name, (start, end) in expected_spans.items(): self.assertEqual(by_name[name]["start"], start, name) self.assertEqual(by_name[name]["end"], end, name) self.assertEqual(by_name[name]["loc"], 2, name) # Complete infix/generic/simple bodies stay one line, and no sibling # after either a generic closer or comparison is swallowed. expected_single_line_spans = { "C.Int.join": 2, "C.String.then": 3, "C.genericCast": 26, "C.genericTyped": 27, "C.plain": 28, "C.done": 29, } for name, line in expected_single_line_spans.items(): actual = (by_name[name]["start"], by_name[name]["end"], by_name[name]["loc"]) self.assertEqual(actual, (line, line, 1), name) def test_kotlin_generic_type_projection_closers_preserve_siblings(self) -> None: """Complete Kotlin type projections do not turn their `>` into comparisons.""" source = textwrap.dedent("""\ class C { fun functionType(value: Any) = value as List<(Int) -> String> fun parenthesized(value: Any) = value as List<(String)> fun annotated(value: Any) = value as List<@Ann String> fun nonNull(value: Any) = value as List fun escaped(value: Any) = value as List<`when`> fun genericComparison(value: Any) = value as List > value fun done() = 3 } """) by_name = {f["name"]: f for f in ra._extract_functions_kotlin(source, "test.kt")} expected = { "C.functionType": (2, 2, 1), "C.parenthesized": (3, 3, 1), "C.annotated": (4, 4, 1), "C.nonNull": (5, 5, 1), "C.escaped": (6, 6, 1), "C.genericComparison": (7, 8, 2), "C.done": (9, 9, 1), } self.assertEqual(set(by_name), set(expected)) for name, span in expected.items(): actual = (by_name[name]["start"], by_name[name]["end"], by_name[name]["loc"]) self.assertEqual(actual, span, name) def test_kotlin_qualified_annotation_type_projection_closer_preserves_siblings(self) -> None: """Qualified annotations keep their outer generic closer local.""" source = textwrap.dedent("""\ class C { fun qualified(value: Any) = value as @com.example.Ann("reason") List fun safeCast(value: Any) = value as?@com.example.Ann("reason") List fun typed(value: Any) = value is @com.example.Ann("reason") List fun simple(value: Any) = value as @com.example.Ann List fun simpleSafeCast(value: Any) = value as?@com.example.Ann List fun simpleTyped(value: Any) = value is @com.example.Ann List fun simpleMultiple(value: Any) = value as @Ann @com.example.Other List fun constructorThenSimple(value: Any) = value as @Ann("reason") @com.example.Other List fun simpleThenConstructor(value: Any) = value as @Ann @com.example.Other("reason") List fun constructorMultiple(value: Any) = value as @Ann("one") @com.example.Other("two") List fun annotationArgument(value: Any) = value as List<@com.example.Ann([value as String, 1 > 0]) String> fun afterOperator(value: Any) = value as @com.example.Ann List fun afterAnnotation(value: Any) = value as? @com.example.Ann List fun betweenAnnotations(value: Any) = value is @Ann @com.example.Other List fun afterConstructor(value: Any) = value !is @com.example.Ann("reason") List fun mixedMultiline(value: Any) = value as @Ann @com.example.Other("reason") List fun done() = 3 } """) by_name = {f["name"]: f for f in ra._extract_functions_kotlin(source, "test.kt")} expected = { "C.qualified": (2, 2, 1), "C.safeCast": (3, 3, 1), "C.typed": (4, 4, 1), "C.simple": (5, 5, 1), "C.simpleSafeCast": (6, 6, 1), "C.simpleTyped": (7, 7, 1), "C.simpleMultiple": (8, 8, 1), "C.constructorThenSimple": (9, 9, 1), "C.simpleThenConstructor": (10, 10, 1), "C.constructorMultiple": (11, 11, 1), "C.annotationArgument": (12, 12, 1), "C.afterOperator": (13, 14, 2), "C.afterAnnotation": (15, 16, 2), "C.betweenAnnotations": (17, 18, 2), "C.afterConstructor": (19, 20, 2), "C.mixedMultiline": (21, 23, 3), "C.done": (24, 24, 1), } self.assertEqual(set(by_name), set(expected)) for name, span in expected.items(): actual = (by_name[name]["start"], by_name[name]["end"], by_name[name]["loc"]) self.assertEqual(actual, span, name) def test_kotlin_adjacent_safe_cast_generic_closer_preserves_sibling(self) -> None: """`as?` accepts an adjacent type name but identifier tails do not.""" source = textwrap.dedent("""\ class C { fun adjacent(value: Any) = value as?List fun spaced(value: Any) = value as? List fun done() = 3 } """) by_name = {f["name"]: f for f in ra._extract_functions_kotlin(source, "test.kt")} expected = { "C.adjacent": (2, 2, 1), "C.spaced": (3, 3, 1), "C.done": (4, 4, 1), } self.assertEqual(set(by_name), set(expected)) for name, span in expected.items(): actual = (by_name[name]["start"], by_name[name]["end"], by_name[name]["loc"]) self.assertEqual(actual, span, name) self.assertFalse(ra._kotlin_trailing_generic_type_closer("value asList")) self.assertFalse(ra._kotlin_trailing_generic_type_closer("value isReady")) def test_kotlin_multiline_generic_type_arguments_preserve_span_and_siblings(self) -> None: """Kotlin generic arguments stay open through their outer closer.""" source = textwrap.dedent("""\ class C { fun simple(value: Any) = value as List< Int > fun nested(value: Any) = value as Map< String, List > fun done() = 3 } """) by_name = {f["name"]: f for f in ra._extract_functions_kotlin(source, "test.kt")} expected = { "C.simple": (2, 4, 3), "C.nested": (5, 8, 4), "C.done": (9, 9, 1), } self.assertEqual(set(by_name), set(expected)) for name, span in expected.items(): actual = (by_name[name]["start"], by_name[name]["end"], by_name[name]["loc"]) self.assertEqual(actual, span, name) def test_shell_escaped_and_operator_comment_braces_are_ignored(self) -> None: """`:;# }` is a comment and `\\}` is a literal: neither closes a body.""" source = textwrap.dedent("""\ close_brace() { :;# } echo done } escaped() { echo \\} echo done } """) by_name = {f["name"]: f for f in ra._extract_functions_shell(source, "test.sh")} self.assertEqual(set(by_name), {"close_brace", "escaped"}) self.assertEqual(by_name["close_brace"]["start"], 1) self.assertEqual(by_name["close_brace"]["end"], 4) self.assertEqual(by_name["close_brace"]["loc"], 4) self.assertEqual(by_name["escaped"]["start"], 6) self.assertEqual(by_name["escaped"]["end"], 9) self.assertEqual(by_name["escaped"]["loc"], 4) def test_shell_parameter_expansion_and_quotes_still_scan(self) -> None: """Token-aware blanking leaves `${...}` and quoted `#` lexically intact.""" source = textwrap.dedent("""\ expand() { echo "${VAR}" echo "${OTHER:-#default}" local hash='#not-a-comment' echo "$hash" } """) funcs = ra._extract_functions_shell(source, "test.sh") self.assertEqual(len(funcs), 1) self.assertEqual(funcs[0]["name"], "expand") self.assertEqual(funcs[0]["loc"], 6) def test_shell_hash_comment_boundary_preserves_function_close(self) -> None: """A non-comment `#` never erases a same-line function close.""" source = textwrap.dedent("""\ hash_word() { echo foo#bar; } trim() { echo ${value#x}; } next() { echo done } """) by_name = {f["name"]: f for f in ra._extract_functions_shell(source, "test.sh")} self.assertEqual(set(by_name), {"hash_word", "trim", "next"}) # `#` inside a word is a token, so `; }` still closes the body. self.assertEqual(by_name["hash_word"]["start"], 1) self.assertEqual(by_name["hash_word"]["end"], 2) self.assertEqual(by_name["hash_word"]["loc"], 2) # `#` inside `${...}` is an expansion operator, not a comment. self.assertEqual(by_name["trim"]["start"], 3) self.assertEqual(by_name["trim"]["end"], 4) self.assertEqual(by_name["trim"]["loc"], 2) # The sibling after the same-line closes keeps its own identity. self.assertEqual(by_name["next"]["start"], 5) self.assertEqual(by_name["next"]["end"], 7) self.assertEqual(by_name["next"]["loc"], 3) def test_go_type_braces_do_not_open_function_body(self) -> None: """Generic constraint and result-type braces are not the body opener.""" source = textwrap.dedent("""\ package p func Map[T interface{ ~int | ~string }](in []T) []T { return in } func Meta() (out struct{ A int }) { return out } """) by_name = {f["name"]: f for f in ra._extract_functions_go(source, "test.go")} self.assertEqual(set(by_name), {"Map", "Meta"}) # The generic parameter list is not the signature; the value list is. lines = source.splitlines() self.assertEqual( ra._declaration_signature(lines, 2, lines[2].index("[")), "(in []T)") self.assertEqual( ra._declaration_signature(lines, 6, lines[6].index("(")), "()") self.assertEqual(by_name["Map"]["start"], 3) self.assertEqual(by_name["Map"]["end"], 5) self.assertEqual(by_name["Map"]["loc"], 3) self.assertEqual(by_name["Meta"]["start"], 7) self.assertEqual(by_name["Meta"]["end"], 9) self.assertEqual(by_name["Meta"]["loc"], 3) # ============================================================ # TestAllowlistIdentityAndReasonRatchet [REVIEW_REVIEW_REFACTOR-2] # ============================================================ def _baseline(file_thresholds=None, function_thresholds=None, task_totals=None): return { "file_thresholds": file_thresholds or [], "function_thresholds": function_thresholds or [], "task_read_set_totals": task_totals or [], } def _current(violations): return { "violations": violations, "summary": {"total_files": 1, "total_lines_of_code": 1000, "total_functions": 1, "violations_count": len(violations)}, } class TestAllowlistIdentityAndReasonRatchet(_RepoFixture): """Stable (path, metric, function) identity: drops pass, new/increases fail.""" prefix = "ra-test-ratchet-" def test_report_function_identities_are_unique_per_path(self) -> None: """Ratchet identity is (path, metric, function): duplicates would collide.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("src/overload.kt", textwrap.dedent("""\ class C { fun run(value: Int) { println(value) } fun run(value: String) { println(value) } } """)) self._write("src/prop.py", textwrap.dedent("""\ class C: @property def value(self): return self._v @value.setter def value(self, v): self._v = v """)) result = ra.audit(["src/overload.kt", "src/prop.py"]) identities = [(entry["path"], fn["name"]) for entry in result["files"] for fn in entry.get("functions", [])] self.assertEqual(len(identities), 4) self.assertEqual(len(identities), len(set(identities))) finally: os.chdir(old_cwd) def test_same_value_with_identity_passes(self) -> None: baseline = _baseline( file_thresholds=[ {"path": "src/big.go", "metric": "file_loc", "level": "exception", "value": 1000, "reason": "existing large file"}, ], function_thresholds=[ {"path": "src/big.go", "function": "MyType.DoSomething", "metric": "function_loc", "level": "split_review", "value": 120, "reason": "existing large function"}, ], ) current = _current([ {"path": "src/big.go", "metric": "file_loc", "level": "exception", "value": 1000, "reason": "file exceeds threshold"}, {"path": "src/big.go", "metric": "function_loc", "level": "split_review", "value": 120, "function": "MyType.DoSomething", "reason": "function exceeds threshold"}, ]) self.assertEqual(ra.ratchet_check(current, baseline, None), []) def test_new_function_identity_fails(self) -> None: current = _current([ {"path": "src/new.go", "metric": "function_loc", "level": "split_review", "value": 150, "function": "NewFunc", "reason": "new oversized function"}, ]) new = ra.ratchet_check(current, _baseline(), None) self.assertEqual(len(new), 1) self.assertEqual(new[0]["function"], "NewFunc") self.assertIn("new violation", new[0]["reason"]) def test_level_drop_passes(self) -> None: """An improvement that lowers the level is not a new violation.""" baseline = _baseline(file_thresholds=[ {"path": "src/big.go", "metric": "file_loc", "level": "exception", "value": 1200, "reason": "existing large file"}, ]) current = _current([ {"path": "src/big.go", "metric": "file_loc", "level": "split_review", "value": 900, "reason": "file exceeds threshold"}, ]) self.assertEqual(ra.ratchet_check(current, baseline, None), []) def test_level_increase_fails(self) -> None: baseline = _baseline(file_thresholds=[ {"path": "src/big.go", "metric": "file_loc", "level": "split_review", "value": 900, "reason": "existing large file"}, ]) current = _current([ {"path": "src/big.go", "metric": "file_loc", "level": "exception", "value": 1200, "reason": "file exceeds threshold"}, ]) new = ra.ratchet_check(current, baseline, None) self.assertEqual(len(new), 1) self.assertIn("level increased", new[0]["reason"]) def test_value_increase_fails(self) -> None: baseline = _baseline(file_thresholds=[ {"path": "src/big.go", "metric": "file_loc", "level": "exception", "value": 1100, "reason": "smaller baseline"}, ]) current = _current([ {"path": "src/big.go", "metric": "file_loc", "level": "exception", "value": 1150, "reason": "got bigger"}, ]) new = ra.ratchet_check(current, baseline, None) self.assertEqual(len(new), 1) self.assertIn("value increased", new[0]["reason"]) def test_empty_reason_in_baseline_fails(self) -> None: baseline = _baseline(file_thresholds=[ {"path": "src/big.go", "metric": "file_loc", "level": "exception", "value": 1000, "reason": ""}, ]) new = ra.ratchet_check(_current([]), baseline, None) self.assertEqual(len(new), 1) self.assertEqual(new[0]["path"], "") self.assertIn("reason", new[0]["reason"]) def test_duplicate_function_in_baseline_fails(self) -> None: baseline = _baseline(function_thresholds=[ {"path": "src/fn.go", "function": "DoSomething", "metric": "function_loc", "level": "split_review", "value": 120, "reason": "first entry"}, {"path": "src/fn.go", "function": "DoSomething", "metric": "function_loc", "level": "split_review", "value": 120, "reason": "duplicate entry"}, ]) new = ra.ratchet_check(_current([]), baseline, None) self.assertEqual(new[0]["path"], "") self.assertIn("duplicate identity", new[0]["reason"]) def test_same_identity_across_levels_is_duplicate(self) -> None: """One identity may hold exactly one allowlisted level.""" baseline = _baseline(file_thresholds=[ {"path": "src/fn.go", "metric": "file_loc", "level": "warning", "value": 600, "reason": "entry a"}, {"path": "src/fn.go", "metric": "file_loc", "level": "exception", "value": 1200, "reason": "entry b"}, ]) new = ra.ratchet_check(_current([]), baseline, None) self.assertEqual(new[0]["path"], "") self.assertIn("duplicate identity", new[0]["reason"]) def test_malformed_metric_fails(self) -> None: baseline = _baseline(file_thresholds=[ {"path": "src/big.go", "metric": "function_loc", "level": "exception", "value": 1000, "reason": "wrong metric"}, ]) new = ra.ratchet_check(_current([]), baseline, None) self.assertIn("invalid 'metric'", new[0]["reason"]) def test_malformed_level_fails(self) -> None: baseline = _baseline(function_thresholds=[ {"path": "src/big.go", "function": "Foo", "metric": "function_loc", "level": "exception", "value": 130, "reason": "functions have no exception level"}, ]) new = ra.ratchet_check(_current([]), baseline, None) self.assertIn("invalid 'level'", new[0]["reason"]) def test_malformed_value_fails(self) -> None: baseline = _baseline(file_thresholds=[ {"path": "src/big.go", "metric": "file_loc", "level": "exception", "value": "1000", "reason": "string value"}, ]) new = ra.ratchet_check(_current([]), baseline, None) self.assertIn("invalid 'value'", new[0]["reason"]) def test_unknown_key_fails(self) -> None: baseline = _baseline(file_thresholds=[ {"path": "src/big.go", "metric": "file_loc", "level": "exception", "value": 1000, "reason": "ok", "owner": "team"}, ]) new = ra.ratchet_check(_current([]), baseline, None) self.assertIn("unknown keys", new[0]["reason"]) def test_missing_function_key_fails(self) -> None: baseline = _baseline(function_thresholds=[ {"path": "src/big.go", "metric": "function_loc", "level": "split_review", "value": 130, "reason": "no function identity"}, ]) new = ra.ratchet_check(_current([]), baseline, None) self.assertIn("missing keys", new[0]["reason"]) # ============================================================ # TestTaskReadSetTotalsAndBudget [REVIEW_REVIEW_REFACTOR-3] # ============================================================ class TestTaskReadSetTotalsAndBudget(_RepoFixture): """Read-set config failures, budget/reason, and baseline total ratchet.""" prefix = "ra-test-rs-" def _config(self, files, max_total_loc=100, reason="test budget", task_id="test-task", version="2.0", generated_at="2026-07-16", description="fixture task"): return {"version": version, "generated_at": generated_at, "tasks": [{ "task_id": task_id, "description": description, "files": files, "budget": {"max_total_loc": max_total_loc, "reason": reason}, }]} def test_ordered_path_totals(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("a.go", "package p\n\nfunc main() {}\n") # 3 lines self._write("b.py", "def f(): pass\n") # 1 line self._write("c.md", "line1\nline2\n") # 2 lines errors, task_read_sets = ra.audit_read_sets( None, self._config(["b.py", "a.go", "c.md"])) self.assertEqual(errors, []) self.assertEqual(len(task_read_sets), 1) entry = task_read_sets[0] self.assertEqual(entry["task_id"], "test-task") self.assertEqual([f["path"] for f in entry["files"]], ["b.py", "a.go", "c.md"]) self.assertEqual(entry["total_loc"], 6) finally: os.chdir(old_cwd) def test_missing_path_is_configuration_failure(self) -> None: """A read-set path that does not exist is a configuration failure, not 0 LOC.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("exists.go", "package p\n") errors, task_read_sets = ra.audit_read_sets( None, self._config(["exists.go", "nonexistent.go"])) self.assertEqual(len(errors), 1) self.assertEqual(errors[0]["metric"], "read_set_config") self.assertIn("missing read_set path 'nonexistent.go'", errors[0]["reason"]) self.assertEqual([f["path"] for f in task_read_sets[0]["files"]], ["exists.go"]) finally: os.chdir(old_cwd) def test_duplicate_path_error(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("a.go", "package p\n") errors, _ = ra.audit_read_sets(None, self._config(["a.go", "a.go"])) self.assertEqual(len(errors), 1) self.assertIn("duplicate", errors[0]["reason"]) finally: os.chdir(old_cwd) def test_duplicate_task_id_error(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("a.go", "package p\n") config = {"version": "2.0", "generated_at": "2026-07-16", "tasks": [ {"task_id": "dup", "description": "fixture task", "files": ["a.go"], "budget": {"max_total_loc": 10, "reason": "r"}}, {"task_id": "dup", "description": "fixture task", "files": ["a.go"], "budget": {"max_total_loc": 10, "reason": "r"}}, ]} errors, task_read_sets = ra.audit_read_sets(None, config) self.assertEqual(len(errors), 1) self.assertIn("duplicate task_id", errors[0]["reason"]) self.assertEqual(len(task_read_sets), 1) finally: os.chdir(old_cwd) def test_read_set_schema_requires_typed_metadata(self) -> None: """Untyped/absent root metadata and task description are config errors.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("a.go", "package p\n") task = {"task_id": "t", "description": "fixture task", "files": ["a.go"], "budget": {"max_total_loc": 100, "reason": "r"}} valid = {"version": "2.0", "generated_at": "2026-07-16", "tasks": [task]} # The contrast case: the exact schema is satisfied, so no issue. errors, task_read_sets = ra.audit_read_sets(None, valid) self.assertEqual(errors, []) self.assertEqual(len(task_read_sets), 1) cases = [ ("missing version", {k: v for k, v in valid.items() if k != "version"}, "missing required root key 'version'"), ("missing generated_at", {k: v for k, v in valid.items() if k != "generated_at"}, "missing required root key 'generated_at'"), ("wrong-type version", {**valid, "version": 2.0}, "root 'version' must be a non-empty string"), ("empty generated_at", {**valid, "generated_at": " "}, "root 'generated_at' must be a non-empty string"), ("unsupported version", {**valid, "version": "1.0"}, "unsupported version '1.0'"), ("missing description", {**valid, "tasks": [{k: v for k, v in task.items() if k != "description"}]}, "missing required non-empty 'description'"), ("wrong-type description", {**valid, "tasks": [{**task, "description": ["fixture"]}]}, "missing required non-empty 'description'"), ("empty description", {**valid, "tasks": [{**task, "description": ""}]}, "missing required non-empty 'description'"), ] for label, config, expected in cases: with self.subTest(case=label): errors, _ = ra.audit_read_sets(None, config) self.assertEqual(len(errors), 1, f"{label}: {errors}") self.assertEqual(errors[0]["metric"], "read_set_config") self.assertIn(expected, errors[0]["reason"]) finally: os.chdir(old_cwd) def test_invalid_budget_schema_error(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("a.go", "package p\n") config = {"version": "2.0", "generated_at": "2026-07-16", "tasks": [ {"task_id": "t", "description": "fixture task", "files": ["a.go"], "budget": {"max_total_loc": "100", "reason": "r"}}]} errors, _ = ra.audit_read_sets(None, config) self.assertEqual(len(errors), 1) self.assertIn("invalid 'max_total_loc'", errors[0]["reason"]) finally: os.chdir(old_cwd) def test_unknown_task_key_error(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("a.go", "package p\n") config = {"version": "2.0", "generated_at": "2026-07-16", "tasks": [ {"task_id": "t", "description": "fixture task", "files": ["a.go"], "total_loc": 1, "budget": {"max_total_loc": 100, "reason": "r"}}]} errors, _ = ra.audit_read_sets(None, config) self.assertEqual(len(errors), 1) self.assertIn("unknown keys ['total_loc']", errors[0]["reason"]) finally: os.chdir(old_cwd) def test_exact_budget_passes_and_one_over_without_reason_fails(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("ten.py", "\n".join(f"# {i}" for i in range(10)) + "\n") errors, task_read_sets = ra.audit_read_sets( None, self._config(["ten.py"], max_total_loc=10, reason="")) self.assertEqual(errors, []) self.assertEqual(task_read_sets[0]["total_loc"], 10) errors, _ = ra.audit_read_sets( None, self._config(["ten.py"], max_total_loc=9, reason="")) self.assertEqual(len(errors), 1) self.assertEqual(errors[0]["metric"], "read_set_budget") self.assertEqual(errors[0]["value"], 10) self.assertIn("empty reason", errors[0]["reason"]) finally: os.chdir(old_cwd) def test_over_budget_with_reason_needs_baseline_total(self) -> None: """Over budget with a reason still has to be allowlisted in the baseline.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("big.py", "\n".join(f"# {i}" for i in range(200)) + "\n") config = self._config(["big.py"], max_total_loc=100, reason="known large file") errors, task_read_sets = ra.audit_read_sets(None, config) self.assertEqual(errors, []) self.assertEqual(task_read_sets[0]["total_loc"], 200) issues = ra.ratchet_read_set_totals(task_read_sets, _baseline()) self.assertEqual(len(issues), 1) self.assertEqual(issues[0]["metric"], "read_set_total") self.assertEqual(issues[0]["value"], 200) self.assertIn("not allowlisted", issues[0]["reason"]) finally: os.chdir(old_cwd) def test_task_total_same_or_lower_than_baseline_passes(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("big.py", "\n".join(f"# {i}" for i in range(200)) + "\n") config = self._config(["big.py"], max_total_loc=100, reason="known large file") _, task_read_sets = ra.audit_read_sets(None, config) same = _baseline(task_totals=[{"task_id": "test-task", "metric": "read_set_total", "value": 200, "reason": "allowlisted total"}]) self.assertEqual(ra.ratchet_read_set_totals(task_read_sets, same), []) higher = _baseline(task_totals=[{"task_id": "test-task", "metric": "read_set_total", "value": 300, "reason": "allowlisted total"}]) self.assertEqual(ra.ratchet_read_set_totals(task_read_sets, higher), []) finally: os.chdir(old_cwd) def test_task_total_increase_fails(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("big.py", "\n".join(f"# {i}" for i in range(200)) + "\n") config = self._config(["big.py"], max_total_loc=100, reason="known large file") _, task_read_sets = ra.audit_read_sets(None, config) baseline = _baseline(task_totals=[{"task_id": "test-task", "metric": "read_set_total", "value": 150, "reason": "allowlisted total"}]) issues = ra.ratchet_read_set_totals(task_read_sets, baseline) self.assertEqual(len(issues), 1) self.assertIn("task total increased from 150 to 200", issues[0]["reason"]) finally: os.chdir(old_cwd) def test_task_total_baseline_schema_validated(self) -> None: baseline = _baseline(task_totals=[{"task_id": "t", "metric": "read_set_total", "value": 10, "reason": ""}]) new = ra.ratchet_check(_current([]), baseline, None) self.assertEqual(new[0]["path"], "") self.assertIn("task_read_set_totals[0]", new[0]["reason"]) def test_ratchet_check_covers_read_set_totals(self) -> None: """When read-sets are supplied, one ratchet_check call covers task totals.""" current = _current([]) current["task_read_sets"] = [{ "task_id": "test-task", "files": [{"path": "big.py", "loc": 200}], "total_loc": 200, "budget": {"max_total_loc": 100, "reason": "known large file"}, "reason": "known large file", }] self.assertEqual(ra.ratchet_check(current, _baseline(), None), []) new = ra.ratchet_check(current, _baseline(), {"tasks": []}) self.assertEqual(len(new), 1) self.assertEqual(new[0]["path"], "") def test_required_config_load_reports_each_failure(self) -> None: """load_required_json_file separates missing, malformed, and valid configs.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: config, error = ra.load_required_json_file("absent.json") self.assertIsNone(config) self.assertIn("required config file not found", error) self._write("broken.json", '{"tasks": [ truncated\n') config, error = ra.load_required_json_file("broken.json") self.assertIsNone(config) self.assertIn("malformed JSON", error) self._write("list.json", "[]\n") config, error = ra.load_required_json_file("list.json") self.assertIsNone(config) self.assertIn("not a JSON object", error) self._write("valid.json", '{"tasks": []}\n') config, error = ra.load_required_json_file("valid.json") self.assertEqual(config, {"tasks": []}) self.assertIsNone(error) finally: os.chdir(old_cwd) def test_absent_config_is_a_configuration_failure(self) -> None: """A read-set config of None must fail rather than report zero issues.""" errors, task_read_sets = ra.audit_read_sets(None, None) self.assertEqual(len(errors), 1) self.assertEqual(errors[0]["metric"], "read_set_config") self.assertEqual(task_read_sets, []) def test_unknown_root_key_error(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("a.go", "package p\n") config = dict(self._config(["a.go"]), task_totals=[]) errors, task_read_sets = ra.audit_read_sets(None, config) self.assertEqual(len(errors), 1) self.assertIn("unknown root keys ['task_totals']", errors[0]["reason"]) self.assertEqual(task_read_sets, []) finally: os.chdir(old_cwd) def test_documented_root_keys_are_accepted(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("a.go", "package p\n") config = dict(self._config(["a.go"]), version="2.0", generated_at="2026-07-16") errors, task_read_sets = ra.audit_read_sets(None, config) self.assertEqual(errors, []) self.assertEqual(len(task_read_sets), 1) finally: os.chdir(old_cwd) def test_read_set_determinism(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("a.go", "package p\nfunc main() {}\n") self._write("b.py", "def f(): pass\n") config = self._config(["b.py", "a.go"]) rendered = set() for _ in range(3): _, task_read_sets = ra.audit_read_sets(None, config) rendered.add(json.dumps(task_read_sets, sort_keys=True)) self.assertEqual(len(rendered), 1) finally: os.chdir(old_cwd) # ============================================================ # TestCliExitCodes [REVIEW_REVIEW_REFACTOR-3, REVIEW_REVIEW_REFACTOR-5] # ============================================================ class TestCliExitCodes(_RepoFixture): """The CLI renders every failure and exits with a defined status.""" prefix = "ra-test-cli-" def setUp(self): super().setUp() self._write("a.go", "package p\n\nfunc main() {}\n") self._write_json("scripts/readability_baseline.json", _baseline()) self._write_json("scripts/readability_read_sets.json", { "version": "2.0", "generated_at": "2026-07-16", "tasks": [{ "task_id": "t", "description": "fixture task", "files": ["a.go"], "budget": {"max_total_loc": 100, "reason": "fixture budget"}, }]}) self._git_init() def _write_json(self, relpath: str, payload) -> None: self._write(relpath, json.dumps(payload, indent=2) + "\n") def _run_cli(self, *args: str) -> subprocess.CompletedProcess: return subprocess.run([sys.executable, AUDIT_SCRIPT, *args], cwd=self.tmpdir, capture_output=True, text=True) def test_clean_check_exits_zero(self) -> None: proc = self._run_cli("--check", "--input-mode", "worktree") self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("RATCHET OK", proc.stderr) self.assertIn("readability-audit:", proc.stdout) def test_missing_read_set_path_exits_two(self) -> None: self._write_json("scripts/readability_read_sets.json", { "version": "2.0", "generated_at": "2026-07-16", "tasks": [{ "task_id": "t", "description": "fixture task", "files": ["a.go", "gone.go"], "budget": {"max_total_loc": 100, "reason": "fixture budget"}, }]}) proc = self._run_cli("--check", "--input-mode", "worktree") self.assertEqual(proc.returncode, 2, proc.stderr) self.assertIn("READ-SET CONFIGURATION FAIL", proc.stderr) self.assertIn("missing read_set path 'gone.go'", proc.stderr) def test_reasonless_over_budget_exits_four(self) -> None: self._write_json("scripts/readability_read_sets.json", { "version": "2.0", "generated_at": "2026-07-16", "tasks": [{ "task_id": "t", "description": "fixture task", "files": ["a.go"], "budget": {"max_total_loc": 1, "reason": ""}, }]}) proc = self._run_cli("--check", "--input-mode", "worktree") self.assertEqual(proc.returncode, 4, proc.stderr) self.assertIn("RATCHET FAIL", proc.stderr) self.assertIn("read_set_budget=3", proc.stderr) def test_unallowlisted_over_budget_total_exits_four(self) -> None: self._write_json("scripts/readability_read_sets.json", { "version": "2.0", "generated_at": "2026-07-16", "tasks": [{ "task_id": "t", "description": "fixture task", "files": ["a.go"], "budget": {"max_total_loc": 1, "reason": "documented"}, }]}) proc = self._run_cli("--check", "--input-mode", "worktree") self.assertEqual(proc.returncode, 4, proc.stderr) self.assertIn(": read_set_total=3", proc.stderr) def test_missing_read_set_config_exits_two(self) -> None: """Deleting the required config must not silently skip the total ratchet.""" os.remove(os.path.join(self.tmpdir, "scripts/readability_read_sets.json")) proc = self._run_cli("--check", "--input-mode", "worktree") self.assertEqual(proc.returncode, 2, proc.stderr) self.assertIn("READ-SET CONFIGURATION FAIL", proc.stderr) self.assertIn("required config file not found", proc.stderr) self.assertNotIn("RATCHET OK", proc.stderr) def test_malformed_read_set_config_exits_two(self) -> None: self._write("scripts/readability_read_sets.json", '{"tasks": [ truncated\n') proc = self._run_cli("--check", "--input-mode", "worktree") self.assertEqual(proc.returncode, 2, proc.stderr) self.assertIn("READ-SET CONFIGURATION FAIL", proc.stderr) self.assertIn("malformed JSON", proc.stderr) self.assertNotIn("RATCHET OK", proc.stderr) def test_non_object_read_set_config_root_exits_two(self) -> None: self._write_json("scripts/readability_read_sets.json", [{"task_id": "t"}]) proc = self._run_cli("--check", "--input-mode", "worktree") self.assertEqual(proc.returncode, 2, proc.stderr) self.assertIn("is not a JSON object", proc.stderr) def test_unknown_read_set_root_key_exits_two(self) -> None: self._write_json("scripts/readability_read_sets.json", { "tasks": [{"task_id": "t", "files": ["a.go"], "budget": {"max_total_loc": 100, "reason": "fixture budget"}}], "task_totals": [], }) proc = self._run_cli("--check", "--input-mode", "worktree") self.assertEqual(proc.returncode, 2, proc.stderr) self.assertIn("unknown root keys ['task_totals']", proc.stderr) def test_valid_read_set_config_reaches_total_ratchet(self) -> None: """The contrast case: a valid config is always ratcheted, and can fail.""" self._write_json("scripts/readability_read_sets.json", { "version": "2.0", "generated_at": "2026-07-16", "tasks": [{"task_id": "t", "description": "fixture task", "files": ["a.go"], "budget": {"max_total_loc": 1, "reason": "documented"}}], }) proc = self._run_cli("--check", "--input-mode", "worktree") self.assertEqual(proc.returncode, 4, proc.stderr) self.assertIn(": read_set_total=3", proc.stderr) def test_read_set_schema_error_exits_two(self) -> None: """Every schema error is a configuration failure, not a silent pass.""" valid = { "version": "2.0", "generated_at": "2026-07-16", "tasks": [{"task_id": "t", "description": "fixture task", "files": ["a.go"], "budget": {"max_total_loc": 100, "reason": "fixture budget"}}], } cases = [ ("missing version", {k: v for k, v in valid.items() if k != "version"}, "missing required root key 'version'"), ("wrong-type generated_at", {**valid, "generated_at": 20260716}, "root 'generated_at' must be a non-empty string"), ("unsupported version", {**valid, "version": "3.0"}, "unsupported version '3.0'"), ("missing description", {**valid, "tasks": [ {"task_id": "t", "files": ["a.go"], "budget": {"max_total_loc": 100, "reason": "fixture budget"}}]}, "missing required non-empty 'description'"), ("wrong-type description", {**valid, "tasks": [ {"task_id": "t", "description": 7, "files": ["a.go"], "budget": {"max_total_loc": 100, "reason": "fixture budget"}}]}, "missing required non-empty 'description'"), ] for label, config, expected in cases: with self.subTest(case=label): self._write_json("scripts/readability_read_sets.json", config) proc = self._run_cli("--check", "--input-mode", "worktree") self.assertEqual(proc.returncode, 2, proc.stderr) self.assertIn("READ-SET CONFIGURATION FAIL:", proc.stderr) self.assertIn(expected, proc.stderr) def test_canonical_read_set_config_satisfies_exact_schema(self) -> None: """The tracked read-set config is itself a valid exact-schema config.""" with open(os.path.join(SCRIPTS_DIR, "readability_read_sets.json"), encoding="utf-8") as handle: config = json.load(handle) self._write_json("scripts/readability_read_sets.json", { **config, "tasks": [{**task, "files": ["a.go"]} for task in config["tasks"]], }) proc = self._run_cli("--input-mode", "worktree") self.assertEqual(proc.returncode, 0, proc.stderr) self.assertNotIn("READ-SET CONFIGURATION FAIL:", proc.stderr) def test_missing_baseline_exits_three(self) -> None: os.remove(os.path.join(self.tmpdir, "scripts/readability_baseline.json")) proc = self._run_cli("--check", "--input-mode", "worktree") self.assertEqual(proc.returncode, 3, proc.stderr) self.assertIn("baseline not found", proc.stderr) def test_report_is_written_even_when_ratchet_fails(self) -> None: """The projection is evidence, so it is written before the ratchet verdict.""" self._write_json("scripts/readability_read_sets.json", { "version": "2.0", "generated_at": "2026-07-16", "tasks": [{ "task_id": "t", "description": "fixture task", "files": ["a.go"], "budget": {"max_total_loc": 1, "reason": ""}, }]}) out = os.path.join(self.tmpdir, "out", "report.json") proc = self._run_cli("--check", "--input-mode", "worktree", "--output", out) self.assertEqual(proc.returncode, 4, proc.stderr) with open(out, encoding="utf-8") as handle: report = json.load(handle) self.assertEqual([e["path"] for e in report["files"]], ["a.go"]) self.assertEqual(report["input_mode"], "worktree") # ============================================================ # TestInputModeTrackedVsWorktree [REVIEW_REVIEW_REFACTOR-1] # ============================================================ class TestInputModeTrackedVsWorktree(_RepoFixture): """worktree is the post-commit projection: modified files stay, deletions go.""" prefix = "ra-test-mode-" def setUp(self): super().setUp() self._write("tracked.go", "package p\nfunc main() {}\n") self._write("other.go", "package p\n") self._git_init() def test_tracked_mode_only_index(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("untracked.go", "package q\n") files = ra.collect_files("tracked") self.assertIn("tracked.go", files) self.assertNotIn("untracked.go", files) finally: os.chdir(old_cwd) def test_worktree_mode_includes_untracked(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("untracked.go", "package q\n") files = ra.collect_files("worktree") self.assertIn("tracked.go", files) self.assertIn("untracked.go", files) finally: os.chdir(old_cwd) def test_worktree_mode_includes_modified_tracked(self) -> None: """A modified tracked file is still part of the projection.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("tracked.go", "package p\nfunc main() {}\nfunc extra() {}\n") self._write("other.go", "package p\nfunc staged() {}\n") self._git("add", "other.go") files = ra.collect_files("worktree") self.assertIn("tracked.go", files) self.assertIn("other.go", files) finally: os.chdir(old_cwd) def test_worktree_mode_excludes_unstaged_deletion(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: os.remove(os.path.join(self.tmpdir, "other.go")) files = ra.collect_files("worktree") self.assertIn("tracked.go", files) self.assertNotIn("other.go", files) # tracked mode still reports the index entry self.assertIn("other.go", ra.collect_files("tracked")) finally: os.chdir(old_cwd) def test_worktree_mode_excludes_staged_deletion(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._git("rm", "-q", "other.go") files = ra.collect_files("worktree") self.assertIn("tracked.go", files) self.assertNotIn("other.go", files) finally: os.chdir(old_cwd) def test_worktree_mode_excludes_deleted(self) -> None: """git rm --cached leaves the file on disk but stages its removal.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("todelete.go", "package r\n") self._git("add", "todelete.go") self._git("commit", "-q", "-m", "add") self._git("rm", "--cached", "-q", "todelete.go") files = ra.collect_files("worktree") self.assertIn("tracked.go", files) self.assertNotIn("todelete.go", files) finally: os.chdir(old_cwd) def test_worktree_mode_excludes_excluded_patterns(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("proto/gen/generated.go", "package gen\n") self.assertNotIn("proto/gen/generated.go", ra.collect_files("worktree")) finally: os.chdir(old_cwd) def test_tracked_and_worktree_produce_same_for_clean_repo(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self.assertEqual(ra.collect_files("tracked"), ra.collect_files("worktree")) finally: os.chdir(old_cwd) def test_worktree_matches_tracked_after_commit(self) -> None: """The worktree projection equals what tracked reports once committed.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: self._write("tracked.go", "package p\nfunc main() {}\nfunc extra() {}\n") self._write("added.go", "package q\n") os.remove(os.path.join(self.tmpdir, "other.go")) projected = ra.collect_files("worktree") self._git("add", "-A") self._git("commit", "-q", "-m", "apply worktree") self.assertEqual(ra.collect_files("tracked"), projected) self.assertEqual(ra.collect_files("worktree"), projected) finally: os.chdir(old_cwd) # ============================================================ # TestThresholdBoundaries # ============================================================ class TestThresholdBoundaries(_RepoFixture): """Threshold enforcement at boundary values (production defaults).""" def setUp(self): super().setUp() lines_1000 = "\n".join(f"line {i}" for i in range(1000)) + "\n" self._write("src/at_boundary.go", lines_1000) lines_1001 = "\n".join(f"line {i}" for i in range(1001)) + "\n" self._write("src/over_boundary.go", lines_1001) body_118 = "\n".join(f" line{i}" for i in range(118)) self._write("src/fn_at.go", f"func fn_at() {{\n{body_118}\n}}\n") body_119 = "\n".join(f" line{i}" for i in range(119)) self._write("src/fn_over.go", f"func fn_over() {{\n{body_119}\n}}\n") def test_exact_boundary_passes(self) -> None: """1000 LOC production → split_review only; 120 LOC function → warning only.""" old_cwd = os.getcwd() os.chdir(self.tmpdir) try: result = ra.audit(["src/at_boundary.go", "src/fn_at.go"]) violations = result["violations"] self.assertEqual(len(violations), 2) vmap = {(v["path"], v["metric"]): v for v in violations} self.assertEqual(vmap[("src/at_boundary.go", "file_loc")]["level"], "split_review") self.assertEqual(vmap[("src/fn_at.go", "function_loc")]["level"], "warning") finally: os.chdir(old_cwd) def test_one_over_violates(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: result = ra.audit(["src/over_boundary.go", "src/fn_over.go"]) file_violations = [v for v in result["violations"] if v["metric"] == "file_loc"] func_violations = [v for v in result["violations"] if v["metric"] == "function_loc"] self.assertEqual(len(file_violations), 1) self.assertEqual(file_violations[0]["path"], "src/over_boundary.go") self.assertEqual(file_violations[0]["level"], "exception") self.assertEqual(len(func_violations), 1) self.assertEqual(func_violations[0]["path"], "src/fn_over.go") self.assertEqual(func_violations[0]["level"], "split_review") finally: os.chdir(old_cwd) # ============================================================ # TestNewOrIncreasedViolationFails # ============================================================ class TestNewOrIncreasedViolationFails(unittest.TestCase): """Ratchet: new or increased violations should FAIL.""" def test_new_violation_fails(self) -> None: current = _current([ {"path": "src/new_big.go", "metric": "file_loc", "level": "exception", "value": 1500, "reason": "new violation"}, ]) new = ra.ratchet_check(current, _baseline(), None) self.assertEqual(len(new), 1) self.assertEqual(new[0]["path"], "src/new_big.go") self.assertIn("new", new[0]["reason"]) def test_increased_violation_fails(self) -> None: baseline = _baseline(file_thresholds=[ {"path": "src/big.go", "metric": "file_loc", "level": "exception", "value": 1100, "reason": "old"}, ]) current = _current([ {"path": "src/big.go", "metric": "file_loc", "level": "exception", "value": 1200, "reason": "now bigger"}, ]) new = ra.ratchet_check(current, baseline, None) self.assertEqual(len(new), 1) self.assertEqual(new[0]["path"], "src/big.go") self.assertIn("increased", new[0]["reason"]) # ============================================================ # TestReadSetDeterministic # ============================================================ class TestReadSetDeterministic(_RepoFixture): """Read-set (import) extraction is deterministic across runs.""" def setUp(self): super().setUp() self._write("pkg/a.go", textwrap.dedent("""\ package pkg import "fmt" import "io" func A() { fmt.Println("hello") } """)) self._write("pkg/b.py", textwrap.dedent("""\ import os import sys def b_func(): pass """)) self._write("pkg/c.dart", textwrap.dedent("""\ import 'package:foo/bar.dart'; import 'utils.dart'; void c_func() {} """)) def test_read_set_totals_are_deterministic(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: files = ["pkg/a.go", "pkg/b.py", "pkg/c.dart"] results = [ra.audit(files) for _ in range(3)] self.assertEqual(len({json.dumps(r, sort_keys=True) for r in results}), 1) entries = {e["path"]: e for e in results[0]["files"]} self.assertEqual(entries["pkg/a.go"]["read_set"], ["fmt", "io"]) self.assertEqual(entries["pkg/b.py"]["read_set"], ["os", "sys"]) self.assertEqual(len(entries["pkg/c.dart"]["read_set"]), 2) finally: os.chdir(old_cwd) # ============================================================ # TestFunctionExtraction # ============================================================ class TestFunctionExtraction(unittest.TestCase): """Verify function extraction across languages.""" def test_python_function_extraction(self) -> None: source = textwrap.dedent("""\ class Foo: def bar(self): pass def baz(): x = 1 y = 2 return x + y """) funcs = ra._extract_functions_python(source, "test.py") names = {f["name"] for f in funcs} self.assertNotIn("Foo", names) self.assertEqual(names, {"Foo.bar", "baz"}) baz = next(f for f in funcs if f["name"] == "baz") self.assertEqual(baz["loc"], 4) def test_go_function_extraction(self) -> None: source = textwrap.dedent("""\ package main func main() { fmt.Println("hello") } func helper() { x := 1 y := 2 z := 3 return x + y + z } """) names = {f["name"] for f in ra._extract_functions_go(source, "main.go")} self.assertEqual(names, {"main", "helper"}) def test_dart_function_extraction(self) -> None: source = textwrap.dedent("""\ void main() { print('hello'); } int compute(int a, int b) { return a + b; } """) names = {f["name"] for f in ra._extract_functions_dart(source, "test.dart")} self.assertEqual(names, {"main", "compute"}) # ============================================================ # TestJSONDeterminism # ============================================================ class TestJSONDeterminism(_RepoFixture): """Full audit output is deterministic and byte-identical across runs.""" prefix = "ra-test-json-" def setUp(self): super().setUp() self._write("a.go", "package p\n\nfunc main() {}\n") self._write("b.py", "def f(): pass\n") def test_audit_output_deterministic(self) -> None: old_cwd = os.getcwd() os.chdir(self.tmpdir) try: r1 = ra.audit(["a.go", "b.py"]) r2 = ra.audit(["a.go", "b.py"]) self.assertEqual(json.dumps(r1, sort_keys=True), json.dumps(r2, sort_keys=True)) finally: os.chdir(old_cwd) if __name__ == "__main__": unittest.main()