"""Comprehensive tests for ``muse code invariants``. Coverage: I. Unit — check_forbidden_dependency (new rule in plugin engine) II. Unit — check_layer_boundary (new rule in plugin engine) III. Integration — CLI run() with a real repo: all 6 rule types IV. Integration — --commit, --rule, --strict, --json flags V. Integration — exit-code contract VI. Integration — JSON schema validation VII. Integration — no rules file → built-in defaults VIII. Regression — bugs fixed in this review IX. Stress — 100-file repo with deliberate violations """ from __future__ import annotations import hashlib import json import pathlib import pytest from typing import TypedDict from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.invariants import BaseViolation from muse.core._types import Manifest from muse.plugins.code._invariants import ( check_forbidden_dependency, check_layer_boundary, load_invariant_rules, run_invariants, ) class _InvariantsCliJson(TypedDict, total=False): """Shape of the JSON output from ``muse code invariants --json``.""" commit_id: str domain: str branch: str ref: str using_defaults: bool rule_filter: str | None strict: bool rules_checked: int violations_total: int errors: int warnings: int violations: list[BaseViolation] cli = None runner = CliRunner() type _FilesMap = dict[str, bytes] # --------------------------------------------------------------------------- # Shared fixtures and helpers # --------------------------------------------------------------------------- @pytest.fixture def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Fresh Muse repo in tmp_path.""" monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) result = runner.invoke(cli, ["init"]) assert result.exit_code == 0, result.output return tmp_path def _write(repo: pathlib.Path, rel: str, content: str) -> None: p = repo / rel p.parent.mkdir(parents=True, exist_ok=True) p.write_text(content) def _commit(msg: str = "snapshot") -> None: r = runner.invoke(cli, ["code", "add", "."]) assert r.exit_code == 0, r.output r = runner.invoke(cli, ["commit", "-m", msg]) assert r.exit_code == 0, r.output def _inv(args: list[str] | None = None) -> InvokeResult: return runner.invoke(cli, ["code", "invariants"] + (args or [])) def _write_object(root: pathlib.Path, content: bytes) -> str: h = hashlib.sha256(content).hexdigest() obj = root / ".muse" / "objects" / h[:2] / h[2:] obj.parent.mkdir(parents=True, exist_ok=True) obj.write_bytes(content) return h def _make_bare_repo(tmp_path: pathlib.Path) -> pathlib.Path: muse = tmp_path / ".muse" muse.mkdir() (muse / "repo.json").write_text('{"repo_id":"test"}') (muse / "HEAD").write_text("ref: refs/heads/main") (muse / "commits").mkdir() (muse / "snapshots").mkdir() (muse / "refs" / "heads").mkdir(parents=True) (muse / "objects").mkdir() return tmp_path # --------------------------------------------------------------------------- # Section I — Unit: check_forbidden_dependency # --------------------------------------------------------------------------- class TestCheckForbiddenDependency: def _manifest( self, root: pathlib.Path, files: _FilesMap ) -> Manifest: manifest: Manifest = {} for fp, src in files.items(): h = _write_object(root, src) manifest[fp] = h return manifest def test_violation_detected(self, tmp_path: pathlib.Path) -> None: root = _make_bare_repo(tmp_path) src = { "core/engine.py": b"from cli import app\n", "cli/app.py": b"def run(): pass\n", } manifest = self._manifest(root, src) violations = check_forbidden_dependency( manifest, root, "core→cli", "error", source_pattern="core/", forbidden_pattern="cli/", ) assert len(violations) == 1 assert "core/engine.py" in violations[0]["address"] assert violations[0]["severity"] == "error" def test_no_violation_when_no_match(self, tmp_path: pathlib.Path) -> None: root = _make_bare_repo(tmp_path) src = { "core/engine.py": b"def process(): pass\n", "cli/app.py": b"from core.engine import process\n", } manifest = self._manifest(root, src) violations = check_forbidden_dependency( manifest, root, "no-op", "error", source_pattern="core/", forbidden_pattern="cli/", ) # cli imports from core, not the other way around assert violations == [] def test_empty_pattern_skips_with_no_violations( self, tmp_path: pathlib.Path ) -> None: root = _make_bare_repo(tmp_path) manifest = self._manifest(root, {"a.py": b"import b\n", "b.py": b""}) # Missing forbidden_pattern → should log warning and return [] violations = check_forbidden_dependency( manifest, root, "bad-rule", "error", source_pattern="a", forbidden_pattern="", ) assert violations == [] def test_multiple_violations(self, tmp_path: pathlib.Path) -> None: root = _make_bare_repo(tmp_path) src = { "core/a.py": b"from cli import x\n", "core/b.py": b"from cli import y\n", "cli/x.py": b"def x(): pass\n", "cli/y.py": b"def y(): pass\n", } manifest = self._manifest(root, src) violations = check_forbidden_dependency( manifest, root, "core→cli", "error", source_pattern="core/", forbidden_pattern="cli/", ) assert len(violations) == 2 def test_warning_severity_respected(self, tmp_path: pathlib.Path) -> None: root = _make_bare_repo(tmp_path) src = { "core/engine.py": b"from cli import app\n", "cli/app.py": b"def run(): pass\n", } manifest = self._manifest(root, src) violations = check_forbidden_dependency( manifest, root, "soft-rule", "warning", source_pattern="core/", forbidden_pattern="cli/", ) assert all(v["severity"] == "warning" for v in violations) # --------------------------------------------------------------------------- # Section II — Unit: check_layer_boundary # --------------------------------------------------------------------------- class TestCheckLayerBoundary: def _manifest( self, root: pathlib.Path, files: _FilesMap ) -> Manifest: manifest: Manifest = {} for fp, src in files.items(): h = _write_object(root, src) manifest[fp] = h return manifest def test_lower_imports_upper_is_violation( self, tmp_path: pathlib.Path ) -> None: root = _make_bare_repo(tmp_path) src = { "core/engine.py": b"from cli import app\n", "cli/app.py": b"def run(): pass\n", } manifest = self._manifest(root, src) violations = check_layer_boundary( manifest, root, "layer", "error", lower="core/", upper="cli/", ) assert len(violations) == 1 assert violations[0]["severity"] == "error" assert "core/engine.py" in violations[0]["address"] def test_upper_importing_lower_is_allowed( self, tmp_path: pathlib.Path ) -> None: root = _make_bare_repo(tmp_path) src = { "core/engine.py": b"def process(): pass\n", "cli/app.py": b"from core import engine\n", } manifest = self._manifest(root, src) # cli (upper) importing core (lower) should NOT be a violation violations = check_layer_boundary( manifest, root, "layer", "error", lower="core/", upper="cli/", ) assert violations == [] def test_empty_params_skip_with_no_violations( self, tmp_path: pathlib.Path ) -> None: root = _make_bare_repo(tmp_path) manifest = self._manifest(root, {"a.py": b"import b\n", "b.py": b""}) violations = check_layer_boundary( manifest, root, "bad-rule", "error", lower="", upper="" ) assert violations == [] # --------------------------------------------------------------------------- # Section III — Integration: all 6 rule types via CLI # --------------------------------------------------------------------------- _SIMPLE_MODULE = """\ def compute(x: int) -> int: return x * 2 """ _COMPLEX_MODULE = """\ def very_complex(x: int) -> int: if x > 0: if x > 10: if x > 100: if x > 1000: if x > 10000: if x > 100000: if x > 1000000: if x > 10000000: if x > 100000000: if x > 1000000000: return x return 0 """ _RULES_MAX_COMPLEXITY = """\ [[rule]] name = "complexity gate" severity = "warning" scope = "function" rule_type = "max_complexity" [rule.params] threshold = 5 """ _RULES_NO_CYCLES = """\ [[rule]] name = "no cycles" severity = "error" scope = "file" rule_type = "no_circular_imports" """ _RULES_FORBIDDEN = """\ [[rule]] name = "core must not import cli" severity = "error" scope = "file" rule_type = "forbidden_dependency" [rule.params] source_pattern = "src/core/" forbidden_pattern = "src/cli/" """ _RULES_LAYER = """\ [[rule]] name = "layer boundary" severity = "error" scope = "file" rule_type = "layer_boundary" [rule.params] lower = "src/core/" upper = "src/cli/" """ _RULES_TEST_COVERAGE = """\ [[rule]] name = "test coverage floor" severity = "warning" scope = "repo" rule_type = "test_coverage_floor" [rule.params] min_ratio = 0.99 """ _RULES_DEAD_EXPORTS = """\ [[rule]] name = "no dead exports" severity = "warning" scope = "file" rule_type = "no_dead_exports" """ class TestIntegrationRuleTypes: def _rules(self, repo: pathlib.Path, content: str) -> None: _write(repo, ".muse/code_invariants.toml", content) def test_max_complexity_violation(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_MAX_COMPLEXITY) _write(repo, "src/complex.py", _COMPLEX_MODULE) _commit("add complex") result = _inv() assert "complexity" in result.output.lower() # warnings don't exit 1 without --strict assert result.exit_code == 0 def test_max_complexity_pass(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_MAX_COMPLEXITY) _write(repo, "src/simple.py", _SIMPLE_MODULE) _commit("add simple") result = _inv() assert result.exit_code == 0 assert "✅" in result.output def test_no_circular_imports_cycle_detected( self, repo: pathlib.Path ) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", "from src import b\n") _write(repo, "src/b.py", "from src import a\n") _commit("cycle") result = _inv() assert "cycle" in result.output.lower() or "circular" in result.output.lower() assert result.exit_code == 1 # error severity def test_no_circular_imports_clean(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", "def foo(): pass\n") _write(repo, "src/b.py", "from src import a\n") _commit("no cycle") result = _inv() assert result.exit_code == 0 def test_forbidden_dependency_violation( self, repo: pathlib.Path ) -> None: self._rules(repo, _RULES_FORBIDDEN) _write(repo, "src/core/engine.py", "from src.cli import app\n") _write(repo, "src/cli/app.py", "def run(): pass\n") _commit("forbidden import") result = _inv() assert "core" in result.output or "forbidden" in result.output.lower() assert result.exit_code == 1 def test_forbidden_dependency_clean(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_FORBIDDEN) _write(repo, "src/core/engine.py", "def process(): pass\n") _write(repo, "src/cli/app.py", "from src.core import engine\n") _commit("allowed import direction") result = _inv() assert result.exit_code == 0 def test_layer_boundary_violation(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_LAYER) _write(repo, "src/core/engine.py", "from src.cli import app\n") _write(repo, "src/cli/app.py", "def run(): pass\n") _commit("layer violation") result = _inv() assert result.exit_code == 1 def test_test_coverage_floor_violation(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_TEST_COVERAGE) _write(repo, "src/billing.py", "def pay(): pass\ndef refund(): pass\n") # No test file → coverage is 0% < 99% _commit("no tests") result = _inv() assert "coverage" in result.output.lower() assert result.exit_code == 0 # warning severity def test_test_coverage_floor_pass(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_TEST_COVERAGE.replace("0.99", "0.0")) _write(repo, "src/billing.py", "def pay(): pass\n") _write(repo, "tests/test_billing.py", "def test_pay(): pass\n") _commit("with tests, floor=0") result = _inv() assert result.exit_code == 0 def test_no_dead_exports_flags_unreferenced( self, repo: pathlib.Path ) -> None: self._rules(repo, _RULES_DEAD_EXPORTS) _write(repo, "src/utils.py", "def orphan(): pass\n") _commit("orphan function") result = _inv() # Dead exports are warnings — no exit 1 without --strict assert result.exit_code == 0 # --------------------------------------------------------------------------- # Section IV — Integration: flags # --------------------------------------------------------------------------- class TestFlags: def _rules(self, repo: pathlib.Path, content: str) -> None: _write(repo, ".muse/code_invariants.toml", content) def test_commit_flag_resolves_branch_name( self, repo: pathlib.Path ) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("v1") result = _inv(["--commit", "main"]) assert result.exit_code == 0 def test_commit_flag_invalid_ref_exits_nonzero( self, repo: pathlib.Path ) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("v1") result = _inv(["--commit", "no-such-ref-xyz"]) assert result.exit_code != 0 def test_rule_filter_restricts_to_matching_rules( self, repo: pathlib.Path ) -> None: combined = _RULES_MAX_COMPLEXITY + "\n" + _RULES_NO_CYCLES self._rules(repo, combined) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("simple") result = _inv(["--rule", "no_circular_imports"]) # Should only run the no_circular_imports rule, not complexity gate assert "complexity" not in result.output.lower() assert result.exit_code == 0 def test_rule_filter_by_name(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_MAX_COMPLEXITY + "\n" + _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("simple") result = _inv(["--rule", "no cycles"]) assert result.exit_code == 0 def test_rule_filter_no_match_exits_zero( self, repo: pathlib.Path ) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("simple") result = _inv(["--rule", "nonexistent_rule_type_xyz"]) assert result.exit_code == 0 def test_strict_makes_warnings_exit_one(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_MAX_COMPLEXITY) _write(repo, "src/complex.py", _COMPLEX_MODULE) _commit("complex") # Without --strict: warning, exit 0 assert _inv().exit_code == 0 # With --strict: warning → exit 1 assert _inv(["--strict"]).exit_code == 1 def test_strict_no_violations_exits_zero( self, repo: pathlib.Path ) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("clean") assert _inv(["--strict"]).exit_code == 0 def test_no_rules_file_uses_defaults(self, repo: pathlib.Path) -> None: # Ensure no code_invariants.toml exists (repo / ".muse" / "code_invariants.toml").unlink(missing_ok=True) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("no rules file") result = _inv() # Built-in defaults should still produce output assert "built-in default" in result.output # --------------------------------------------------------------------------- # Section V — Exit-code contract # --------------------------------------------------------------------------- class TestExitCode: def _rules(self, repo: pathlib.Path, content: str) -> None: _write(repo, ".muse/code_invariants.toml", content) def test_all_pass_exits_zero(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("clean") assert _inv().exit_code == 0 def test_error_violations_exit_one(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", "from src import b\n") _write(repo, "src/b.py", "from src import a\n") _commit("cycle") assert _inv().exit_code == 1 def test_warning_violations_exit_zero_without_strict( self, repo: pathlib.Path ) -> None: self._rules(repo, _RULES_MAX_COMPLEXITY) _write(repo, "src/c.py", _COMPLEX_MODULE) _commit("complex") assert _inv().exit_code == 0 def test_warning_violations_exit_one_with_strict( self, repo: pathlib.Path ) -> None: self._rules(repo, _RULES_MAX_COMPLEXITY) _write(repo, "src/c.py", _COMPLEX_MODULE) _commit("complex") assert _inv(["--strict"]).exit_code == 1 def test_json_exit_code_matches_human(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", "from src import b\n") _write(repo, "src/b.py", "from src import a\n") _commit("cycle") human_code = _inv().exit_code json_code = _inv(["--json"]).exit_code assert human_code == json_code == 1 # --------------------------------------------------------------------------- # Section VI — JSON output schema # --------------------------------------------------------------------------- class TestJsonSchema: def _j(self, args: list[str] | None = None) -> _InvariantsCliJson: result = _inv((args or []) + ["--json"]) data: _InvariantsCliJson = json.loads(result.output) return data def _rules(self, repo: pathlib.Path, content: str) -> None: _write(repo, ".muse/code_invariants.toml", content) def test_required_keys_present(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("baseline") d = self._j() required = { "schema_version", "commit", "branch", "ref", "using_defaults", "rule_filter", "strict", "rules_checked", "violations_total", "errors", "warnings", "violations", } assert required <= d.keys() def test_zero_violations_when_clean(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("baseline") d = self._j() assert d["violations_total"] == 0 assert d["errors"] == 0 assert d["warnings"] == 0 assert d["violations"] == [] def test_violation_fields_present(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", "from src import b\n") _write(repo, "src/b.py", "from src import a\n") _commit("cycle") d = self._j() assert d["violations_total"] >= 1 for v in d["violations"]: assert {"rule_name", "severity", "address", "description"} <= v.keys() def test_errors_and_warnings_counted_correctly( self, repo: pathlib.Path ) -> None: # Mix error + warning rules self._rules(repo, _RULES_NO_CYCLES + "\n" + _RULES_MAX_COMPLEXITY) _write(repo, "src/a.py", "from src import b\n") _write(repo, "src/b.py", "from src import a\n") _write(repo, "src/c.py", _COMPLEX_MODULE) _commit("mixed") d = self._j() assert d["errors"] >= 1 assert d["warnings"] >= 1 assert d["errors"] + d["warnings"] == d["violations_total"] def test_strict_reflected_in_json(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("baseline") d = self._j(["--strict"]) assert d["strict"] is True def test_rule_filter_reflected_in_json(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("baseline") d = self._j(["--rule", "no_circular_imports"]) assert d["rule_filter"] == "no_circular_imports" def test_using_defaults_false_when_rules_file_exists( self, repo: pathlib.Path ) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("baseline") d = self._j() assert d["using_defaults"] is False def test_using_defaults_true_when_no_rules_file( self, repo: pathlib.Path ) -> None: (repo / ".muse" / "code_invariants.toml").unlink(missing_ok=True) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("no rules file") d = self._j() assert d["using_defaults"] is True def test_branch_is_nonempty_string(self, repo: pathlib.Path) -> None: self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("baseline") d = self._j() assert isinstance(d["branch"], str) and d["branch"] def test_rules_checked_matches_loaded_rules( self, repo: pathlib.Path ) -> None: combined = _RULES_NO_CYCLES + "\n" + _RULES_MAX_COMPLEXITY self._rules(repo, combined) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("two rules") d = self._j() assert d["rules_checked"] == 2 # --------------------------------------------------------------------------- # Section VII — No rules file → built-in defaults # --------------------------------------------------------------------------- class TestBuiltinDefaults: def test_runs_without_error_when_no_file( self, repo: pathlib.Path ) -> None: (repo / ".muse" / "code_invariants.toml").unlink(missing_ok=True) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("no file") result = _inv() # Should succeed (may warn but not error out) assert "built-in" in result.output.lower() or "default" in result.output.lower() def test_json_using_defaults_true(self, repo: pathlib.Path) -> None: (repo / ".muse" / "code_invariants.toml").unlink(missing_ok=True) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("no file") d = json.loads(_inv(["--json"]).output) assert d["using_defaults"] is True def test_builtin_defaults_include_complexity_and_cycles( self, repo: pathlib.Path ) -> None: (repo / ".muse" / "code_invariants.toml").unlink(missing_ok=True) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("no file") d = json.loads(_inv(["--json"]).output) # Built-in defaults are: complexity_gate, no_cycles, dead_exports assert d["rules_checked"] == 3 # --------------------------------------------------------------------------- # Section VIII — Regression: bugs fixed in this review # --------------------------------------------------------------------------- class TestRegressions: def _rules(self, repo: pathlib.Path, content: str) -> None: _write(repo, ".muse/code_invariants.toml", content) def test_exit_code_1_on_error_violations(self, repo: pathlib.Path) -> None: """Old CLI always exited 0 — useless as a CI gate.""" self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", "from src import b\n") _write(repo, "src/b.py", "from src import a\n") _commit("cycle") assert _inv().exit_code == 1 def test_invalid_ref_exits_nonzero_not_silently( self, repo: pathlib.Path ) -> None: """Old CLI silently defaulted `manifest = None or {}` — ran with empty snapshot.""" self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("baseline") result = _inv(["--commit", "no-such-ref-8675309"]) assert result.exit_code != 0 def test_branch_name_resolves_for_commit_flag( self, repo: pathlib.Path ) -> None: """Old code passed branch name to resolve_commit_ref which doesn't handle it.""" self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("v1") result = _inv(["--commit", "main"]) assert result.exit_code == 0, result.output def test_json_has_branch_field(self, repo: pathlib.Path) -> None: """Old JSON output was missing branch field.""" self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("baseline") d = json.loads(_inv(["--json"]).output) assert "branch" in d assert isinstance(d["branch"], str) def test_json_has_errors_and_warnings_fields( self, repo: pathlib.Path ) -> None: """Old JSON output was missing errors/warnings split.""" self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("baseline") d = json.loads(_inv(["--json"]).output) assert "errors" in d assert "warnings" in d def test_forbidden_dependency_available(self, repo: pathlib.Path) -> None: """Old CLI only had 4 rule types; forbidden_dependency was not in plugin engine.""" self._rules(repo, _RULES_FORBIDDEN) _write(repo, "src/core/engine.py", "from src.cli import app\n") _write(repo, "src/cli/app.py", "def run(): pass\n") _commit("forbidden import") result = _inv() assert result.exit_code == 1 def test_layer_boundary_available(self, repo: pathlib.Path) -> None: """Old CLI only had 4 rule types; layer_boundary was not in plugin engine.""" self._rules(repo, _RULES_LAYER) _write(repo, "src/core/engine.py", "from src.cli import app\n") _write(repo, "src/cli/app.py", "def run(): pass\n") _commit("layer violation") result = _inv() assert result.exit_code == 1 def test_no_silent_or_fallback_on_corrupt_commit( self, repo: pathlib.Path ) -> None: """Old code: `manifest = ... or {}` silently ran with empty snapshot.""" self._rules(repo, _RULES_NO_CYCLES) _write(repo, "src/a.py", _SIMPLE_MODULE) _commit("baseline") # Corrupt the snapshot by pointing to a nonexistent ref result = _inv(["--commit", "deadbeef00"]) # Should fail, not silently succeed with zero violations assert result.exit_code != 0 # --------------------------------------------------------------------------- # Section IX — Stress tests # --------------------------------------------------------------------------- _CLEAN_MOD = "def fn_{i}(x: int) -> int:\n return x + {i}\n" _CYCLIC_MOD = "from src import mod_{j}\n" class TestStress: @pytest.mark.slow def test_100_clean_files_no_false_positives( self, repo: pathlib.Path ) -> None: """100 clean files with a no_circular_imports rule — zero violations.""" _write(repo, ".muse/code_invariants.toml", _RULES_NO_CYCLES) for i in range(100): _write(repo, f"src/mod_{i:03d}.py", _CLEAN_MOD.format(i=i)) _commit("100 clean files") d = json.loads(_inv(["--json"]).output) assert d["violations_total"] == 0 @pytest.mark.slow def test_large_repo_with_multiple_rule_types( self, repo: pathlib.Path ) -> None: """50 files: mix of complexity, cycle, and forbidden rules.""" combined = ( _RULES_MAX_COMPLEXITY.replace("5", "20") # threshold=20, most pass + "\n" + _RULES_NO_CYCLES ) _write(repo, ".muse/code_invariants.toml", combined) for i in range(50): _write(repo, f"src/mod_{i:03d}.py", _CLEAN_MOD.format(i=i)) # Introduce one cycle _write(repo, "src/mod_000.py", "from src import mod_001\n") _write(repo, "src/mod_001.py", "from src import mod_000\n") _commit("50 files + cycle") d = json.loads(_inv(["--json"]).output) assert d["errors"] >= 1 # the cycle @pytest.mark.slow def test_rule_filter_on_large_repo(self, repo: pathlib.Path) -> None: """Rule filter works correctly on a large repo — only selected rule runs.""" combined = _RULES_MAX_COMPLEXITY + "\n" + _RULES_NO_CYCLES _write(repo, ".muse/code_invariants.toml", combined) for i in range(50): _write(repo, f"src/mod_{i:03d}.py", _CLEAN_MOD.format(i=i)) _commit("50 files") # With filter: only no_circular_imports — result should have 1 rule checked d = json.loads(_inv(["--json", "--rule", "no_circular_imports"]).output) assert d["rules_checked"] == 1 @pytest.mark.slow def test_json_violation_schema_consistent_across_all_rule_types( self, repo: pathlib.Path ) -> None: """Every violation across all 6 rule types has the required JSON fields.""" all_rules = ( _RULES_MAX_COMPLEXITY.replace("5", "3") # low threshold → many violations + "\n" + _RULES_DEAD_EXPORTS + "\n" + _RULES_TEST_COVERAGE + "\n" + _RULES_FORBIDDEN + "\n" + _RULES_LAYER ) _write(repo, ".muse/code_invariants.toml", all_rules) _write(repo, "src/core/engine.py", "from src.cli import app\n" + _COMPLEX_MODULE) _write(repo, "src/cli/app.py", "def run(): pass\n") _commit("all violations") d = json.loads(_inv(["--json"]).output) required_keys = {"rule_name", "severity", "address", "description"} for v in d["violations"]: missing = required_keys - v.keys() assert not missing, f"Violation missing keys {missing}: {v}" @pytest.mark.slow def test_strict_on_100_file_repo_with_one_warning( self, repo: pathlib.Path ) -> None: """--strict exits 1 with even a single warning in a 100-file clean repo.""" _write( repo, ".muse/code_invariants.toml", _RULES_MAX_COMPLEXITY.replace("5", "3"), # low threshold ) # Only one file violates for i in range(99): _write(repo, f"src/mod_{i:03d}.py", _CLEAN_MOD.format(i=i)) _write(repo, "src/mod_099.py", _COMPLEX_MODULE) _commit("99 clean + 1 complex") assert _inv().exit_code == 0 # warning → 0 assert _inv(["--strict"]).exit_code == 1 # warning + strict → 1