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()