test_code_invariants.py
python
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b
fix: try fetch/presign before fetch/mpack to avoid Cloudfla…
Sonnet 4.6
patch
7 days ago
| 1 | """Tests for the code-domain invariants engine.""" |
| 2 | |
| 3 | import pathlib |
| 4 | import tempfile |
| 5 | |
| 6 | import pytest |
| 7 | |
| 8 | from muse.core.invariants import InvariantChecker |
| 9 | from muse.plugins.code._invariants import ( |
| 10 | CodeChecker, |
| 11 | CodeInvariantRule, |
| 12 | check_max_complexity, |
| 13 | check_no_circular_imports, |
| 14 | check_no_dead_exports, |
| 15 | check_test_coverage_floor, |
| 16 | load_invariant_rules, |
| 17 | run_invariants, |
| 18 | ) |
| 19 | from muse.core.object_store import object_path |
| 20 | from muse.core.paths import code_invariants_path, muse_dir |
| 21 | |
| 22 | |
| 23 | # --------------------------------------------------------------------------- |
| 24 | # Helpers |
| 25 | # --------------------------------------------------------------------------- |
| 26 | |
| 27 | |
| 28 | def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 29 | """Set up a minimal .muse/ structure.""" |
| 30 | dot_muse = muse_dir(tmp_path) |
| 31 | dot_muse.mkdir() |
| 32 | (dot_muse / "repo.json").write_text('{"repo_id":"test"}') |
| 33 | (dot_muse / "HEAD").write_text("ref: refs/heads/main") |
| 34 | (dot_muse / "commits").mkdir() |
| 35 | (dot_muse / "snapshots").mkdir() |
| 36 | (dot_muse / "refs" / "heads").mkdir(parents=True) |
| 37 | (dot_muse / "objects").mkdir() |
| 38 | return tmp_path |
| 39 | |
| 40 | |
| 41 | def _write_object(root: pathlib.Path, content: bytes) -> str: |
| 42 | from muse.core.types import blob_id |
| 43 | from muse.core.object_store import write_object |
| 44 | oid = blob_id(content) |
| 45 | write_object(root, oid, content) |
| 46 | return oid |
| 47 | |
| 48 | |
| 49 | # --------------------------------------------------------------------------- |
| 50 | # _estimate_complexity (via check_max_complexity) |
| 51 | # --------------------------------------------------------------------------- |
| 52 | |
| 53 | |
| 54 | class TestMaxComplexity: |
| 55 | def test_simple_function_no_violation(self) -> None: |
| 56 | with tempfile.TemporaryDirectory() as tmp: |
| 57 | root = _make_repo(pathlib.Path(tmp)) |
| 58 | src = b"def simple():\n return 1\n" |
| 59 | h = _write_object(root, src) |
| 60 | manifest = {"mod.py": h} |
| 61 | violations = check_max_complexity(manifest, root, "test", "error", threshold=10) |
| 62 | assert violations == [] |
| 63 | |
| 64 | def test_complex_function_triggers_violation(self) -> None: |
| 65 | # 15+ branches = definitely over threshold 5. |
| 66 | src = b""" |
| 67 | def complex(): |
| 68 | if True: |
| 69 | pass |
| 70 | if True: |
| 71 | pass |
| 72 | if True: |
| 73 | pass |
| 74 | if True: |
| 75 | pass |
| 76 | if True: |
| 77 | pass |
| 78 | if True: |
| 79 | pass |
| 80 | if True: |
| 81 | pass |
| 82 | return 1 |
| 83 | """ |
| 84 | with tempfile.TemporaryDirectory() as tmp: |
| 85 | root = _make_repo(pathlib.Path(tmp)) |
| 86 | h = _write_object(root, src) |
| 87 | manifest = {"mod.py": h} |
| 88 | violations = check_max_complexity(manifest, root, "gate", "error", threshold=5) |
| 89 | assert len(violations) >= 1 |
| 90 | assert violations[0]["rule_name"] == "gate" |
| 91 | assert "complexity" in violations[0]["description"].lower() |
| 92 | |
| 93 | def test_non_python_file_skipped(self) -> None: |
| 94 | with tempfile.TemporaryDirectory() as tmp: |
| 95 | root = _make_repo(pathlib.Path(tmp)) |
| 96 | src = b"def hello() { return 1; }" |
| 97 | h = _write_object(root, src) |
| 98 | manifest = {"mod.js": h} |
| 99 | violations = check_max_complexity(manifest, root, "c", "error", threshold=1) |
| 100 | assert violations == [] |
| 101 | |
| 102 | |
| 103 | # --------------------------------------------------------------------------- |
| 104 | # check_no_circular_imports |
| 105 | # --------------------------------------------------------------------------- |
| 106 | |
| 107 | |
| 108 | class TestNoCircularImports: |
| 109 | def test_no_cycle_returns_empty(self) -> None: |
| 110 | with tempfile.TemporaryDirectory() as tmp: |
| 111 | root = _make_repo(pathlib.Path(tmp)) |
| 112 | a = b"import b\n" |
| 113 | b_src = b"x = 1\n" |
| 114 | ha = _write_object(root, a) |
| 115 | hb = _write_object(root, b_src) |
| 116 | manifest = {"a.py": ha, "b.py": hb} |
| 117 | violations = check_no_circular_imports(manifest, root, "no_cycles", "error") |
| 118 | assert violations == [] |
| 119 | |
| 120 | def test_cycle_detected(self) -> None: |
| 121 | with tempfile.TemporaryDirectory() as tmp: |
| 122 | root = _make_repo(pathlib.Path(tmp)) |
| 123 | # a imports b, b imports a → cycle |
| 124 | a = b"import b\n" |
| 125 | b_src = b"import a\n" |
| 126 | ha = _write_object(root, a) |
| 127 | hb = _write_object(root, b_src) |
| 128 | manifest = {"a.py": ha, "b.py": hb} |
| 129 | violations = check_no_circular_imports(manifest, root, "no_cycles", "error") |
| 130 | assert len(violations) >= 1 |
| 131 | assert "cycle" in violations[0]["description"].lower() |
| 132 | |
| 133 | def test_three_file_cycle_detected(self) -> None: |
| 134 | with tempfile.TemporaryDirectory() as tmp: |
| 135 | root = _make_repo(pathlib.Path(tmp)) |
| 136 | a = b"import b\n" |
| 137 | b_src = b"import c\n" |
| 138 | c_src = b"import a\n" |
| 139 | ha = _write_object(root, a) |
| 140 | hb = _write_object(root, b_src) |
| 141 | hc = _write_object(root, c_src) |
| 142 | manifest = {"a.py": ha, "b.py": hb, "c.py": hc} |
| 143 | violations = check_no_circular_imports(manifest, root, "cycles", "error") |
| 144 | assert len(violations) >= 1 |
| 145 | |
| 146 | |
| 147 | # --------------------------------------------------------------------------- |
| 148 | # check_no_dead_exports |
| 149 | # --------------------------------------------------------------------------- |
| 150 | |
| 151 | |
| 152 | class TestNoDeadExports: |
| 153 | def test_used_function_not_reported(self) -> None: |
| 154 | with tempfile.TemporaryDirectory() as tmp: |
| 155 | root = _make_repo(pathlib.Path(tmp)) |
| 156 | lib = b"def my_func():\n return 1\n" |
| 157 | main = b"from lib import my_func\n" |
| 158 | hl = _write_object(root, lib) |
| 159 | hm = _write_object(root, main) |
| 160 | manifest = {"lib.py": hl, "main.py": hm} |
| 161 | violations = check_no_dead_exports(manifest, root, "dead", "warning") |
| 162 | # lib.my_func is imported by main.py → should not be reported. |
| 163 | addresses = [v["address"] for v in violations] |
| 164 | assert "lib.py::my_func" not in addresses |
| 165 | |
| 166 | def test_unused_function_reported(self) -> None: |
| 167 | with tempfile.TemporaryDirectory() as tmp: |
| 168 | root = _make_repo(pathlib.Path(tmp)) |
| 169 | lib = b"def orphan_fn():\n return 1\n" |
| 170 | other = b"x = 1\n" |
| 171 | hl = _write_object(root, lib) |
| 172 | ho = _write_object(root, other) |
| 173 | manifest = {"lib.py": hl, "other.py": ho} |
| 174 | violations = check_no_dead_exports(manifest, root, "dead", "warning") |
| 175 | addresses = [v["address"] for v in violations] |
| 176 | assert "lib.py::orphan_fn" in addresses |
| 177 | |
| 178 | def test_private_function_exempt(self) -> None: |
| 179 | with tempfile.TemporaryDirectory() as tmp: |
| 180 | root = _make_repo(pathlib.Path(tmp)) |
| 181 | lib = b"def _private():\n return 1\n" |
| 182 | h = _write_object(root, lib) |
| 183 | manifest = {"lib.py": h} |
| 184 | violations = check_no_dead_exports(manifest, root, "dead", "warning") |
| 185 | # Private functions are exempt. |
| 186 | assert all("_private" not in v["address"] for v in violations) |
| 187 | |
| 188 | def test_test_file_exempt(self) -> None: |
| 189 | with tempfile.TemporaryDirectory() as tmp: |
| 190 | root = _make_repo(pathlib.Path(tmp)) |
| 191 | lib = b"def test_something():\n assert True\n" |
| 192 | h = _write_object(root, lib) |
| 193 | manifest = {"test_stuff.py": h} |
| 194 | violations = check_no_dead_exports(manifest, root, "dead", "warning") |
| 195 | assert violations == [] |
| 196 | |
| 197 | |
| 198 | # --------------------------------------------------------------------------- |
| 199 | # check_test_coverage_floor |
| 200 | # --------------------------------------------------------------------------- |
| 201 | |
| 202 | |
| 203 | class TestTestCoverageFloor: |
| 204 | def test_well_covered_code_no_violation(self) -> None: |
| 205 | with tempfile.TemporaryDirectory() as tmp: |
| 206 | root = _make_repo(pathlib.Path(tmp)) |
| 207 | src = b"def foo():\n return 1\n" |
| 208 | test_src = b"def test_foo():\n assert True\n" |
| 209 | hs = _write_object(root, src) |
| 210 | ht = _write_object(root, test_src) |
| 211 | manifest = {"src.py": hs, "test_src.py": ht} |
| 212 | violations = check_test_coverage_floor(manifest, root, "coverage", "warning", min_ratio=0.5) |
| 213 | assert violations == [] |
| 214 | |
| 215 | def test_uncovered_code_violates(self) -> None: |
| 216 | with tempfile.TemporaryDirectory() as tmp: |
| 217 | root = _make_repo(pathlib.Path(tmp)) |
| 218 | src = b"def foo():\n pass\ndef bar():\n pass\ndef baz():\n pass\n" |
| 219 | h = _write_object(root, src) |
| 220 | manifest = {"src.py": h} |
| 221 | violations = check_test_coverage_floor(manifest, root, "coverage", "warning", min_ratio=0.5) |
| 222 | assert len(violations) == 1 |
| 223 | assert "coverage floor" in violations[0]["description"].lower() |
| 224 | |
| 225 | def test_no_functions_no_violation(self) -> None: |
| 226 | with tempfile.TemporaryDirectory() as tmp: |
| 227 | root = _make_repo(pathlib.Path(tmp)) |
| 228 | src = b"X = 1\n" |
| 229 | h = _write_object(root, src) |
| 230 | manifest = {"config.py": h} |
| 231 | violations = check_test_coverage_floor(manifest, root, "coverage", "warning", min_ratio=0.5) |
| 232 | assert violations == [] |
| 233 | |
| 234 | |
| 235 | # --------------------------------------------------------------------------- |
| 236 | # load_invariant_rules |
| 237 | # --------------------------------------------------------------------------- |
| 238 | |
| 239 | |
| 240 | class TestLoadInvariantRules: |
| 241 | def test_no_file_returns_defaults(self) -> None: |
| 242 | """load_invariant_rules(repo_root, None) with no file on disk returns built-in defaults.""" |
| 243 | with tempfile.TemporaryDirectory() as tmp: |
| 244 | root = _make_repo(pathlib.Path(tmp)) |
| 245 | rules = load_invariant_rules(root, None) |
| 246 | assert len(rules) >= 1 |
| 247 | rule_types = {r["rule_type"] for r in rules} |
| 248 | assert "max_complexity" in rule_types |
| 249 | |
| 250 | def test_default_path_from_repo_root(self) -> None: |
| 251 | """When rules_file is None, the default is .muse/code_invariants.toml inside repo_root.""" |
| 252 | toml = "[[rule]]\nname='repo_rule'\nseverity='error'\nscope='function'\nrule_type='max_complexity'\n" |
| 253 | with tempfile.TemporaryDirectory() as tmp: |
| 254 | root = _make_repo(pathlib.Path(tmp)) |
| 255 | code_invariants_path(root).write_text(toml) |
| 256 | rules = load_invariant_rules(root, None) |
| 257 | assert any(r["name"] == "repo_rule" for r in rules) |
| 258 | |
| 259 | def test_explicit_missing_path_returns_empty(self) -> None: |
| 260 | """An explicit path that does not exist yields no rules (caller opts out of defaults).""" |
| 261 | with tempfile.TemporaryDirectory() as tmp: |
| 262 | root = _make_repo(pathlib.Path(tmp)) |
| 263 | rules = load_invariant_rules(root, pathlib.Path("/no/such/file.toml")) |
| 264 | assert rules == [] |
| 265 | |
| 266 | def test_toml_file_loaded(self) -> None: |
| 267 | toml = "[[rule]]\nname='r1'\nseverity='error'\nscope='function'\nrule_type='max_complexity'\n" |
| 268 | with tempfile.TemporaryDirectory() as tmp: |
| 269 | root = _make_repo(pathlib.Path(tmp)) |
| 270 | with tempfile.NamedTemporaryFile(suffix=".toml", mode="w", delete=False) as f: |
| 271 | f.write(toml) |
| 272 | path = pathlib.Path(f.name) |
| 273 | try: |
| 274 | rules = load_invariant_rules(root, path) |
| 275 | assert any(r["rule_type"] == "max_complexity" for r in rules) |
| 276 | finally: |
| 277 | path.unlink(missing_ok=True) |
| 278 | |
| 279 | |
| 280 | # --------------------------------------------------------------------------- |
| 281 | # CodeChecker (protocol) |
| 282 | # --------------------------------------------------------------------------- |
| 283 | |
| 284 | |
| 285 | class TestCodeChecker: |
| 286 | def test_satisfies_invariant_checker_protocol(self) -> None: |
| 287 | checker = CodeChecker() |
| 288 | assert isinstance(checker, InvariantChecker) |
| 289 | |
| 290 | def test_check_returns_base_report(self) -> None: |
| 291 | with tempfile.TemporaryDirectory() as tmp: |
| 292 | root = _make_repo(pathlib.Path(tmp)) |
| 293 | # No commits — check should return a report with 0 violations. |
| 294 | from muse.core.commits import ( |
| 295 | CommitRecord, |
| 296 | write_commit, |
| 297 | ) |
| 298 | from muse.core.snapshots import ( |
| 299 | SnapshotRecord, |
| 300 | write_snapshot, |
| 301 | ) |
| 302 | from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id |
| 303 | import datetime |
| 304 | snap_id = compute_snapshot_id({}) |
| 305 | snap = SnapshotRecord(snapshot_id=snap_id, manifest={}) |
| 306 | write_snapshot(root, snap) |
| 307 | ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) |
| 308 | cid = compute_commit_id( |
| 309 | parent_ids=[], |
| 310 | snapshot_id=snap_id, |
| 311 | message="init", |
| 312 | committed_at_iso=ts.isoformat(), |
| 313 | ) |
| 314 | commit = CommitRecord( |
| 315 | commit_id=cid, |
| 316 | branch="main", |
| 317 | snapshot_id=snap_id, |
| 318 | message="init", |
| 319 | committed_at=ts, |
| 320 | ) |
| 321 | write_commit(root, commit) |
| 322 | report = CodeChecker().check(root, cid) |
| 323 | assert report["commit_id"] == cid |
| 324 | assert report["domain"] == "code" |
| 325 | assert isinstance(report["violations"], list) |
File History
1 commit
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b
fix: try fetch/presign before fetch/mpack to avoid Cloudfla…
Sonnet 4.6
patch
7 days ago