From 96aaa17e6056a41cb1c40957e5e70609c5d416c6 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Wed, 4 Nov 2026 21:48:58 +0100 Subject: [PATCH] feat(principles): principles.py turns the mechanical lenses into tool hits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit principles.py runs ruff (C901, PLR0913, BLE001, S110, B006, B008, ARG), lizard (CCN, NLOC, parameters per function span) and jscpd (clones) on the lines a change touches and maps each hit to SRP, fail fast, KISS/YAGNI or DRY; a missing tool prints skip, every threshold is a required flag. It reuses the diff parsing of defaults.py, registered in sys.modules before it runs because dataclasses look the module up there, and hands lizard only the suffixes its 1.24.1 readers claim, since lizard parses any other file as C. kvb-code-clean step 3 calls it with McCabe's 10, pylint's 5 parameters, jscpd's 50 tokens and 50 NLOC; make install adds lizard 1.24.1 (pipx) and jscpd 5.4.2 (npm). The lizard and jscpd parsers are tested on fixture output only: neither tool has run for real yet. 🎯 Fix: 90 % 💥 Break: 4 % 💸 Cost: 100 % ⏱️ Time: new 🛡️ Security: 100 % 💰 Value: ? 🏆 Score: 86 🚦 Gate: ⚠️ time new: ask the operator before the commit Co-Authored-By: Claude Opus 5.5 --- scripts/install.sh | 20 ++ skills/kvb-check-principles/SKILL.md | 10 +- .../scripts/principles.py | 335 ++++++++++++++++++ skills/kvb-code-clean/SKILL.md | 25 +- tests/test_principles.py | 264 ++++++++++++++ 5 files changed, 637 insertions(+), 17 deletions(-) create mode 100755 skills/kvb-check-principles/scripts/principles.py create mode 100644 tests/test_principles.py diff --git a/scripts/install.sh b/scripts/install.sh index 40fea09..4d13ab8 100644 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -82,6 +82,26 @@ if [[ "${TARGET}" == "${HOME}" ]]; then else warn "skills: the MEDUSA install failed; kvb-check-security reviews by hand." fi + lizard_pin="lizard==1.24.1" + if command -v lizard &>/dev/null; then + log "skills: lizard is installed" + elif ! command -v pipx &>/dev/null; then + warn "skills: pipx not found; install lizard with 'pipx install ${lizard_pin}'. Until then principles.py skips its complexity hits." + elif pipx install "${lizard_pin}"; then + log "skills: installed ${lizard_pin} for the principles.py complexity hits" + else + warn "skills: the lizard install failed; principles.py skips its complexity hits." + fi + jscpd_pin="jscpd@5.4.2" + if command -v jscpd &>/dev/null; then + log "skills: jscpd is installed" + elif ! command -v npm &>/dev/null; then + warn "skills: npm not found; install jscpd with 'npm install --global ${jscpd_pin}'. Until then principles.py skips its clone hits." + elif npm install --global "${jscpd_pin}"; then + log "skills: installed ${jscpd_pin} for the principles.py clone hits" + else + warn "skills: the jscpd install failed; principles.py skips its clone hits." + fi fi log "skills: done. Restart your agent to load the skills." diff --git a/skills/kvb-check-principles/SKILL.md b/skills/kvb-check-principles/SKILL.md index f01c543..9a97f88 100644 --- a/skills/kvb-check-principles/SKILL.md +++ b/skills/kvb-check-principles/SKILL.md @@ -46,9 +46,11 @@ Apply each independently; a finding belongs to exactly one: ## Method -- **Graph first**: explore through the codebase-memory-mcp graph - (`search_graph`, `trace_path`, `get_architecture`, `detect_changes`; the - `kvb-agent-optimize` skill sets it up), grep and read after. +- **Tools and graph first**: on a diff, `scripts/principles.py` (thresholds + as in `kvb-code-clean` step 3) turns SRP, fail fast, KISS/YAGNI and textual + DRY into `file:line` leads; then the codebase-memory-mcp graph + (`search_graph`, `trace_path`, `get_architecture`, `detect_changes`), grep + and read after. - **Evidence before verdict**: every finding carries a `file:line` you read. Grep the sibling occurrences before calling something a duplicate; one hit is not a pattern. @@ -58,8 +60,6 @@ Apply each independently; a finding belongs to exactly one: - **Attack your own finding**: run anything consequential or expensive to fix through the `kvb-think-dialectic` skill before reporting it; the strongest counter is usually "not the same knowledge" or "already checked upstream". - Skeptics scale with the stakes; each brief names the `file:line` and the - claim and caps the answer at five lines. - **Respect intent**: a deliberate marker (`ponytail:`, a documented exception, a nocheck directive with a reason) is a known ceiling, a finding only once exceeded. diff --git a/skills/kvb-check-principles/scripts/principles.py b/skills/kvb-check-principles/scripts/principles.py new file mode 100755 index 0000000..b42f547 --- /dev/null +++ b/skills/kvb-check-principles/scripts/principles.py @@ -0,0 +1,335 @@ +#!/usr/bin/env python3 +"""Report the principle-lens hits of ruff, lizard and jscpd on the lines a change touches. + +Args: + --repo: a directory inside the git work tree to scan. + --staged: scan only the staged changes (git diff --cached); the tools + read a snapshot of the index. Without it: every change against HEAD, + staged and unstaged, in the working tree. + --file: count every line of this file as changed; repeatable, relative + to --repo or absolute. Untracked files are in no diff. + --ccn: a function above this cyclomatic complexity is a hit (ruff C901, + lizard). + --nloc: a function above this many lines of code, blank and comment + lines excluded, is a hit (lizard). + --params: a function above this many parameters is a hit (ruff PLR0913, + lizard). + --min-tokens: a clone of at least this many tokens is a hit (jscpd). + --json: print the hits as a JSON list instead of tab-separated lines. + +Output: + One line per hit: file:line, lens, tool, message. ruff C901 and PLR0913 + and lizard map to SRP; ruff BLE001, S110, B006 and B008 to fail fast; + ruff ARG to KISS/YAGNI; jscpd to DRY. A ruff hit counts when its range + touches a changed line, a lizard function when its span does, a jscpd + clone when either side does. ruff reads Python files, lizard the + languages it has a reader for, jscpd the whole tree for clone partners. + A tool missing from PATH writes `skip: missing` to stderr instead + of hits. + +Exit codes: + 0 no hits, 1 hits, 2 usage error, git failure, a failed tool or + unreadable tool output, or a missing kvb-code-clean/scripts/defaults.py, + which parses the diff. +""" + +from __future__ import annotations + +import argparse +import importlib.util +import json +import os +import re +import shutil +import subprocess +import sys +import tempfile +from dataclasses import asdict, dataclass +from pathlib import Path +from types import ModuleType + +DEFAULTS = Path(__file__).resolve().parents[2] / "kvb-code-clean" / "scripts" / "defaults.py" +RUFF_LENSES = { + "C901": "SRP", + "PLR0913": "SRP", + "BLE001": "fail fast", + "S110": "fail fast", + "B006": "fail fast", + "B008": "fail fast", + "ARG": "KISS/YAGNI", +} +PYTHON = frozenset({".py", ".pyi"}) +LIZARD = frozenset( + { + ".c", ".cc", ".cpp", ".cxx", ".h", ".hpp", ".cs", ".java", ".js", ".cjs", ".mjs", ".jsx", ".ts", + ".tsx", ".vue", ".py", ".rb", ".php", ".go", ".rs", ".swift", ".kt", ".kts", ".scala", ".lua", + ".m", ".mm", ".pl", ".pm", ".r", ".erl", ".hrl", ".es", ".escript", ".sol", ".sql", ".pks", + ".pkb", ".pls", ".plb", ".pck", ".gd", ".zig", ".st", ".sdl", ".ttcn", ".ttcnpp", ".f", ".f70", + ".f90", ".f95", ".f03", ".f08", ".for", ".ftn", ".fpp", + } +) +LIZARD_WARNING = re.compile( + r"^(?P.+?):(?P\d+): warning: " + r"(?P.+? has \d+ NLOC, \d+ CCN, \d+ token, \d+ PARAM, (?P\d+) length\b.*)$" +) +REPORT = "jscpd-report.json" +BLOBS = frozenset({b"100644", b"100755"}) + + +def load(path: Path) -> ModuleType: + if not path.is_file(): + print(f"principles: {path} is missing; it parses the diff", file=sys.stderr) + sys.exit(2) + spec = importlib.util.spec_from_file_location("kvb_code_clean_defaults", path) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +defaults = load(DEFAULTS) +ScanError = defaults.ScanError + + +@dataclass(frozen=True, order=True) +class Hit: + file: str + line: int + lens: str + tool: str + message: str + + +@dataclass(frozen=True) +class Target: + path: Path + lines: frozenset[int] | None + + def overlaps(self, start: int, end: int) -> bool: + return self.lines is None or not self.lines.isdisjoint(range(start, end + 1)) + + +@dataclass(frozen=True) +class Scan: + base: Path + scratch: Path + targets: dict[str, Target] + shown: dict[str, str] + limits: argparse.Namespace + + def name(self, raw: str) -> str: + path = (self.base / raw).resolve() + if str(path) in self.shown: + return self.shown[str(path)] + return path.relative_to(self.base).as_posix() if path.is_relative_to(self.base) else str(path) + + def touched(self, raw: str, start: int, end: int) -> str | None: + name = self.name(raw) + target = self.targets.get(name) + return name if target is not None and target.overlaps(start, end) else None + + def files(self, suffixes: frozenset[str]) -> list[str]: + return [str(target.path) for target in self.targets.values() if target.path.suffix.lower() in suffixes] + + +def available(tool: str) -> bool: + if shutil.which(tool): + return True + print(f"skip: {tool} missing", file=sys.stderr) + return False + + +def run(command: list[str], cwd: Path) -> subprocess.CompletedProcess: + return subprocess.run( + command, cwd=cwd, capture_output=True, text=True, encoding="utf-8", errors="replace", check=False + ) + + +def failed(result: subprocess.CompletedProcess) -> ScanError: + return ScanError(f"{result.args[0]} exited {result.returncode}: {result.stderr.strip()}") + + +def ruff_lens(code: str | None) -> str | None: + if code is None: + return None + return next((lens for prefix, lens in RUFF_LENSES.items() if code.startswith(prefix)), None) + + +def ruff(scan: Scan) -> list[Hit]: + paths = scan.files(PYTHON) + if not paths or not available("ruff"): + return [] + result = run( + [ + "ruff", "check", "--isolated", "--no-cache", "--exit-zero", "--output-format", "json", + "--select", ",".join(RUFF_LENSES), + "--config", f"lint.mccabe.max-complexity = {scan.limits.ccn}", + "--config", f"lint.pylint.max-args = {scan.limits.params}", + *paths, + ], + scan.base, + ) + if result.returncode != 0: + raise failed(result) + hits = [] + for row in json.loads(result.stdout): + lens = ruff_lens(row["code"]) + start = row["location"]["row"] + name = scan.touched(row["filename"], start, row["end_location"]["row"]) + if lens and name: + hits.append(Hit(name, start, lens, "ruff", f"{row['code']} {row['message']}")) + return hits + + +def parse_lizard(text: str) -> list[tuple[str, int, int, str]]: + found = [] + for row in text.splitlines(): + match = LIZARD_WARNING.match(row) + if match: + line = int(match["line"]) + found.append((match["path"], line, line + int(match["length"]) - 1, match["message"])) + return found + + +def lizard(scan: Scan) -> list[Hit]: + paths = scan.files(LIZARD) + if not paths or not available("lizard"): + return [] + limits = scan.limits + result = run( + [ + "lizard", "--warnings_only", "--CCN", str(limits.ccn), "--arguments", str(limits.params), + "--Threshold", f"nloc={limits.nloc}", *paths, + ], + scan.base, + ) + if result.returncode not in (0, 1): + raise failed(result) + return [ + Hit(name, start, "SRP", "lizard", message) + for path, start, end, message in parse_lizard(result.stdout) + if (name := scan.touched(path, start, end)) + ] + + +def side(part: dict) -> tuple[str, int, int]: + return part["name"], part["start"], part["end"] + + +def parse_jscpd(report: dict) -> list[tuple[tuple[str, int, int], tuple[str, int, int], int]]: + return [(side(clone["firstFile"]), side(clone["secondFile"]), clone["tokens"]) for clone in report["duplicates"]] + + +def jscpd(scan: Scan) -> list[Hit]: + if not scan.targets or not available("jscpd"): + return [] + extra = [str(target.path) for target in scan.targets.values() if not target.path.is_relative_to(scan.base)] + output = scan.scratch / "jscpd" + output.mkdir() + result = run( + [ + "jscpd", "--min-tokens", str(scan.limits.min_tokens), "--reporters", "json", "--output", str(output), + "--absolute", "--silent", str(scan.base), *extra, + ], + scan.base, + ) + report = output / REPORT + if not report.is_file(): + raise failed(result) + hits = [] + for first, second, tokens in parse_jscpd(json.loads(report.read_text(encoding="utf-8"))): + for (path, start, end), (other, other_start, other_end) in ((first, second), (second, first)): + name = scan.touched(path, start, end) + if name: + clone = f"clone of {scan.name(other)}:{other_start}-{other_end}, {tokens} tokens" + hits.append(Hit(name, start, "DRY", "jscpd", clone)) + break + return hits + + +def git_bytes(root: Path, *args: str, stdin: bytes) -> bytes: + result = subprocess.run(["git", *args], cwd=root, input=stdin, capture_output=True, check=False) + if result.returncode != 0: + raise ScanError(f"git {args[0]}: {result.stderr.decode('utf-8', 'replace').strip()}") + return result.stdout + + +def snapshot(root: Path, target: Path) -> None: + entries = [] + for record in git_bytes(root, "ls-files", "--stage", "-z", stdin=b"").split(b"\0"): + meta, _, path = record.partition(b"\t") + fields = meta.split() + if len(fields) == 3 and fields[0] in BLOBS and fields[2] == b"0": + entries.append((fields[1], os.fsdecode(path))) + data = git_bytes(root, "cat-file", "--batch", stdin=b"".join(oid + b"\n" for oid, _ in entries)) + offset = 0 + for _, path in entries: + newline = data.index(b"\n", offset) + size = int(data[offset:newline].split()[2]) + file = target / path + file.parent.mkdir(parents=True, exist_ok=True) + file.write_bytes(data[newline + 1 : newline + 1 + size]) + offset = newline + 2 + size + + +def collect(repo: Path, staged: bool, names: list[str], limits: argparse.Namespace) -> list[Hit]: + if not repo.is_dir(): + raise ScanError(f"{repo} is not a directory") + root = Path(defaults.git(repo, "rev-parse", "--show-toplevel").strip()).resolve() + changed = defaults.added_lines(root, staged) + with tempfile.TemporaryDirectory() as tmp: + scratch = Path(tmp).resolve() + base = root + if staged: + base = scratch / "index" + base.mkdir() + snapshot(root, base) + targets = { + path: Target(base / path, frozenset(lines)) for path, lines in changed.items() if (base / path).is_file() + } + for name in names: + source = (repo / name).resolve() + if not source.is_file(): + raise ScanError(f"{name}: no such file") + if not source.is_relative_to(root): + targets[str(source)] = Target(source, None) + continue + shown = source.relative_to(root).as_posix() + if staged: + (base / shown).parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(source, base / shown) + targets[shown] = Target(base / shown, None) + shown = {str(target.path.resolve()): name for name, target in targets.items()} + scan = Scan(base, scratch, targets, shown, limits) + return sorted(hit for probe in (ruff, lizard, jscpd) for hit in probe(scan)) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--repo", required=True) + parser.add_argument("--staged", action="store_true") + parser.add_argument("--file", action="append", default=[]) + parser.add_argument("--ccn", type=int, required=True) + parser.add_argument("--nloc", type=int, required=True) + parser.add_argument("--params", type=int, required=True) + parser.add_argument("--min-tokens", type=int, required=True) + parser.add_argument("--json", action="store_true") + args = parser.parse_args() + try: + hits = collect(Path(args.repo), args.staged, args.file, args) + except ScanError as error: + print(f"principles: {error}", file=sys.stderr) + return 2 + except (KeyError, ValueError) as error: + print(f"principles: unreadable tool output: {error!r}", file=sys.stderr) + return 2 + if args.json: + print(json.dumps([asdict(hit) for hit in hits], indent=1)) + return 1 if hits else 0 + for hit in hits: + print("\t".join([f"{hit.file}:{hit.line}", hit.lens, hit.tool, hit.message])) + return 1 if hits else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/skills/kvb-code-clean/SKILL.md b/skills/kvb-code-clean/SKILL.md index 5577df3..20fde13 100644 --- a/skills/kvb-code-clean/SKILL.md +++ b/skills/kvb-code-clean/SKILL.md @@ -40,11 +40,9 @@ lint proves nothing here. ## Step 2: defaults -Every parameter, option, setting, environment lookup and template variable -is supplied by its caller. A missing value aborts at the earliest moment -(argument parsing, config load, container start) with an error naming the -key and where to set it; optionality is an explicit `None`/`null` branch at -the call site, never a fallback inside the accessor. +Every parameter, setting and environment lookup comes from its caller; a +missing one aborts early with an error naming the key and where to set it. +Optionality is an explicit `None` branch at the call site. `false`, `null`, `""` and an empty list or map are free. Any other default carries a comment saying why the value is right for every caller that omits @@ -57,11 +55,14 @@ misses. ## Step 3: principles -Apply the lenses of the `kvb-check-principles` skill (SPOT, DRY, SRP, -KISS/YAGNI, security at trust boundaries, fail fast, coupling) and its -method (graph first, siblings grepped, every `file:line` read) to the -touched units. Judge what this change adds or worsens; an older violation in -a touched unit is one report line, not a fix. +Run `${CLAUDE_SKILL_DIR}/../kvb-check-principles/scripts/principles.py +--repo --ccn 10 --nloc 50 --params 5 --min-tokens 50` (`--staged`, +`--file` as above; McCabe's, pylint's and jscpd's limits, 50 NLOC one +screen). Its hits on changed lines are leads under SRP, fail fast, +KISS/YAGNI and DRY; a `skip:` line leaves that tool's lens to you. Apply the +other `kvb-check-principles` lenses (SPOT, security, coupling, DRY of +knowledge) by its method. Judge what this change adds or worsens; an older +violation is one report line, not a fix. - Fix in place a violation the change introduced when the fix stays inside the bundle and keeps behaviour, proven by the tests covering the touched @@ -75,5 +76,5 @@ a touched unit is one report line, not a fix. One line per step: comments deleted, defaults rewritten or justified, principle findings fixed and open, each with file and count. Whatever you wrote yourself is admitted without excuses, lowers the bundle's 🎯 fix and -raises its 💥 break in the `kvb-check-score` block, and goes into persistent -agent memory (date and pattern), so the calibration persists. +raises its 💥 break, and goes into persistent agent memory (date and +pattern). diff --git a/tests/test_principles.py b/tests/test_principles.py new file mode 100644 index 0000000..4ca01a3 --- /dev/null +++ b/tests/test_principles.py @@ -0,0 +1,264 @@ +"""Exercise principles.py: principle-lens tool hits on the changed lines of a temporary git repository.""" + +from __future__ import annotations + +import importlib.util +import json +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +SCRIPT = REPO_ROOT / "skills" / "kvb-check-principles" / "scripts" / "principles.py" +SPEC = importlib.util.spec_from_file_location("principles_under_test", SCRIPT) +principles = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = principles +SPEC.loader.exec_module(principles) + +GIT = shutil.which("git") +RUFF = shutil.which("ruff") +LIMITS = ("--ccn", "10", "--nloc", "50", "--params", "5", "--min-tokens", "40") +SWALLOW = "def old():\n try:\n return 1\n except Exception:\n pass\n" +ADDED = SWALLOW + "\n\ndef new(unused):\n try:\n return 2\n except Exception:\n pass\n" +RUFF_HITS = sorted( + [ + ("app.py:8", "KISS/YAGNI", "ruff", "ARG001"), + ("app.py:11", "fail fast", "ruff", "BLE001"), + ("app.py:11", "fail fast", "ruff", "S110"), + ] +) +LIZARD_TEXT = ( + "/src/app.py:3: warning: grow has 31 NLOC, 12 CCN, 140 token, 2 PARAM, 40 length, 3 ND\n" + "/src/a b.cpp:7: warning: Foo::operator () has 5 NLOC, 16 CCN, 50 token, 7 PARAM, 6 length, 1 ND\n" + "/src/x.py:1: warning: f has 1 NLOC, 20 CCN, 5 token, 0 PARAM, 1 length\n" + "================================================\n" + "No thresholds exceeded (cyclomatic_complexity > 10)\n" +) +JSCPD_REPORT = { + "duplicates": [ + { + "format": "python", + "lines": 6, + "fragment": "x = 1\n", + "tokens": 55, + "firstFile": { + "name": "/src/a.py", + "start": 1, + "end": 6, + "startLoc": {"line": 1, "column": 1, "position": 0}, + "endLoc": {"line": 6, "column": 9, "position": 80}, + }, + "secondFile": { + "name": "/src/b.py", + "start": 10, + "end": 15, + "startLoc": {"line": 10, "column": 1, "position": 120}, + "endLoc": {"line": 15, "column": 9, "position": 200}, + }, + "isNew": False, + "kind": "exact", + } + ], + "statistics": {"total": {"clones": 1}}, +} +LINES = "".join(f"x{n} = {n}\n" for n in range(1, 31)) +OTHER = "".join(f"y{n} = {n}\n" for n in range(1, 11)) +FAKE_LIZARD = """#!{python} +import json +import sys +from pathlib import Path + +Path({log!r}).write_text(json.dumps(sys.argv[1:]), encoding="utf-8") +for path in [arg for arg in sys.argv[1:] if arg.endswith(".py")]: + print(path + ":1: warning: before has 4 NLOC, 12 CCN, 40 token, 1 PARAM, 4 length, 1 ND") + print(path + ":3: warning: around has 5 NLOC, 12 CCN, 40 token, 1 PARAM, 5 length, 1 ND") + print(path + ":5: warning: single has 1 NLOC, 12 CCN, 9 token, 1 PARAM, 1 length, 1 ND") + print(path + ":6: warning: after has 5 NLOC, 12 CCN, 40 token, 1 PARAM, 5 length, 1 ND") +sys.exit(1) +""" +FAKE_JSCPD = """#!{python} +import json +import sys +from pathlib import Path + +args = sys.argv[1:] +Path({log!r}).write_text(json.dumps(args), encoding="utf-8") +Path(args[args.index("--output") + 1], "jscpd-report.json").write_text({report!r}, encoding="utf-8") +""" + + +def clone(first: tuple[str, int, int], second: tuple[str, int, int], tokens: int) -> dict: + return { + "tokens": tokens, + "firstFile": dict(zip(("name", "start", "end"), first)), + "secondFile": dict(zip(("name", "start", "end"), second)), + } + + +class TestParsers(unittest.TestCase): + def test_lizard_warnings_with_their_spans(self): + self.assertEqual( + principles.parse_lizard(LIZARD_TEXT), + [ + ("/src/app.py", 3, 42, "grow has 31 NLOC, 12 CCN, 140 token, 2 PARAM, 40 length, 3 ND"), + ("/src/a b.cpp", 7, 12, "Foo::operator () has 5 NLOC, 16 CCN, 50 token, 7 PARAM, 6 length, 1 ND"), + ("/src/x.py", 1, 1, "f has 1 NLOC, 20 CCN, 5 token, 0 PARAM, 1 length"), + ], + ) + + def test_jscpd_report_sides(self): + self.assertEqual(principles.parse_jscpd(JSCPD_REPORT), [(("/src/a.py", 1, 6), ("/src/b.py", 10, 15), 55)]) + + def test_span_overlap_bounds(self): + target = principles.Target(Path("/src/app.py"), frozenset({5})) + self.assertEqual( + [target.overlaps(*span) for span in ((1, 4), (3, 7), (5, 5), (6, 10))], [False, True, True, False] + ) + self.assertTrue(principles.Target(Path("/src/app.py"), None).overlaps(100, 200)) + + +class TestPrinciples(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.tmp = Path(tmp.name).resolve() + self.repo = self.tmp / "repo" + self.bin = self.tmp / "bin" + self.repo.mkdir() + self.bin.mkdir() + (self.bin / "git").symlink_to(GIT) + self.env = { + "PATH": str(self.bin), + "HOME": str(self.tmp), + "TMPDIR": str(self.tmp), + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_CEILING_DIRECTORIES": str(self.tmp), + "GIT_AUTHOR_NAME": "t", + "GIT_AUTHOR_EMAIL": "t@example.org", + "GIT_COMMITTER_NAME": "t", + "GIT_COMMITTER_EMAIL": "t@example.org", + } + self._git("init", "-q", "-b", "main") + + def _git(self, *args: str) -> str: + return subprocess.run( + ["git", *args], cwd=self.repo, env=self.env, capture_output=True, text=True, check=True + ).stdout + + def _write(self, files: dict[str, str]) -> None: + for name, text in files.items(): + (self.repo / name).write_text(text, encoding="utf-8") + + def _commit(self, files: dict[str, str]) -> None: + self._write(files) + self._git("add", ".") + self._git("commit", "-q", "-m", "base") + + def _fake(self, name: str, template: str, **values: str) -> Path: + log = self.tmp / f"{name}.argv" + script = self.bin / name + script.write_text(template.format(python=sys.executable, log=str(log), **values), encoding="utf-8") + script.chmod(0o755) + return log + + def _run(self, *args: str) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, str(SCRIPT), *args], env=self.env, capture_output=True, text=True, check=False + ) + + def _hits(self, *args: str) -> tuple[int, list[tuple]]: + result = self._run("--repo", str(self.repo), "--json", *LIMITS, *args) + self.assertIn(result.returncode, (0, 1), result.stderr) + rows = json.loads(result.stdout) + return result.returncode, sorted( + (f"{r['file']}:{r['line']}", r["lens"], r["tool"], r["message"].split()[0]) for r in rows + ) + + def test_spans_and_clone_sides_filter_to_changed_lines(self): + self._commit({"app.py": LINES, "other.py": OTHER}) + self._write({"app.py": LINES.replace("x5 = 5\n", "x5 = 50\n")}) + app, other = str(self.repo / "app.py"), str(self.repo / "other.py") + report = { + "duplicates": [ + clone((app, 4, 6), (other, 1, 3), 50), + clone((app, 20, 25), (other, 1, 6), 60), + clone((other, 2, 4), (app, 1, 5), 70), + clone((other, 1, 3), (other, 5, 7), 80), + ] + } + lizard_log = self._fake("lizard", FAKE_LIZARD) + jscpd_log = self._fake("jscpd", FAKE_JSCPD, report=json.dumps(report)) + result = self._run("--repo", str(self.repo), *LIMITS) + self.assertEqual(result.returncode, 1, result.stderr) + self.assertEqual(result.stderr, "skip: ruff missing\n") + self.assertEqual( + result.stdout, + "app.py:1\tDRY\tjscpd\tclone of other.py:2-4, 70 tokens\n" + "app.py:3\tSRP\tlizard\taround has 5 NLOC, 12 CCN, 40 token, 1 PARAM, 5 length, 1 ND\n" + "app.py:4\tDRY\tjscpd\tclone of other.py:1-3, 50 tokens\n" + "app.py:5\tSRP\tlizard\tsingle has 1 NLOC, 12 CCN, 9 token, 1 PARAM, 1 length, 1 ND\n", + ) + self.assertEqual( + json.loads(lizard_log.read_text(encoding="utf-8")), + ["--warnings_only", "--CCN", "10", "--arguments", "5", "--Threshold", "nloc=50", app], + ) + jscpd_args = json.loads(jscpd_log.read_text(encoding="utf-8")) + self.assertEqual(jscpd_args[:4], ["--min-tokens", "40", "--reporters", "json"]) + self.assertEqual(jscpd_args[6:], ["--absolute", "--silent", str(self.repo)]) + + def test_missing_tools_are_skipped(self): + self._commit({"app.py": SWALLOW}) + self._write({"app.py": ADDED}) + result = self._run("--repo", str(self.repo), *LIMITS) + self.assertEqual( + (result.returncode, result.stdout, result.stderr), + (0, "", "skip: ruff missing\nskip: lizard missing\nskip: jscpd missing\n"), + ) + + @unittest.skipUnless(RUFF, "ruff not installed") + def test_ruff_flags_added_swallow_not_the_untouched_one(self): + (self.bin / "ruff").symlink_to(RUFF) + self._commit({"app.py": SWALLOW}) + self._write({"app.py": ADDED}) + code, hits = self._hits() + self.assertEqual((code, hits), (1, RUFF_HITS)) + self.assertNotIn("app.py:4", [hit[0] for hit in hits]) + + @unittest.skipUnless(RUFF, "ruff not installed") + def test_staged_reads_the_index_snapshot(self): + (self.bin / "ruff").symlink_to(RUFF) + self._commit({"app.py": SWALLOW}) + self._write({"app.py": ADDED}) + self._git("add", "app.py") + self._write({"app.py": SWALLOW}) + self.assertEqual(self._hits("--staged"), (1, RUFF_HITS)) + self.assertEqual(self._hits(), (0, [])) + + @unittest.skipUnless(RUFF, "ruff not installed") + def test_file_counts_every_line_of_an_untracked_file(self): + (self.bin / "ruff").symlink_to(RUFF) + self._commit({"app.py": SWALLOW}) + self._write({"extra.py": SWALLOW}) + self.assertEqual(self._hits(), (0, [])) + self.assertEqual( + self._hits("--file", "extra.py"), + (1, [("extra.py:4", "fail fast", "ruff", "BLE001"), ("extra.py:4", "fail fast", "ruff", "S110")]), + ) + + def test_usage_and_git_errors(self): + self._commit({"app.py": SWALLOW}) + missing = self._run("--repo", str(self.repo), "--ccn", "10", "--nloc", "50", "--params", "5") + self.assertEqual(missing.returncode, 2) + self.assertIn("--min-tokens", missing.stderr) + self.assertEqual(self._run("--repo", str(self.repo)).returncode, 2) + outside = self.tmp / "plain" + outside.mkdir() + self.assertEqual(self._run("--repo", str(outside), *LIMITS).returncode, 2) + self.assertEqual(self._run("--repo", str(self.repo), *LIMITS, "--file", "missing.py").returncode, 2) + + +if __name__ == "__main__": + unittest.main()