feat(principles): principles.py turns the mechanical lenses into tool hits
🧪 Test / 🧪 Lock + lint (push) Waiting to run

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 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed 2026-11-04 21:48:58 +01:00
1 parent 3dc80a1954
commit 96aaa17e60
5 files changed
+637 -17

No files matched your search

+20
View File
@@ -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."
+5 -5
View File
@@ -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.
+335
View File
@@ -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: <tool> 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<path>.+?):(?P<line>\d+): warning: "
r"(?P<message>.+? has \d+ NLOC, \d+ CCN, \d+ token, \d+ PARAM, (?P<length>\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())
+13 -12
View File
@@ -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 <dir> --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).
+264
View File
@@ -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()