Add a multi-language CRAP tool that writes .metrics/crap.edn for uml-viewer.

Score Clojure, Java, Go, TypeScript, Rust, and Python with each language's complexity rules and coverage report.
This commit is contained in:
Robert C. Martin
2026-09-30 08:21:44 -05:00
commit a4b5fc2aed
33 changed files with 3913 additions and 0 deletions
+8
View File
@@ -0,0 +1,8 @@
.venv/
__pycache__/
.pytest_cache/
*.egg-info/
.metrics/
target/
dist/
build/
+93
View File
@@ -0,0 +1,93 @@
# crapper
CRAP scores for Clojure, Java, Go, TypeScript, Rust, and Python. One run detects the language of each source file, applies that language's complexity and coverage rules, and writes the snapshot [uml-viewer](https://github.com/unclebob/uml-viewer) already reads.
The formula is the one from [crap4clj](https://github.com/unclebob/crap4clj), [crap4java](https://github.com/unclebob/crap4java), and [crap4go](https://github.com/unclebob/crap4go):
```text
CRAP = CC² × (1 − coverage)³ + CC
```
`CC` is cyclomatic complexity. `coverage` is the fraction of the function exercised by tests. A score of 1–5 is low risk, 5–30 is worth a look, and 30+ is complex and under-tested.
## Run
```bash
./crapper
```
The first run creates `.venv` and installs the tool. From a project root it walks the tree, skips `test`, `spec`, `vendor`, `node_modules`, and `target`, and writes two results:
- a table on stdout, worst score first
- `.metrics/crap.edn`, replaced on every successful analysis
```bash
./crapper --no-coverage # complexity only; coverage is N/A
./crapper --use-existing-coverage # read reports already on disk
./crapper src/demo/core.clj src/ui # these files and trees
./crapper --changed # git additions and edits
./crapper combat # path fragment, same idea as crap4clj
./crapper --threshold 30 # exit 2 when the worst score is higher
```
## Snapshot
`.metrics/crap.edn` has the same shape crap4clj writes:
```clojure
{:entries [{:name "place"
:namespace "demo.Board"
:complexity 3
:coverage 75.0
:crap 3.1}]}
```
uml-viewer groups entries by `:namespace` and joins each operation on `:name`. `nil` coverage means the function was found but no coverage data applied, and CRAP is `nil` too.
| Language | `:namespace` | `:name` |
| --- | --- | --- |
| Clojure | the `ns` | the `defn` / `defn-` name |
| Java | `package.Class`, or `package.Outer.Inner` | the method name |
| Go | the package import path, or `import/path.Receiver` | the function or method name |
| TypeScript | the dotted module path, or `module.Class` | the function or method name |
| Rust | `crate::module`, or `crate::module::Type` | the function or method name |
| Python | the dotted module path, or `module.Class` | the function or method name |
Rename or move is a new entry. There is no identity matching across runs.
## What each language counts
Clojure follows crap4clj: `if` / `when` and their variants, `and`, `or`, `loop`, `catch`, and each clause of `cond`, `condp`, `case`, `cond->`, `cond->>`, `some->`, and `some->>`. Coverage prefers Cloverage's per-line form counts and falls back to LCOV.
Java follows crap4java: `if`, loops, `catch`, `?:`, each `switch` label, and `&&` / `||`. Constructors and methods inside anonymous classes are omitted. Coverage is JaCoCo's `INSTRUCTION` counter.
Go follows crap4go: `if`, `for`, `range`, each `switch` and `select` clause, and `&&` / `||`. Coverage is `go test -coverprofile`.
TypeScript and TSX use the same structural decisions as Java, including `&&` inside JSX. Top-level functions, class methods, and top-level arrow functions are entries. Nested callbacks stay inside the enclosing function. Coverage is LCOV. A `coverage` script is used as-is. A Vitest project runs `vitest --coverage` (installing `@vitest/coverage-v8` into `node_modules` when it is missing, without editing `package.json`). Other test scripts are wrapped in `c8`.
Rust counts `if`, loops, each `match` arm, `?`, and `&&` / `||`. `mod tests` is skipped. Coverage is LCOV from `cargo llvm-cov` or `cargo tarpaulin`, run in the nearest directory that contains `Cargo.toml`. When neither tool is installed, the run installs `cargo-llvm-cov`.
Python counts `if`, `elif`, `for`, `while`, `except`, each `match` case, a comprehension filter, a conditional expression, and each `and` / `or`. Nested functions stay inside the enclosing function. Coverage is LCOV from `coverage.py`, running pytest when the project uses it and `unittest` otherwise. `coverage` and `pytest` are installed into the project's interpreter when they are missing.
## Coverage commands
By default a run deletes the previous report for each language it is about to measure and regenerates it:
| Language | Command | Report |
| --- | --- | --- |
| Clojure | `clj -M:cov --lcov` (needs `deps.edn` or `bb.edn`) | `target/coverage/` |
| Java | JaCoCo Maven plugin `0.8.12` in each module with `pom.xml` | `target/site/jacoco/jacoco.xml` |
| Go | `go test ./... -coverprofile=...` | `target/coverage/go/coverage.out` |
| TypeScript | `npm run coverage`, or Vitest `--coverage`, or `npx c8 ... npm test` | `coverage/lcov.info` or `target/coverage/typescript/lcov.info` |
| Rust | `cargo llvm-cov` or `cargo tarpaulin`, per Cargo package | `target/coverage/rust/lcov.info` |
| Python | `coverage run` with pytest or unittest, then `coverage lcov` | `target/coverage/python/lcov.info` |
A missing tool or a failed test run leaves that language at N/A and still writes the snapshot. Pass `--coverage-command` to replace those defaults with one command of your own.
## Development
```bash
python3 -m venv .venv
.venv/bin/pip install -e '.[dev]'
.venv/bin/pytest
```
Executable
+9
View File
@@ -0,0 +1,9 @@
#!/bin/sh
# Run the CRAP tool, creating a local virtualenv on first use.
ROOT=$(CDPATH= cd -- "$(dirname "$0")" && pwd)
PY="$ROOT/.venv/bin/python"
if [ ! -x "$PY" ]; then
python3 -m venv "$ROOT/.venv"
"$ROOT/.venv/bin/pip" install -e "$ROOT"
fi
exec "$PY" -m crapper "$@"
+26
View File
@@ -0,0 +1,26 @@
[project]
name = "crapper"
version = "0.1.0"
description = "CRAP scores for Clojure, Java, Go, TypeScript, Rust, and Python, in the snapshot uml-viewer reads."
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"tree-sitter>=0.24",
"tree-sitter-language-pack>=0.7",
]
[project.scripts]
crapper = "crapper.cli:main"
[project.optional-dependencies]
dev = ["pytest>=8"]
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
testpaths = ["tests"]
+3
View File
@@ -0,0 +1,3 @@
"""CRAP scores for Clojure, Java, Go, TypeScript, Rust, and Python."""
__version__ = "0.1.0"
+4
View File
@@ -0,0 +1,4 @@
from crapper.cli import main
if __name__ == "__main__":
main()
+32
View File
@@ -0,0 +1,32 @@
"""Turn source files and coverage into CRAP entries."""
from pathlib import Path
from crapper.coverage import CoverageBundle
from crapper.crap import make_entry, sort_entries
from crapper.discover import language_of
from crapper.languages import functions_in_file
from crapper.model import Entry
def analyze_files(
files: list[Path],
project_root: Path,
bundle: CoverageBundle | None,
) -> list[Entry]:
root = project_root.resolve()
entries: list[Entry] = []
for file in files:
file = file.resolve()
language = language_of(file)
if language is None:
continue
source = file.read_text(encoding="utf-8", errors="replace")
try:
relative = file.relative_to(root).as_posix()
except ValueError:
relative = file.as_posix()
for function in functions_in_file(language, source, relative, str(root)):
coverage = None if bundle is None else bundle.percent_for(function)
entries.append(make_entry(function, coverage))
return sort_entries(entries)
+259
View File
@@ -0,0 +1,259 @@
"""Command line for the multi-language CRAP tool."""
from __future__ import annotations
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path
from crapper.analyze import analyze_files
from crapper.coverage import load_bundle
from crapper.discover import is_test_file, iter_source_files, language_of
from crapper.metrics import write_metrics
from crapper.report import format_report
from crapper.runners import run_coverage
HELP = """\
Usage: crapper [options] [path-or-filter ...]
Detect the language of each source file and score it with the CRAP metric
CRAP = CC² × (1 − coverage)³ + CC. Writes .metrics/crap.edn for uml-viewer
and prints a report sorted worst first.
Languages: Clojure (.clj .cljc .cljs .bb), Java (.java), Go (.go),
TypeScript (.ts .tsx .mts .cts), Rust (.rs), Python (.py).
Options:
-h, --help Print this help and exit.
--root <path> Project root. Metrics are written here.
Default: the current directory.
-s, --source-root <path> Walk this tree instead of the project root.
May be repeated.
--changed Analyze added and modified source files from
git status.
--no-coverage Score complexity only. Coverage and CRAP are N/A.
--use-existing-coverage Read coverage already on disk. Do not rerun tests.
--coverage-command <cmd> Run this command instead of the per-language
coverage tools, then read the reports it wrote.
--threshold <number> Exit 2 when the worst CRAP score is above this.
Arguments:
path File or directory to analyze. Test paths are included when
you name them explicitly.
filter When the argument is not a path, only source files whose
path contains this text are analyzed.
With no paths, source files under the project root are analyzed. Directories
named test, tests, spec, specs, vendor, node_modules, and target are skipped,
as are *_test.go, *.spec.ts, test_*.py, and *_test.py files.
Coverage, when it is produced:
Clojure clj -M:cov --lcov, then Cloverage form counts or LCOV
Java Maven JaCoCo instruction coverage, for each pom.xml module
Go go test ./... -coverprofile=...
TypeScript npm run coverage, Vitest's own LCOV, or c8 around npm test
Rust cargo llvm-cov, or cargo tarpaulin, in the nearest Cargo.toml
Python coverage.py LCOV, via pytest or unittest, per project
uml-viewer joins .metrics/crap.edn on :namespace and :name.
Clojure the ns, and the defn name
Java package.Class, and the method name
Go import path, or import.path.Receiver for methods
TypeScript dotted module path, or module.Class for methods
Rust crate::module, or crate::module::Type for methods
Python dotted module path, or module.Class for methods
"""
@dataclass
class Options:
action: str
message: str = ""
exit_code: int = 0
project_root: Path = field(default_factory=lambda: Path("."))
source_roots: list[str] = field(default_factory=list)
positionals: list[str] = field(default_factory=list)
no_coverage: bool = False
use_existing_coverage: bool = False
coverage_command: str | None = None
threshold: float | None = None
changed: bool = False
def _take(args: list[str], index: int, option: str) -> str:
if index + 1 >= len(args) or not args[index + 1] or args[index + 1].startswith("-"):
raise ValueError(f"{option} requires a value")
return args[index + 1]
def parse_args(argv: list[str] | None = None) -> Options:
args = list(sys.argv[1:] if argv is None else argv)
if any(arg in {"-h", "--help"} for arg in args):
return Options(action="help", message=HELP, exit_code=0)
options = Options(action="analyze")
index = 0
try:
while index < len(args):
arg = args[index]
if arg in {"-s", "--source-root"}:
options.source_roots.append(_take(args, index, arg))
index += 2
continue
if arg == "--root":
options.project_root = Path(_take(args, index, arg))
index += 2
continue
if arg == "--coverage-command":
options.coverage_command = _take(args, index, arg)
index += 2
continue
if arg == "--threshold":
raw = _take(args, index, arg)
try:
options.threshold = float(raw)
except ValueError as exc:
raise ValueError("--threshold requires a number") from exc
index += 2
continue
if arg == "--no-coverage":
options.no_coverage = True
index += 1
continue
if arg == "--use-existing-coverage":
options.use_existing_coverage = True
index += 1
continue
if arg == "--changed":
options.changed = True
index += 1
continue
if arg.startswith("-"):
raise ValueError(f"Unknown option: {arg}")
options.positionals.append(arg)
index += 1
except ValueError as exc:
return Options(action="help", message=f"{exc}\n\n{HELP}", exit_code=1)
if options.no_coverage and options.coverage_command:
return Options(
action="help",
message=f"--no-coverage cannot be combined with --coverage-command\n\n{HELP}",
exit_code=1,
)
return options
def _changed_files(root: Path) -> list[Path]:
result = subprocess.run(
["git", "status", "--porcelain"],
cwd=root,
check=False,
capture_output=True,
text=True,
)
if result.returncode != 0:
print(result.stderr.strip() or "git status failed", file=sys.stderr)
return []
found: list[Path] = []
for line in result.stdout.splitlines():
if len(line) < 4:
continue
path_text = line[3:].strip()
if " -> " in path_text:
path_text = path_text.split(" -> ", 1)[1]
path_text = path_text.strip('"')
found.append((root / path_text).resolve())
return found
def _positionals(root: Path, args: list[str]) -> tuple[list[Path], list[str]]:
existing: list[Path] = []
filters: list[str] = []
for arg in args:
candidate = Path(arg)
if not candidate.is_absolute():
candidate = root / arg
if candidate.exists():
existing.append(candidate.resolve())
else:
filters.append(arg)
return existing, filters
def _changed_source(root: Path) -> list[Path]:
return [
path
for path in _changed_files(root)
if language_of(path) is not None and not is_test_file(path)
]
def _explicit_files(existing: list[Path]) -> list[Path]:
files: list[Path] = []
for path in existing:
if path.is_dir():
files.extend(iter_source_files([path]))
elif language_of(path) is not None:
files.append(path)
return files
def _selected(options: Options, root: Path, existing: list[Path]) -> list[Path]:
if options.changed:
return _changed_source(root)
if options.source_roots:
return iter_source_files([(root / path).resolve() for path in options.source_roots])
if existing:
return _explicit_files(existing)
return iter_source_files([root])
def _apply_filters(files: list[Path], filters: list[str]) -> list[Path]:
if not filters:
return files
return [path for path in files if any(item in path.as_posix() for item in filters)]
def select_files(options: Options) -> list[Path]:
root = options.project_root.resolve()
existing, filters = _positionals(root, options.positionals)
files = _apply_filters(_selected(options, root, existing), filters)
return sorted({path.resolve() for path in files}, key=lambda path: path.as_posix())
def run(argv: list[str] | None = None) -> int:
options = parse_args(argv)
if options.action == "help":
stream = sys.stdout if options.exit_code == 0 else sys.stderr
print(options.message, file=stream, end="" if options.message.endswith("\n") else "\n")
return options.exit_code
root = options.project_root.resolve()
files = select_files(options)
if not files:
print("No source files to analyze.")
return 0
bundle = None
if not options.no_coverage:
if not options.use_existing_coverage:
run_coverage(root, files, options.coverage_command)
bundle = load_bundle(root)
entries = analyze_files(files, root, bundle)
metrics = write_metrics(entries, root)
print(format_report(entries), end="", flush=True)
print(f"Wrote {metrics}", file=sys.stderr)
if options.threshold is not None:
scored = [entry.crap for entry in entries if entry.crap is not None]
if scored and max(scored) > options.threshold:
print(
f"CRAP threshold exceeded: {max(scored):.1f} > {options.threshold:.1f}",
file=sys.stderr,
)
return 2
return 0
def main(argv: list[str] | None = None) -> None:
sys.exit(run(argv))
+375
View File
@@ -0,0 +1,375 @@
"""Load Cloverage, LCOV, JaCoCo, and Go coverage profiles.
Clojure prefers Cloverage's per-line form counts, then LCOV. Java uses JaCoCo
instruction counters. Go uses statement profiles. TypeScript and Rust use LCOV
line hits. A file with no coverage data scores N/A rather than 0%.
"""
from __future__ import annotations
import re
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from pathlib import Path
from urllib.parse import unquote
from crapper.model import Function
_SPAN = re.compile(
r'<span[^>]*title="(\d+) out of (\d+) forms covered"[^>]*>\s*(\d+)&nbsp;'
)
_LCOV_DA = re.compile(r"DA:(\d+),(\d+)")
_DOCTYPE = re.compile(r"<!DOCTYPE[^>]*>", re.IGNORECASE)
def normalize_path(path: str) -> str:
text = unquote(path).replace("\\", "/")
if text.startswith("file:"):
text = text[5:]
while text.startswith("./"):
text = text[2:]
while "//" in text:
text = text.replace("//", "/")
return text
def _segments(path: str) -> list[str]:
return [part for part in normalize_path(path).split("/") if part]
def suffix_match(path: str, suffix: str) -> bool:
path_parts = _segments(path)
suffix_parts = _segments(suffix)
if not suffix_parts or len(suffix_parts) > len(path_parts):
return False
return path_parts[-len(suffix_parts) :] == suffix_parts
def _candidates(source_path: str) -> list[str]:
relative = normalize_path(source_path)
absolute = normalize_path(str(Path(source_path).resolve())) if source_path else relative
no_src = re.sub(r"^src/", "", relative)
absolute_no_src = re.sub(r"/src/", "/", absolute)
values = [relative, absolute, no_src, absolute_no_src]
found: list[str] = []
for value in values:
if value and value not in found:
found.append(value)
return found
def coverage_percent(covered: int, total: int) -> float:
if total <= 0:
return 0.0
return 100.0 * covered / total
def percent_for_range(lines: dict[int, tuple[int, int]] | None, start: int, end: int) -> float | None:
if lines is None:
return None
covered = 0
total = 0
for line in range(start, end + 1):
entry = lines.get(line)
if entry is None:
continue
covered += entry[0]
total += entry[1]
if total == 0:
return 0.0
return coverage_percent(covered, total)
def parse_form_coverage(html: str) -> dict[int, tuple[int, int]]:
found: dict[int, tuple[int, int]] = {}
for covered, total, line in _SPAN.findall(html):
total_forms = int(total)
if total_forms <= 0:
continue
found[int(line)] = (int(covered), total_forms)
return found
def parse_lcov(text: str) -> dict[str, dict[int, tuple[int, int]]]:
out: dict[str, dict[int, tuple[int, int]]] = {}
current_file = None
current: dict[int, tuple[int, int]] = {}
for raw in text.splitlines():
line = raw.strip()
if line.startswith("SF:"):
if current_file is not None:
out[current_file] = current
current_file = line[3:]
current = {}
elif line == "end_of_record":
if current_file is not None:
out[current_file] = current
current_file = None
current = {}
elif current_file is not None:
match = _LCOV_DA.match(line)
if match:
hits = int(match.group(2))
current[int(match.group(1))] = (1 if hits > 0 else 0, 1)
if current_file is not None:
out[current_file] = current
return out
def parse_go_profile(text: str) -> dict[str, list[tuple[int, int, int, int]]]:
"""Map file → (start_line, end_line, statements, hits)."""
out: dict[str, list[tuple[int, int, int, int]]] = {}
for line_number, raw in enumerate(text.splitlines(), start=1):
line = raw.strip()
if not line or line.startswith("mode:"):
continue
fields = line.split()
if len(fields) != 3 or ":" not in fields[0]:
raise ValueError(f"invalid coverage segment on line {line_number}: {raw!r}")
file_part, range_part = fields[0].split(":", 1)
start_text, end_text = range_part.split(",", 1)
start_line = int(start_text.split(".", 1)[0])
end_line = int(end_text.split(".", 1)[0])
statements = int(fields[1])
hits = int(fields[2])
out.setdefault(file_part, []).append((start_line, end_line, statements, hits))
return out
def _profile_segments(profile, path: str):
if profile is None:
return None
for candidate in _candidates(path):
for key, value in profile.items():
if (
normalize_path(key) == candidate
or suffix_match(key, candidate)
or suffix_match(candidate, key)
):
return value
return None
def go_percent(
profile: dict[str, list[tuple[int, int, int, int]]] | None,
path: str,
start: int,
end: int,
) -> float | None:
segments = _profile_segments(profile, path)
if segments is None:
return None
total = 0
covered = 0
for seg_start, seg_end, statements, hits in segments:
if seg_end < start or seg_start > end:
continue
total += statements
if hits > 0:
covered += statements
if total == 0:
return 0.0
return coverage_percent(covered, total)
@dataclass
class JacocoMethod:
missed: int
covered: int
line: int
@property
def percent(self) -> float:
return coverage_percent(self.covered, self.missed + self.covered)
def _jacoco_from_index(classes, class_names, method_name, line) -> float | None:
pool: list[JacocoMethod] = []
for name in class_names:
for method in classes.get(f"{name}#{method_name}", []):
pool.append(method)
if not pool:
return None
for method in pool:
if method.line == line:
return method.percent
nearest = min(pool, key=lambda method: abs(method.line - line))
return nearest.percent
def parse_jacoco_index(text: str) -> dict[str, list[JacocoMethod]]:
"""Key methods as `binary.class.name#method`."""
cleaned = _DOCTYPE.sub("", text)
root = ET.fromstring(cleaned)
found: dict[str, list[JacocoMethod]] = {}
for class_node in root.iter("class"):
class_name = class_node.get("name", "").replace("/", ".")
for method in list(class_node):
if method.tag != "method":
continue
counter = next(
(
child
for child in list(method)
if child.tag == "counter" and child.get("type") == "INSTRUCTION"
),
None,
)
if counter is None:
continue
try:
line = int(method.get("line") or "0")
except ValueError:
line = 0
key = f"{class_name}#{method.get('name', '')}"
found.setdefault(key, []).append(
JacocoMethod(
missed=int(counter.get("missed") or "0"),
covered=int(counter.get("covered") or "0"),
line=line,
)
)
return found
@dataclass
class CoverageBundle:
lcov: dict[str, dict[int, tuple[int, int]]] = field(default_factory=dict)
go_profile: dict[str, list[tuple[int, int, int, int]]] | None = None
jacoco: dict[str, list[JacocoMethod]] | None = None
form_html: dict[str, dict[int, tuple[int, int]]] = field(default_factory=dict)
def percent_for(self, function: Function) -> float | None:
if function.language == "java":
names = []
if function.jacoco_class:
names.append(function.jacoco_class)
if function.namespace not in names:
names.append(function.namespace)
return _jacoco_from_index(
self.jacoco or {}, names, function.name, function.start_line
)
if function.language == "go":
return go_percent(
self.go_profile, function.path, function.start_line, function.end_line
)
html = self._html_lines(function.path)
if html is not None:
return percent_for_range(html, function.start_line, function.end_line)
return percent_for_range(
self._lcov_lines(function.path), function.start_line, function.end_line
)
def _html_lines(self, path: str) -> dict[int, tuple[int, int]] | None:
return _lookup(self.form_html, path)
def _lcov_lines(self, path: str) -> dict[int, tuple[int, int]] | None:
return _lookup(self.lcov, path)
def _lookup(index: dict[str, dict], source_path: str):
if not index:
return None
normalized = {normalize_path(key): value for key, value in index.items()}
for candidate in _candidates(source_path):
if candidate in normalized:
return normalized[candidate]
for key, value in normalized.items():
if any(suffix_match(key, candidate) for candidate in _candidates(source_path)):
return value
return None
def _read(path: Path) -> str | None:
if not path.is_file():
return None
return path.read_text(encoding="utf-8", errors="replace")
def _lcov_paths(root: Path) -> list[Path]:
paths = [
root / "target" / "coverage" / "lcov.info",
root / "coverage" / "lcov.info",
root / "target" / "coverage" / "typescript" / "lcov.info",
root / "target" / "coverage" / "rust" / "lcov.info",
root / "target" / "coverage" / "python" / "lcov.info",
]
paths.extend(root.glob("target/coverage/**/lcov.info"))
paths.extend(root.glob("coverage/**/lcov.info"))
return paths
def _merge_texts(paths: list[Path]) -> list[str]:
texts: list[str] = []
seen: set[Path] = set()
for path in paths:
resolved = path.resolve()
if resolved in seen:
continue
seen.add(resolved)
text = _read(path)
if text:
texts.append(text)
return texts
def _merge_lcov(root: Path) -> dict[str, dict[int, tuple[int, int]]]:
merged: dict[str, dict[int, tuple[int, int]]] = {}
for text in _merge_texts(_lcov_paths(root)):
merged.update(parse_lcov(text))
return merged
def _merge_go(root: Path) -> dict[str, list[tuple[int, int, int, int]]]:
paths = [
root / "target" / "coverage" / "coverage.out",
root / "target" / "coverage" / "go" / "coverage.out",
root / "coverage.out",
]
paths.extend(root.glob("*/target/coverage/go/coverage.out"))
paths.extend(root.glob("*/*/target/coverage/go/coverage.out"))
merged: dict[str, list[tuple[int, int, int, int]]] = {}
for text in _merge_texts(paths):
for key, segments in parse_go_profile(text).items():
merged.setdefault(key, []).extend(segments)
return merged
def _merge_jacoco(root: Path) -> dict[str, list[JacocoMethod]]:
paths = [root / "target" / "site" / "jacoco" / "jacoco.xml"]
paths.extend(root.glob("*/target/site/jacoco/jacoco.xml"))
paths.extend(root.glob("*/*/target/site/jacoco/jacoco.xml"))
merged: dict[str, list[JacocoMethod]] = {}
for text in _merge_texts(paths):
for key, methods in parse_jacoco_index(text).items():
merged.setdefault(key, []).extend(methods)
return merged
def _merge_forms(root: Path) -> dict[str, dict[int, tuple[int, int]]]:
coverage_dir = root / "target" / "coverage"
found: dict[str, dict[int, tuple[int, int]]] = {}
if not coverage_dir.is_dir():
return found
for html_path in coverage_dir.rglob("*.html"):
text = _read(html_path)
if not text or "forms covered" not in text:
continue
relative = html_path.relative_to(coverage_dir).as_posix()
if relative.endswith(".html"):
relative = relative[: -len(".html")]
found[relative] = parse_form_coverage(text)
return found
def load_bundle(root: Path) -> CoverageBundle:
go_profile = _merge_go(root)
jacoco = _merge_jacoco(root)
return CoverageBundle(
lcov=_merge_lcov(root),
go_profile=go_profile or None,
jacoco=jacoco or None,
form_html=_merge_forms(root),
)
+47
View File
@@ -0,0 +1,47 @@
"""CRAP formula shared by crap4clj, crap4java, and crap4go."""
from crapper.model import Entry, Function
def crap_score(complexity: int, coverage_pct: float | None) -> float | None:
"""CRAP = CC² × (1 − coverage)³ + CC.
`coverage_pct` is a percentage from 0 to 100. Missing coverage yields no
score, which the report shows as N/A.
"""
if coverage_pct is None:
return None
cc = float(complexity)
uncovered = 1.0 - (float(coverage_pct) / 100.0)
return (cc * cc * uncovered * uncovered * uncovered) + cc
def round_metric(value: float | None) -> float | None:
if value is None:
return None
return round(float(value), 4)
def make_entry(function: Function, coverage: float | None) -> Entry:
coverage = round_metric(coverage)
return Entry(
name=function.name,
namespace=function.namespace,
complexity=function.complexity,
coverage=coverage,
crap=round_metric(crap_score(function.complexity, coverage)),
path=function.path,
language=function.language,
)
def sort_entries(entries: list[Entry]) -> list[Entry]:
"""Worst CRAP first. Unscored rows follow scored rows."""
def key(entry: Entry) -> tuple:
if entry.crap is None:
return (1, -entry.complexity, entry.namespace, entry.name, entry.path)
return (0, -entry.crap, entry.namespace, entry.name, entry.path)
return sorted(entries, key=key)
+100
View File
@@ -0,0 +1,100 @@
"""Find source files and decide which language each one is written in."""
import os
from pathlib import Path
EXTENSIONS = {
".clj": "clojure",
".cljc": "clojure",
".cljs": "clojure",
".bb": "clojure",
".java": "java",
".go": "go",
".ts": "typescript",
".tsx": "typescript",
".mts": "typescript",
".cts": "typescript",
".rs": "rust",
".py": "python",
}
SKIP_DIRS = {
".git",
".hg",
".svn",
".idea",
".venv",
"venv",
"node_modules",
"vendor",
"target",
"dist",
"build",
"out",
"coverage",
".metrics",
"testdata",
"__pycache__",
".clj-kondo",
}
TEST_DIRS = {"test", "tests", "spec", "specs", "__tests__"}
def language_of(path: str | Path) -> str | None:
file_path = Path(path)
name = file_path.name
if name.endswith(".d.ts"):
return None
return EXTENSIONS.get(file_path.suffix)
def is_test_file(path: str | Path) -> bool:
file_path = Path(path)
name = file_path.name
if name.endswith("_test.go"):
return True
if name.endswith(("_test.clj", "_test.cljc", "_test.cljs", "_test.bb")):
return True
if name.endswith(
(
".test.ts",
".spec.ts",
".test.tsx",
".spec.tsx",
".test.mts",
".spec.mts",
)
):
return True
if name == "conftest.py":
return True
if name.endswith("_test.py") or (name.startswith("test_") and name.endswith(".py")):
return True
return any(part in TEST_DIRS for part in file_path.parts)
def _walk(root: Path) -> list[Path]:
if not root.exists():
return []
if root.is_file():
return [root]
found: list[Path] = []
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = [
name for name in dirnames if name not in SKIP_DIRS and name not in TEST_DIRS
]
for name in filenames:
path = Path(dirpath) / name
if language_of(path) is None or is_test_file(path):
continue
found.append(path)
return found
def iter_source_files(roots: list[Path]) -> list[Path]:
files: set[Path] = set()
for root in roots:
for path in _walk(root):
files.add(path.resolve())
return sorted(files, key=lambda path: path.as_posix())
+22
View File
@@ -0,0 +1,22 @@
"""Per-language function extraction."""
from crapper.languages import clojure, golang, java, python, rust, typescript
from crapper.model import Function
def functions_in_file(
language: str, source: str, path: str, project_root: str
) -> list[Function]:
if language == "clojure":
return clojure.functions_in_source(source, path, project_root)
if language == "java":
return java.functions_in_source(source, path)
if language == "go":
return golang.functions_in_source(source, path, project_root)
if language == "typescript":
return typescript.functions_in_source(source, path, project_root)
if language == "python":
return python.functions_in_source(source, path, project_root)
if language == "rust":
return rust.functions_in_source(source, path, project_root)
return []
+287
View File
@@ -0,0 +1,287 @@
"""Clojure cyclomatic complexity, ported from crap4clj.
Decision points are the same forms crap4clj counts: `if` / `when` and their
variants, `and`, `or`, `loop`, `catch`, and each clause of `cond`, `condp`,
`case`, `cond->`, `cond->>`, `some->`, and `some->>`.
"""
import re
from crapper.model import Function
_DECISION = re.compile(
r"\((if-not|if-let|if-some|when-not|when-let|when-some|when-first|if|when|and|or|loop|catch)[\s\)]"
)
_COND = re.compile(r"\((some->>|some->|cond->>|cond->|cond|condp|case)[\s\)]")
_DEFN = re.compile(r"(?s)^\(\s*defn-?\s+([^\s\(\)\[\]\{\}\"]+)")
_NS = re.compile(r"\(\s*ns\s+([A-Za-z0-9*+!_?.\-/]+)")
_IN_NS = re.compile(r"\(\s*in-ns\s+'([A-Za-z0-9*+!_?.\-/]+)\s*\)")
_IN_NS_QUOTE = re.compile(
r"\(\s*in-ns\s+\(quote\s+([A-Za-z0-9*+!_?.\-/]+)\)\s*\)"
)
_FORM_SKIP = {
"condp": 2,
"case": 1,
"cond->": 1,
"cond->>": 1,
"some->": 1,
"some->>": 1,
}
_THREAD_FORMS = {"some->", "some->>"}
_CHAR_DELIMITERS = set("()[]{}\";,")
def _strip_strings(text: str) -> str:
mode = "normal"
escaped = False
out: list[str] = []
for ch in text:
newline = ch == "\n"
if mode == "string":
if escaped:
out.append(ch if newline else " ")
escaped = False
elif ch == "\\":
out.append(" ")
escaped = True
elif ch == '"':
out.append(ch)
mode = "normal"
else:
out.append(ch if newline else " ")
elif ch == '"':
out.append(ch)
mode = "string"
else:
out.append(ch)
return "".join(out)
def _strip_comments(text: str) -> str:
return "\n".join(re.sub(r";.*", "", line) for line in text.splitlines())
def without_strings_and_comments(source: str) -> str:
return _strip_comments(_strip_strings(source))
def _open_bracket(ch: str) -> bool:
return ch in "({["
def _close_bracket(ch: str) -> bool:
return ch in ")}]"
def _count_top_level_forms(text: str, start: int) -> int:
depth = 1
forms = 0
in_form = False
i = start
n = len(text)
while i < n and depth != 0:
ch = text[i]
if _open_bracket(ch):
if depth == 1 and not in_form:
forms += 1
depth += 1
in_form = True
elif _close_bracket(ch):
if depth == 1:
in_form = False
depth -= 1
elif ch.isspace():
in_form = depth != 1
else:
if depth == 1 and not in_form:
forms += 1
in_form = True
i += 1
return forms
def _skip_to_body(text: str, match_start: int) -> int:
i = match_start + 1
while i < len(text) and not text[i].isspace() and text[i] != ")":
i += 1
return i
def _pairwise_clause_count(form_type: str, remaining: int) -> int:
base = remaining // 2
if form_type == "case" and remaining % 2 == 1:
return base + 1
return base
def _count_clauses(text: str, form_type: str, match_start: int) -> int:
body_start = _skip_to_body(text, match_start)
total_forms = _count_top_level_forms(text, body_start)
remaining = total_forms - _FORM_SKIP.get(form_type, 0)
if form_type in _THREAD_FORMS:
return remaining
return _pairwise_clause_count(form_type, remaining)
def _count_cond_decisions(clean: str) -> int:
total = 0
for match in _COND.finditer(clean):
form_type = _COND.match(match.group(0)).group(1)
total += _count_clauses(clean, form_type, match.start())
return total
def cyclomatic_complexity(fn_text: str) -> int:
clean = without_strings_and_comments(fn_text)
simple = len(_DECISION.findall(clean))
return 1 + simple + _count_cond_decisions(clean)
def _char_literal_end(source: str, index: int) -> int:
first = index + 1
if first >= len(source):
return len(source)
i = first + 1
while i < len(source):
ch = source[i]
if ch.isspace() or ch in _CHAR_DELIMITERS:
return i
i += 1
return len(source)
def _consume_comment(source: str, index: int, line: int) -> tuple[int, int]:
index += 1
while index < len(source) and source[index] != "\n":
index += 1
if index < len(source):
return index + 1, line + 1
return index, line
def _consume_string(source: str, index: int, line: int) -> tuple[int, int]:
index += 1
escaped = False
while index < len(source):
ch = source[index]
if escaped:
escaped = False
elif ch == "\\":
escaped = True
elif ch == '"':
return index + 1, line
if ch == "\n":
line += 1
index += 1
return index, line
def _remember_form(
forms: list[dict], source: str, start: int, start_line: int, end: int, line: int
) -> None:
form_text = source[start : end + 1]
matched = _DEFN.match(form_text)
if matched is None:
return
forms.append(
{
"name": matched.group(1),
"start_line": start_line,
"end_line": line,
"text": form_text,
}
)
def _close_form(forms, source, depth, form_start, form_line, index, line):
if depth == 1 and form_start is not None and form_line is not None:
_remember_form(forms, source, form_start, form_line, index, line)
return 0, None, None
return max(0, depth - 1), form_start, form_line
def _extract_top_level_defns(source: str) -> list[dict]:
forms: list[dict] = []
index = 0
line = 1
depth = 0
form_start = None
form_line = None
while index < len(source):
ch = source[index]
if ch == ";":
index, line = _consume_comment(source, index, line)
continue
if ch == '"':
index, line = _consume_string(source, index, line)
continue
if ch == "\\":
index = _char_literal_end(source, index)
continue
if ch == "(":
if depth == 0:
form_start = index
form_line = line
depth += 1
elif ch == ")":
depth, form_start, form_line = _close_form(
forms, source, depth, form_start, form_line, index, line
)
if ch == "\n":
line += 1
index += 1
return forms
def extract_functions(source: str) -> list[dict]:
found = []
for form in _extract_top_level_defns(source):
found.append(
{
"name": form["name"],
"start_line": form["start_line"],
"end_line": form["end_line"],
"complexity": cyclomatic_complexity(form["text"]),
}
)
return found
def declared_namespace(source: str) -> str | None:
for pattern in (_NS, _IN_NS, _IN_NS_QUOTE):
match = pattern.search(source)
if match:
return match.group(1)
return None
def namespace_from_path(path: str, source_root: str | None) -> str:
relative = path.replace("\\", "/")
root = (source_root or "").replace("\\", "/").rstrip("/")
if root and root not in {".", ""} and relative.startswith(root + "/"):
relative = relative[len(root) + 1 :]
if relative.startswith("src/"):
relative = relative[4:]
elif "/src/" in relative:
relative = relative.split("/src/", 1)[1]
relative = re.sub(r"\.(?:clj[cs]?|bb)$", "", relative)
return relative.replace("/", ".").replace("_", "-")
def functions_in_source(
source: str, path: str, source_root: str | None = None
) -> list[Function]:
namespace = declared_namespace(source) or namespace_from_path(path, source_root)
return [
Function(
name=item["name"],
namespace=namespace,
complexity=item["complexity"],
start_line=item["start_line"],
end_line=item["end_line"],
path=path,
language="clojure",
)
for item in extract_functions(source)
]
+151
View File
@@ -0,0 +1,151 @@
"""Go functions and cyclomatic complexity, following crap4go.
Decision points are `if`, `for`, `range`, each `switch` / type-switch case,
each `select` clause, and `&&` / `||`. Coverage later uses Go's statement
profile. The namespace is the package import path; methods are namespaced by
their receiver type (`import/path.Widget`) and named `Run`, which is the join
uml-viewer makes between a class and its operations.
"""
from pathlib import Path
from crapper.languages.treesitter import (
binary_logic,
child_of_type,
complexity,
descendants,
end_line,
node_text,
parse,
start_line,
)
from crapper.model import Function
_DECISIONS = {
"if_statement",
"for_statement",
"expression_case",
"type_case",
"communication_case",
"default_case",
}
def _is_decision(node) -> bool:
return node.type in _DECISIONS or binary_logic(node)
def _skip(_node) -> bool:
return False
def _package_clause(data: bytes, root) -> str:
for node in descendants(root):
if node.type == "package_identifier":
return node_text(data, node)
return ""
def _type_identifiers(node) -> list[str]:
found = []
for item in descendants(node):
if item.type == "type_identifier":
found.append(item)
return found
def _receiver_type(data: bytes, method) -> str | None:
params = child_of_type(method, "parameter_list")
if params is None:
return None
idents = _type_identifiers(params)
if not idents:
return None
return node_text(data, idents[0])
def _method_name(data: bytes, method) -> str | None:
ident = child_of_type(method, "field_identifier")
if ident is None:
return None
return node_text(data, ident)
def _function_name(data: bytes, fn) -> str | None:
ident = child_of_type(fn, "identifier")
if ident is None:
return None
return node_text(data, ident)
def package_namespace(path: str, package_name: str) -> str:
"""Import path of the package, or the package clause when no go.mod exists."""
current = Path(path).resolve().parent
module = None
module_dir = None
while True:
candidate = current / "go.mod"
if candidate.is_file():
for line in candidate.read_text(encoding="utf-8").splitlines():
stripped = line.strip()
if stripped.startswith("module "):
module = stripped.split(None, 1)[1].strip()
module_dir = current
break
break
if current.parent == current:
break
current = current.parent
if not module or module_dir is None:
return package_name
relative = Path(path).resolve().parent.relative_to(module_dir).as_posix()
if relative == ".":
return module
return f"{module}/{relative}"
def _absolute(path: str, project_root: str | None) -> str:
file_path = Path(path)
if file_path.is_absolute():
return str(file_path)
if project_root:
return str((Path(project_root) / file_path).resolve())
return str(file_path.resolve())
def functions_in_source(
source: str, path: str, project_root: str | None = None
) -> list[Function]:
data, tree = parse(source, "go")
package_name = _package_clause(data, tree.root_node)
package_ns = package_namespace(_absolute(path, project_root), package_name)
found: list[Function] = []
for node in descendants(tree.root_node):
if node.type not in {"function_declaration", "method_declaration"}:
continue
if child_of_type(node, "block") is None:
continue
if node.type == "method_declaration":
name = _method_name(data, node)
receiver = _receiver_type(data, node)
if not name or not receiver:
continue
namespace = f"{package_ns}.{receiver}"
else:
name = _function_name(data, node)
if not name:
continue
namespace = package_ns
found.append(
Function(
name=name,
namespace=namespace,
complexity=complexity(node, _is_decision, _skip),
start_line=start_line(node),
end_line=end_line(node),
path=path,
language="go",
)
)
return found
+116
View File
@@ -0,0 +1,116 @@
"""Java methods and cyclomatic complexity, following crap4java.
Constructors, abstract methods, and methods inside anonymous or local classes
are ignored. Nested member types become their own namespace (`pkg.Outer.Inner`)
so uml-viewer can attach scores to that class. JaCoCo's binary name
(`pkg.Outer$Inner`) is kept for coverage lookup.
"""
import re
from crapper.languages.treesitter import (
binary_logic,
child_of_type,
complexity,
end_line,
node_text,
parse,
start_line,
)
from crapper.model import Function
_PACKAGE = re.compile(r"(?m)^\s*package\s+([a-zA-Z_][\w.]*)\s*;")
_NESTED_TYPES = {
"class_declaration",
"interface_declaration",
"enum_declaration",
"record_declaration",
"annotation_type_declaration",
}
_DECISIONS = {
"if_statement",
"for_statement",
"enhanced_for_statement",
"while_statement",
"do_statement",
"catch_clause",
"ternary_expression",
"switch_label",
}
def _is_decision(node) -> bool:
return node.type in _DECISIONS or binary_logic(node)
def _skip(node) -> bool:
return node.type in _NESTED_TYPES or node.type == "class_body"
def _inside_executable(node) -> bool:
current = node.parent
while current is not None:
if current.type in {
"method_declaration",
"constructor_declaration",
"lambda_expression",
}:
return True
current = current.parent
return False
def _package_name(source: str) -> str | None:
match = _PACKAGE.search(source)
return match.group(1) if match else None
def _type_chain(data: bytes, node) -> list[str]:
names: list[str] = []
current = node.parent
while current is not None:
if current.type in _NESTED_TYPES:
ident = child_of_type(current, "identifier")
if ident is not None:
names.append(node_text(data, ident))
current = current.parent
names.reverse()
return names
def _qualify(package: str | None, names: list[str], nested: str) -> str:
body = nested.join(names)
if package:
return f"{package}.{body}" if body else package
return body
def functions_in_source(source: str, path: str) -> list[Function]:
data, tree = parse(source, "java")
package = _package_name(source)
found: list[Function] = []
def visit(node) -> None:
if node.type == "method_declaration" and not _inside_executable(node):
if child_of_type(node, "block") is not None:
ident = child_of_type(node, "identifier")
names = _type_chain(data, node)
if ident is not None and names:
found.append(
Function(
name=node_text(data, ident),
namespace=_qualify(package, names, "."),
complexity=complexity(node, _is_decision, _skip),
start_line=start_line(node),
end_line=end_line(node),
path=path,
language="java",
jacoco_class=_qualify(package, names, "$"),
)
)
return
for child in node.children:
visit(child)
visit(tree.root_node)
return found
+133
View File
@@ -0,0 +1,133 @@
"""Python functions, methods, and cyclomatic complexity.
Decision points follow the same structural rule as the other languages:
`if`, `elif`, `for`, `while`, `except`, each `match` case, a comprehension
filter, a conditional expression, and each `and` / `or`. Module-level
functions and class methods are entries. A nested function stays inside the
enclosing function. A class method's namespace is `module.Class`.
"""
from crapper.languages.treesitter import (
child_of_type,
complexity,
descendants,
end_line,
node_text,
parse,
start_line,
)
from crapper.model import Function
_DECISIONS = {
"if_statement",
"elif_clause",
"for_statement",
"while_statement",
"except_clause",
"conditional_expression",
"case_clause",
"if_clause",
}
def _is_decision(node) -> bool:
if node.type in _DECISIONS:
return True
return node.type == "boolean_operator" and any(
child.type in {"and", "or"} for child in node.children
)
def _skip(node) -> bool:
return node.type == "class_definition"
def _strip_root(relative: str, source_root: str | None) -> str:
root = (source_root or "").replace("\\", "/").rstrip("/")
if root and root != "." and relative.startswith(root + "/"):
return relative[len(root) + 1 :]
if relative.startswith("./"):
return relative[2:]
return relative
def _strip_src(relative: str) -> str:
if relative.startswith("src/"):
return relative[4:]
if "/src/" in relative:
return relative.split("/src/", 1)[1]
return relative
def _strip_module_suffix(relative: str) -> str:
if relative.endswith(".py"):
relative = relative[: -len(".py")]
if relative.endswith("/__init__"):
return relative[: -len("/__init__")]
if relative == "__init__":
return ""
return relative
def _module_namespace(path: str, source_root: str | None) -> str:
relative = _strip_module_suffix(_strip_src(_strip_root(path.replace("\\", "/"), source_root)))
if not relative:
return "__init__"
return relative.replace("/", ".")
def _is_nested_function(node) -> bool:
"""A function inside another function, unless it is a method of a class."""
current = node.parent
while current is not None:
if current.type == "class_definition":
return False
if current.type == "function_definition":
return True
current = current.parent
return False
def _class_names(data: bytes, node) -> list[str]:
names: list[str] = []
current = node.parent
while current is not None:
if current.type == "class_definition":
ident = child_of_type(current, "identifier")
if ident is not None:
names.append(node_text(data, ident))
current = current.parent
names.reverse()
return names
def functions_in_source(
source: str, path: str, source_root: str | None = None
) -> list[Function]:
data, tree = parse(source, "python")
module = _module_namespace(path, source_root)
found: list[Function] = []
for node in descendants(tree.root_node):
if node.type != "function_definition" or _is_nested_function(node):
continue
ident = child_of_type(node, "identifier")
if ident is None:
continue
classes = _class_names(data, node)
if classes:
namespace = ".".join([module, *classes]) if module else ".".join(classes)
else:
namespace = module
found.append(
Function(
name=node_text(data, ident),
namespace=namespace,
complexity=complexity(node, _is_decision, _skip),
start_line=start_line(node),
end_line=end_line(node),
path=path,
language="python",
)
)
return found
+188
View File
@@ -0,0 +1,188 @@
"""Rust functions and cyclomatic complexity.
Decision points are `if` / `if let`, `for`, `while` / `while let`, `loop`,
each `match` arm, `?`, and `&&` / `||`. Free functions are namespaced by the
module path (`crate::foo::bar`). Methods are namespaced by the self type
(`crate::foo::Widget`) so uml-viewer can join them to that type. Bodies inside
`mod tests` are test code and are not scored.
"""
import re
from pathlib import Path
from crapper.languages.treesitter import (
binary_logic,
child_of_type,
complexity,
descendants,
end_line,
node_text,
parse,
start_line,
)
from crapper.model import Function
_DECISIONS = {
"if_expression",
"for_expression",
"while_expression",
"loop_expression",
"match_arm",
"try_expression",
}
_PACKAGE_BLOCK = re.compile(r"(?ms)^\[package\](.*?)(?:^\[|\Z)")
_NAME = re.compile(r'(?m)^name\s*=\s*"([^"]+)"')
def _is_decision(node) -> bool:
return node.type in _DECISIONS or binary_logic(node)
def _skip(_node) -> bool:
return False
def _has_block(node) -> bool:
return child_of_type(node, "block") is not None
def _ancestor_mods(data: bytes, node) -> list[str]:
names: list[str] = []
current = node.parent
while current is not None:
if current.type == "mod_item":
ident = child_of_type(current, "identifier")
if ident is not None:
names.append(node_text(data, ident))
current = current.parent
names.reverse()
return names
def _in_mod_named(data: bytes, node, name: str) -> bool:
return name in _ancestor_mods(data, node)
def _impl_type(data: bytes, impl) -> str | None:
saw_for = False
chosen = None
for child in impl.children:
if child.type == "for":
saw_for = True
chosen = None
continue
if child.type in {"type_identifier", "generic_type", "scoped_type_identifier"}:
if child.type == "type_identifier":
chosen = node_text(data, child)
else:
ident = child_of_type(child, "type_identifier")
if ident is None:
for item in descendants(child):
if item.type == "type_identifier":
ident = item
break
chosen = node_text(data, ident) if ident is not None else None
if saw_for:
return chosen
return chosen
def _crate_name(path: str) -> tuple[str, Path | None]:
current = Path(path).resolve().parent
while True:
cargo = current / "Cargo.toml"
if cargo.is_file():
text = cargo.read_text(encoding="utf-8")
block = _PACKAGE_BLOCK.search(text)
if block:
match = _NAME.search(block.group(1))
name = match.group(1) if match else current.name
return name.replace("-", "_"), current
if current.parent == current:
break
current = current.parent
return "crate", None
def _file_modules(path: str, crate_name: str, crate_root: Path | None) -> list[str]:
file_path = Path(path).resolve()
if crate_root is None:
return [crate_name, file_path.stem] if file_path.stem not in {"lib", "main"} else [crate_name]
try:
relative = file_path.relative_to(crate_root).as_posix()
except ValueError:
return [crate_name]
if relative.startswith("src/"):
relative = relative[4:]
if relative in {"lib.rs", "main.rs"}:
return [crate_name]
parts = relative.split("/")
if parts[-1] == "mod.rs":
parts = parts[:-1]
elif parts[-1].endswith(".rs"):
parts[-1] = parts[-1][:-3]
if parts[:1] == ["bin"] and len(parts) >= 2:
return parts[1:]
return [crate_name, *parts]
def _namespace(modules: list[str], extra: list[str], type_name: str | None) -> str:
parts = [*modules, *extra]
if type_name:
parts.append(type_name)
return "::".join(part for part in parts if part)
def _absolute(path: str, project_root: str | None) -> str:
file_path = Path(path)
if file_path.is_absolute():
return str(file_path)
if project_root:
return str((Path(project_root) / file_path).resolve())
return str(file_path.resolve())
def _impl_owner(data: bytes, node):
parent = node.parent
if parent is None or parent.type != "declaration_list":
return None
owner = parent.parent
if owner is None or owner.type != "impl_item":
return None
return _impl_type(data, owner)
def _record_function(data: bytes, node, modules: list[str], path: str) -> Function | None:
if node.type != "function_item" or not _has_block(node):
return None
if node.parent is not None and node.parent.type == "block":
return None
if _in_mod_named(data, node, "tests"):
return None
ident = child_of_type(node, "identifier")
if ident is None:
return None
return Function(
name=node_text(data, ident),
namespace=_namespace(modules, _ancestor_mods(data, node), _impl_owner(data, node)),
complexity=complexity(node, _is_decision, _skip),
start_line=start_line(node),
end_line=end_line(node),
path=path,
language="rust",
)
def functions_in_source(
source: str, path: str, project_root: str | None = None
) -> list[Function]:
data, tree = parse(source, "rust")
located = _absolute(path, project_root)
crate_name, crate_root = _crate_name(located)
modules = _file_modules(located, crate_name, crate_root)
found: list[Function] = []
for node in descendants(tree.root_node):
recorded = _record_function(data, node, modules, path)
if recorded is not None:
found.append(recorded)
return found
+67
View File
@@ -0,0 +1,67 @@
"""Shared tree-sitter helpers for Java, Go, TypeScript, Rust, and Python."""
from functools import lru_cache
@lru_cache(maxsize=None)
def parser_for(language: str):
from tree_sitter_language_pack import download, get_parser
download([language])
return get_parser(language)
def parse(source: str, language: str):
data = source.encode("utf-8")
tree = parser_for(language).parse(data)
return data, tree
def node_text(data: bytes, node) -> str:
return data[node.start_byte : node.end_byte].decode("utf-8")
def start_line(node) -> int:
return node.start_point[0] + 1
def end_line(node) -> int:
row, col = node.end_point
if col == 0:
return max(start_line(node), row)
return row + 1
def child_of_type(node, *types: str):
for child in node.children:
if child.type in types:
return child
return None
def descendants(node):
stack = [node]
while stack:
current = stack.pop()
yield current
stack.extend(reversed(current.children))
def complexity(node, is_decision, skip) -> int:
"""McCabe complexity of `node`: 1 plus each decision in the subtree."""
def walk(current) -> int:
if skip(current):
return 0
score = 1 if is_decision(current) else 0
for child in current.children:
score += walk(child)
return score
return 1 + walk(node)
def binary_logic(node) -> bool:
return node.type == "binary_expression" and any(
child.type in {"&&", "||"} for child in node.children
)
+180
View File
@@ -0,0 +1,180 @@
"""TypeScript and TSX functions, methods, and cyclomatic complexity.
Decision points follow the same structural rule as crap4java: `if`, loops,
`catch`, `?:`, each `switch` case (including `default`), and `&&` / `||`.
Top-level functions and class methods are entries. Callbacks stay inside the
enclosing function. A class method's namespace is `module.Class`, which is
the uml-viewer class key.
"""
from crapper.languages.treesitter import (
binary_logic,
child_of_type,
complexity,
descendants,
end_line,
node_text,
parse,
start_line,
)
from crapper.model import Function
_DECISIONS = {
"if_statement",
"for_statement",
"for_in_statement",
"while_statement",
"do_statement",
"catch_clause",
"ternary_expression",
"switch_case",
"switch_default",
}
_SKIP = {
"class_declaration",
"abstract_class_declaration",
"interface_declaration",
}
_FUNCTION_TYPES = {
"function_declaration",
"method_definition",
"arrow_function",
"function_expression",
}
def _is_decision(node) -> bool:
return node.type in _DECISIONS or binary_logic(node)
def _skip(node) -> bool:
return node.type in _SKIP
def _grammar(path: str) -> str:
if path.endswith(".tsx"):
return "tsx"
return "typescript"
def _module_namespace(path: str, source_root: str | None) -> str:
relative = path.replace("\\", "/")
root = (source_root or "").replace("\\", "/").rstrip("/")
if root and root != "." and relative.startswith(root + "/"):
relative = relative[len(root) + 1 :]
elif relative.startswith("./"):
relative = relative[2:]
if relative.startswith("src/"):
relative = relative[4:]
elif "/src/" in relative:
relative = relative.split("/src/", 1)[1]
for suffix in (".tsx", ".mts", ".cts", ".ts"):
if relative.endswith(suffix):
relative = relative[: -len(suffix)]
break
return relative.replace("/", ".")
def _inside_function(node) -> bool:
current = node.parent
while current is not None:
if current.type in _FUNCTION_TYPES:
return True
current = current.parent
return False
def _class_names(data: bytes, node) -> list[str]:
names: list[str] = []
current = node.parent
while current is not None:
if current.type in {"class_declaration", "abstract_class_declaration"}:
ident = child_of_type(current, "type_identifier", "identifier")
if ident is not None:
names.append(node_text(data, ident))
current = current.parent
names.reverse()
return names
def _has_body(node) -> bool:
return child_of_type(node, "statement_block") is not None
def _append(found: list[Function], **kwargs) -> None:
found.append(Function(**kwargs))
def _append_function(found, data, node, module, path) -> None:
if _inside_function(node) or not _has_body(node):
return
ident = child_of_type(node, "identifier")
if ident is None:
return
_append(
found,
name=node_text(data, ident),
namespace=module,
complexity=complexity(node, _is_decision, _skip),
start_line=start_line(node),
end_line=end_line(node),
path=path,
language="typescript",
)
def _append_method(found, data, node, module, path) -> None:
if not _has_body(node):
return
ident = child_of_type(node, "property_identifier", "identifier")
classes = _class_names(data, node)
if ident is None or not classes:
return
_append(
found,
name=node_text(data, ident),
namespace=f"{module}.{'.'.join(classes)}",
complexity=complexity(node, _is_decision, _skip),
start_line=start_line(node),
end_line=end_line(node),
path=path,
language="typescript",
)
def _append_arrows(found, data, node, module, path) -> None:
if _inside_function(node):
return
for declarator in node.children:
if declarator.type != "variable_declarator":
continue
ident = child_of_type(declarator, "identifier")
value = child_of_type(declarator, "arrow_function", "function_expression")
if ident is None or value is None:
continue
_append(
found,
name=node_text(data, ident),
namespace=module,
complexity=complexity(value, _is_decision, _skip),
start_line=start_line(declarator),
end_line=end_line(declarator),
path=path,
language="typescript",
)
def functions_in_source(
source: str, path: str, source_root: str | None = None
) -> list[Function]:
data, tree = parse(source, _grammar(path))
module = _module_namespace(path, source_root)
found: list[Function] = []
for node in descendants(tree.root_node):
if node.type == "function_declaration":
_append_function(found, data, node, module, path)
elif node.type == "method_definition":
_append_method(found, data, node, module, path)
elif node.type == "lexical_declaration":
_append_arrows(found, data, node, module, path)
return found
+58
View File
@@ -0,0 +1,58 @@
"""Write the crap4clj snapshot uml-viewer reads."""
from pathlib import Path
from crapper.model import Entry
def metrics_path(root: Path) -> Path:
return root / ".metrics" / "crap.edn"
def _edn_string(value: str) -> str:
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
return f'"{escaped}"'
def _edn_number(value: float) -> str:
text = f"{value:.4f}".rstrip("0").rstrip(".")
if "." not in text:
text += ".0"
return text
def _edn_value(value: float | None) -> str:
if value is None:
return "nil"
return _edn_number(value)
def render_edn(entries: list[Entry]) -> str:
"""EDN map `{:entries [...]}` with crap4clj's keys, in crap4clj's order.
uml-viewer groups this file by `:namespace` and joins each operation on
`:name`. Coverage is a percentage. `nil` coverage and CRAP mean N/A.
"""
rows = []
for entry in entries:
rows.append(
" {"
f":name {_edn_string(entry.name)}, "
f":namespace {_edn_string(entry.namespace)}, "
f":complexity {entry.complexity}, "
f":coverage {_edn_value(entry.coverage)}, "
f":crap {_edn_value(entry.crap)}"
"}"
)
body = "\n".join(rows)
if body:
return "{:entries [\n" + body + "\n]}\n"
return "{:entries []}\n"
def write_metrics(entries: list[Entry], root: Path) -> Path:
path = metrics_path(root)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(render_edn(entries), encoding="utf-8")
return path
+28
View File
@@ -0,0 +1,28 @@
from dataclasses import dataclass
@dataclass(frozen=True)
class Function:
"""One scored unit: a function or method."""
name: str
namespace: str
complexity: int
start_line: int
end_line: int
path: str
language: str
jacoco_class: str | None = None
@dataclass(frozen=True)
class Entry:
"""One row of the CRAP report and of `.metrics/crap.edn`."""
name: str
namespace: str
complexity: int
coverage: float | None
crap: float | None
path: str
language: str
+42
View File
@@ -0,0 +1,42 @@
from crapper.model import Entry
def _format_coverage(coverage: float | None) -> str:
if coverage is None:
return " N/A "
return f"{coverage:5.1f}%"
def _format_crap(score: float | None) -> str:
if score is None:
return " N/A"
return f"{score:8.1f}"
def format_report(entries: list[Entry]) -> str:
name_width = 30
namespace_width = 35
if entries:
name_width = max(name_width, min(48, max(len(entry.name) for entry in entries)))
namespace_width = max(
namespace_width, min(72, max(len(entry.namespace) for entry in entries))
)
header = (
f"{'Function':<{name_width}} {'Namespace':<{namespace_width}} "
f"{'CC':>4} {'Cov%':>7} {'CRAP':>8}"
)
separator = "-" * len(header)
languages = ", ".join(sorted({entry.language for entry in entries}))
lines = ["CRAP Report", "==========="]
if languages:
lines.append(f"Languages: {languages}")
lines.append("")
lines.extend([header, separator])
for entry in entries:
lines.append(
f"{entry.name:<{name_width}} {entry.namespace:<{namespace_width}} "
f"{entry.complexity:4d} {_format_coverage(entry.coverage)} "
f"{_format_crap(entry.crap)}"
)
lines.append("")
return "\n".join(lines)
+501
View File
@@ -0,0 +1,501 @@
"""Run each language's coverage tool, then leave the reports for the loader.
Failures are reported and do not stop analysis. A language with no coverage
tool, or a failed run, keeps N/A coverage for its functions.
"""
from __future__ import annotations
import json
import shlex
import shutil
import subprocess
import sys
from pathlib import Path
from crapper.discover import is_test_file, language_of
_MAVEN = (
"mvn -q "
"org.jacoco:jacoco-maven-plugin:0.8.12:prepare-agent "
"test "
"org.jacoco:jacoco-maven-plugin:0.8.12:report"
)
def _warn(message: str) -> None:
print(message, file=sys.stderr)
def run_shell(command: str, cwd: Path) -> int:
_warn(f"+ ({cwd}) {command}")
try:
completed = subprocess.run(command, cwd=cwd, shell=True)
except OSError as exc:
_warn(f"Coverage command failed to start: {exc}")
return 127
return completed.returncode
def _nearest(start: Path, marker: str, stop: Path) -> Path | None:
current = start if start.is_dir() else start.parent
stop = stop.resolve()
while True:
if (current / marker).is_file():
return current
if current == stop or current.parent == current:
return None
current = current.parent
def _clean_dir(path: Path) -> None:
if path.exists():
shutil.rmtree(path)
def _clean_clojure(root: Path) -> None:
coverage = root / "target" / "coverage"
if not coverage.is_dir():
return
keep = {"typescript", "rust", "go", "python"}
for path in coverage.iterdir():
if path.name in keep:
continue
if path.is_dir():
shutil.rmtree(path)
else:
path.unlink()
_VITEST_CONFIGS = (
"vitest.config.ts",
"vitest.config.mts",
"vitest.config.cts",
"vitest.config.js",
"vitest.config.mjs",
"vitest.config.cjs",
)
def _package_json(package: Path) -> dict | None:
path = package / "package.json"
if not path.is_file():
return None
try:
data = json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError:
return None
return data if isinstance(data, dict) else None
def _uses_vitest(package: Path, scripts: dict) -> bool:
if "vitest" in str(scripts.get("test", "")):
return True
return any((package / name).is_file() for name in _VITEST_CONFIGS)
def _vitest_version(package: Path) -> str | None:
installed = package / "node_modules" / "vitest" / "package.json"
if installed.is_file():
try:
data = json.loads(installed.read_text(encoding="utf-8"))
except json.JSONDecodeError:
data = None
if isinstance(data, dict) and isinstance(data.get("version"), str):
return data["version"]
return None
def _manifest_snapshot(package: Path) -> dict[Path, bytes | None]:
snapshot: dict[Path, bytes | None] = {}
for name in ("package.json", "package-lock.json", "npm-shrinkwrap.json"):
path = package / name
snapshot[path] = path.read_bytes() if path.is_file() else None
return snapshot
def _restore_manifests(snapshot: dict[Path, bytes | None]) -> None:
for path, content in snapshot.items():
if content is None:
if path.is_file():
path.unlink()
elif not path.is_file() or path.read_bytes() != content:
path.write_bytes(content)
def _ensure_vitest_coverage(package: Path) -> bool:
"""Vitest collects coverage itself. c8 never sees its worker processes."""
if (package / "node_modules" / "@vitest" / "coverage-v8").is_dir():
return True
version = _vitest_version(package)
spec = f"@vitest/coverage-v8@{version}" if version else "@vitest/coverage-v8"
_warn(
f"Installing {spec} so Vitest can write LCOV. "
"package.json and the lockfile are restored afterward."
)
snapshot = _manifest_snapshot(package)
code = run_shell(f"npm install --no-save --no-package-lock {spec}", package)
_restore_manifests(snapshot)
ready = code == 0 and (package / "node_modules" / "@vitest" / "coverage-v8").is_dir()
if not ready:
_warn("Vitest's coverage provider is missing. TypeScript coverage will be N/A.")
return ready
def typescript_packages(root: Path, files: list[Path]) -> list[Path]:
packages: set[Path] = set()
root = root.resolve()
for file in files:
if language_of(file) != "typescript":
continue
package = _nearest(file, "package.json", root)
if package is not None:
packages.add(package.resolve())
return sorted(packages)
def typescript_command(package: Path, files: list[Path], report_dir: Path) -> str | None:
"""Shell command that writes LCOV into `report_dir`, or None when there is no test script."""
data = _package_json(package)
if data is None:
return None
scripts = data.get("scripts") or {}
if not isinstance(scripts, dict):
scripts = {}
if "coverage" in scripts:
return "npm run coverage"
quoted_dir = shlex.quote(str(report_dir))
if _uses_vitest(package, scripts):
return _vitest_coverage_command(package, files, quoted_dir)
if "test" not in scripts:
return None
return f"npx --yes c8 --reporter=lcov --reports-dir {quoted_dir} npm test"
def _vitest_includes(package: Path, files: list[Path]) -> list[str]:
includes: list[str] = []
package = package.resolve()
for file in files:
if language_of(file) != "typescript" or is_test_file(file):
continue
try:
relative = Path(file).resolve().relative_to(package)
except ValueError:
continue
includes.append(shlex.quote(relative.as_posix()))
return includes
def _vitest_coverage_command(package: Path, files: list[Path], quoted_dir: str) -> str:
binary = package / "node_modules" / ".bin" / "vitest"
runner = shlex.quote(str(binary)) if binary.is_file() else "npx vitest"
flags = [
"--coverage",
"--coverage.reporter=lcov",
f"--coverage.reportsDirectory={quoted_dir}",
"--coverage.reportOnFailure=true",
]
for include in _vitest_includes(package, files):
flags.append(f"--coverage.include={include}")
return f"{runner} run " + " ".join(flags)
def _coverage_report(root: Path, module: Path, language: str) -> Path:
root = root.resolve()
module = module.resolve()
if module == root:
return root / "target" / "coverage" / language / "lcov.info"
slug = module.relative_to(root).as_posix().replace("/", "__")
return root / "target" / "coverage" / language / slug / "lcov.info"
def rust_modules(root: Path, files: list[Path]) -> list[Path]:
modules: set[Path] = set()
root = root.resolve()
for file in files:
if Path(file).suffix != ".rs":
continue
module = _nearest(file, "Cargo.toml", root)
if module is not None:
modules.add(module.resolve())
return sorted(modules)
def _rust_kind() -> str | None:
if shutil.which("cargo-llvm-cov"):
return "llvm-cov"
if shutil.which("cargo-tarpaulin"):
return "tarpaulin"
_warn("Neither cargo-llvm-cov nor cargo-tarpaulin is installed. Installing cargo-llvm-cov.")
if shutil.which("rustup"):
code = run_shell("rustup component add llvm-tools-preview", Path.home())
if code != 0:
_warn("rustup component add llvm-tools-preview failed.")
code = run_shell("cargo install cargo-llvm-cov --locked", Path.home())
if code == 0 and shutil.which("cargo-llvm-cov"):
return "llvm-cov"
_warn("Rust coverage will be N/A.")
return None
_PYTHON_MARKERS = ("pyproject.toml", "setup.cfg", "setup.py", "pytest.ini", "tox.ini")
def python_roots(root: Path, files: list[Path]) -> list[Path]:
found: set[Path] = set()
root = root.resolve()
for file in files:
if Path(file).suffix != ".py":
continue
module = None
for marker in _PYTHON_MARKERS:
module = _nearest(Path(file), marker, root)
if module is not None:
break
found.add((module or root).resolve())
return sorted(found)
def _python_executable(package: Path) -> str:
for relative in (".venv/bin/python", "venv/bin/python"):
candidate = package / relative
if candidate.is_file():
return shlex.quote(str(candidate))
return "python3"
def uses_pytest(package: Path) -> bool:
if (package / "pytest.ini").is_file() or (package / "conftest.py").is_file():
return True
for name in ("pyproject.toml", "setup.cfg", "tox.ini"):
path = package / name
if path.is_file() and "pytest" in path.read_text(encoding="utf-8", errors="replace"):
return True
return False
def python_sources(package: Path, files: list[Path]) -> str:
"""Directories coverage.py should report even when a file was never imported."""
tops: set[str] = set()
package = package.resolve()
for file in files:
path = Path(file)
if path.suffix != ".py" or is_test_file(path):
continue
try:
relative = path.resolve().relative_to(package)
except ValueError:
continue
tops.add(relative.parts[0])
return ",".join(sorted(tops))
def python_coverage_commands(
py: str, kind: str, data_file: Path, report: Path, source: str
) -> tuple[str, str]:
data = shlex.quote(str(data_file))
out = shlex.quote(str(report))
source_flag = f" --source={shlex.quote(source)}" if source else ""
if kind == "pytest":
module = "pytest"
else:
module = "unittest discover -s ."
run = f"{py} -m coverage run --data-file={data}{source_flag} -m {module}"
lcov = f"{py} -m coverage lcov --data-file={data} -o {out}"
return run, lcov
def _ensure_python_module(py: str, package: Path, module: str) -> bool:
probe = f"{py} -c {shlex.quote('import ' + module)}"
if run_shell(probe, package) == 0:
return True
_warn(
f"Installing {module} for Python coverage. "
"The project requirements are left unchanged."
)
code = run_shell(
f"{py} -m pip install --disable-pip-version-check {module}",
package,
)
return code == 0
def rust_coverage_command(kind: str, report: Path) -> str:
if kind == "llvm-cov":
return f"cargo llvm-cov --lcov --output-path {shlex.quote(str(report))}"
return f"cargo tarpaulin --out Lcov --output-dir {shlex.quote(str(report.parent))}"
def _languages_in(files: list[Path]) -> set[str]:
found = set()
for file in files:
language = language_of(file)
if language:
found.add(language)
return found
def _modules_with(files: list[Path], suffix: str, marker: str, root: Path) -> list[Path]:
modules: set[Path] = set()
for file in files:
if Path(file).suffix != suffix:
continue
module = _nearest(file, marker, root)
if module is not None:
modules.add(module)
return sorted(modules)
def _cover_clojure(root: Path) -> None:
if not (root / "deps.edn").is_file() and not (root / "bb.edn").is_file():
_warn("No deps.edn or bb.edn; skipping Clojure coverage.")
return
_clean_clojure(root)
code = run_shell("clj -M:cov --lcov", root)
if code != 0:
_warn("clj -M:cov --lcov failed; retrying without --lcov.")
code = run_shell("clj -M:cov", root)
if code != 0:
_warn(f"Clojure coverage exited {code}. Clojure coverage will be N/A.")
def _cover_java(root: Path, files: list[Path]) -> None:
modules = _modules_with(files, ".java", "pom.xml", root)
if not modules:
_warn("No pom.xml; skipping Java coverage.")
return
for module in modules:
_clean_dir(module / "target" / "site" / "jacoco")
exec_file = module / "target" / "jacoco.exec"
if exec_file.exists():
exec_file.unlink()
code = run_shell(_MAVEN, module)
if code != 0:
_warn(f"Java coverage exited {code} in {module}. Java coverage will be N/A.")
def _cover_go(root: Path, files: list[Path]) -> None:
modules = _modules_with(files, ".go", "go.mod", root)
if not modules:
_warn("No go.mod; skipping Go coverage.")
return
for module in modules:
profile = module / "target" / "coverage" / "go" / "coverage.out"
profile.parent.mkdir(parents=True, exist_ok=True)
if profile.exists():
profile.unlink()
code = run_shell(f"go test ./... -coverprofile={profile}", module)
if code != 0:
_warn(f"Go coverage exited {code} in {module}. Go coverage will be N/A.")
def _prepare_report(report: Path) -> None:
if report.parent.exists():
shutil.rmtree(report.parent)
report.parent.mkdir(parents=True, exist_ok=True)
def _cover_typescript(root: Path, files: list[Path]) -> None:
packages = typescript_packages(root, files)
if not packages:
_warn("No package.json; skipping TypeScript coverage.")
return
for package in packages:
report = _coverage_report(root, package, "typescript")
command = typescript_command(package, files, report.parent)
if command is None:
_warn(f"No package.json test script in {package}; skipping TypeScript coverage.")
continue
if _needs_vitest_provider(command) and not _ensure_vitest_coverage(package):
continue
_prepare_report(report)
code = run_shell(command, package)
if code != 0:
_warn(f"TypeScript coverage exited {code} in {package}. TypeScript coverage will be N/A.")
def _needs_vitest_provider(command: str) -> bool:
return command != "npm run coverage" and "vitest" in command
def _python_kind(py: str, package: Path) -> str:
if not uses_pytest(package):
return "unittest"
if _ensure_python_module(py, package, "pytest"):
return "pytest"
_warn(f"pytest is missing in {package}. Falling back to unittest.")
return "unittest"
def _record_python_lcov(package: Path, code: int, data_file: Path, lcov_cmd: str) -> None:
if not data_file.exists():
if code != 0:
_warn(f"Python coverage exited {code} in {package}. Python coverage will be N/A.")
return
lcov_code = run_shell(lcov_cmd, package)
if lcov_code != 0:
_warn(f"coverage lcov exited {lcov_code} in {package}. Python coverage will be N/A.")
elif code != 0:
_warn(f"Python tests exited {code} in {package}. Coverage was still recorded.")
def _cover_python(root: Path, files: list[Path]) -> None:
for package in python_roots(root, files):
report = _coverage_report(root, package, "python")
py = _python_executable(package)
if not _ensure_python_module(py, package, "coverage"):
_warn(f"coverage is missing in {package}. Python coverage will be N/A.")
continue
report.parent.mkdir(parents=True, exist_ok=True)
data_file = report.parent / ".coverage"
report.unlink(missing_ok=True)
data_file.unlink(missing_ok=True)
run_cmd, lcov_cmd = python_coverage_commands(
py, _python_kind(py, package), data_file, report, python_sources(package, files)
)
_record_python_lcov(package, run_shell(run_cmd, package), data_file, lcov_cmd)
def _cover_rust(root: Path, files: list[Path]) -> None:
modules = rust_modules(root, files)
if not modules:
_warn("No Cargo.toml; skipping Rust coverage.")
return
kind = _rust_kind()
if kind is None:
return
for module in modules:
report = _coverage_report(root, module, "rust")
report.parent.mkdir(parents=True, exist_ok=True)
report.unlink(missing_ok=True)
code = run_shell(rust_coverage_command(kind, report), module)
if code != 0:
_warn(f"Rust coverage exited {code} in {module}. Rust coverage will be N/A.")
def run_coverage(root: Path, files: list[Path], command: str | None) -> None:
"""Generate coverage reports for the languages present in `files`."""
root = root.resolve()
if command:
code = run_shell(command, root)
if code != 0:
_warn(f"Coverage command exited {code}. Coverage may be N/A.")
return
languages = _languages_in(files)
if "clojure" in languages:
_cover_clojure(root)
if "java" in languages:
_cover_java(root, files)
if "go" in languages:
_cover_go(root, files)
if "typescript" in languages:
_cover_typescript(root, files)
if "python" in languages:
_cover_python(root, files)
if "rust" in languages:
_cover_rust(root, files)
+167
View File
@@ -0,0 +1,167 @@
import subprocess
from pathlib import Path
from crapper.cli import run
from crapper.discover import language_of
def test_language_detection():
assert language_of("src/app/core.clj") == "clojure"
assert language_of("src/app/core.cljc") == "clojure"
assert language_of("Widget.java") == "java"
assert language_of("board.go") == "go"
assert language_of("ui/view.tsx") == "typescript"
assert language_of("src/lib.rs") == "rust"
assert language_of("src/crapper/cli.py") == "python"
assert language_of("types.d.ts") is None
assert language_of("notes.md") is None
def _write(root: Path, relative: str, source: str) -> None:
path = root / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(source, encoding="utf-8")
def test_project_snapshot_is_readable_as_crap4clj_edn(tmp_path):
_write(
tmp_path,
"src/demo/core.clj",
"(ns demo.core)\n\n(defn choose [x]\n (if x 1 0))\n",
)
_write(
tmp_path,
"src/demo/Board.java",
"package demo;\npublic class Board {\n public int place(int x) {\n if (x > 0 && ready) return 1;\n return 0;\n }\n}\n",
)
_write(
tmp_path,
"board.go",
"package demo\n\nfunc Place(x int) int {\n\tif x > 0 {\n\t\treturn x\n\t}\n\treturn 0\n}\n",
)
_write(
tmp_path,
"src/ui/view.ts",
"export function view(ok: boolean): number {\n return ok ? 1 : 0;\n}\n",
)
_write(tmp_path, "Cargo.toml", '[package]\nname = "demo"\nversion = "0.1.0"\n')
_write(
tmp_path,
"src/lib.rs",
"pub fn open(ok: bool) -> i32 { if ok { 1 } else { 0 } }\n",
)
_write(tmp_path, "src/demo/core_test.clj", "(ns demo.core-test)\n(defn should-skip [] 1)\n")
_write(tmp_path, "board_test.go", "package demo\nfunc TestPlace(t *testing.T) {}\n")
code = run(["--root", str(tmp_path), "--no-coverage"])
assert code == 0
text = (tmp_path / ".metrics" / "crap.edn").read_text(encoding="utf-8")
assert ':namespace "demo.core"' in text
assert ':name "choose"' in text
assert ':namespace "demo.Board"' in text
assert ':name "place"' in text
assert ':namespace "demo"' in text and ':name "Place"' in text
assert ':namespace "ui.view"' in text
assert ':namespace "demo"' in text and ':name "open"' in text
assert "should-skip" not in text
assert "TestPlace" not in text
reader = subprocess.run(
[
"bb",
"-e",
(
"(require '[clojure.edn :as edn])"
"(let [data (edn/read-string (slurp *in*))]"
" (doseq [entry (:entries data)]"
" (println (:namespace entry) (:name entry)"
" (:complexity entry) (pr-str (:coverage entry)))))"
),
],
input=text,
text=True,
capture_output=True,
check=False,
)
assert reader.returncode == 0, reader.stderr
assert "demo.core choose 2 nil" in reader.stdout
assert "demo.Board place 3 nil" in reader.stdout
def test_existing_coverage_feeds_the_snapshot(tmp_path):
_write(
tmp_path,
"src/demo/core.clj",
"(ns demo.core)\n\n(defn choose [x]\n (if x 1 0))\n",
)
lcov = tmp_path / "target" / "coverage" / "lcov.info"
lcov.parent.mkdir(parents=True)
lcov.write_text(
"SF:src/demo/core.clj\nDA:3,1\nDA:4,1\nend_of_record\n",
encoding="utf-8",
)
code = run(
["--root", str(tmp_path), "--use-existing-coverage", "src/demo/core.clj"]
)
assert code == 0
text = (tmp_path / ".metrics" / "crap.edn").read_text(encoding="utf-8")
assert ":coverage 100.0" in text
assert ":crap 2.0" in text
def test_threshold_exits_when_the_worst_score_is_too_high(tmp_path):
_write(tmp_path, "src/demo/core.clj", "(ns demo.core)\n\n(defn choose [x]\n (if x 1 0))\n")
lcov = tmp_path / "target" / "coverage" / "lcov.info"
lcov.parent.mkdir(parents=True)
lcov.write_text("SF:src/demo/core.clj\nDA:3,0\nDA:4,0\nend_of_record\n", encoding="utf-8")
code = run(
[
"--root",
str(tmp_path),
"--use-existing-coverage",
"--threshold",
"1",
]
)
assert code == 2
def test_help_and_unknown_option(capsys):
assert run(["--help"]) == 0
assert "uml-viewer" in capsys.readouterr().out
assert run(["--not-an-option"]) == 1
assert run(["--root"]) == 1
assert run(["--threshold", "nope"]) == 1
assert run(["--no-coverage", "--coverage-command", "true"]) == 1
def test_selects_directories_filters_and_changed_files(tmp_path, monkeypatch, capsys):
_write(tmp_path, "src/demo/core.clj", "(ns demo.core)\n\n(defn choose [x]\n (if x 1 0))\n")
_write(tmp_path, "notes.md", "not source\n")
assert run(["--root", str(tmp_path), "--no-coverage", "--source-root", "src"]) == 0
assert run(["--root", str(tmp_path), "--no-coverage", "src"]) == 0
assert run(["--root", str(tmp_path), "--no-coverage", "src/demo/core.clj", "core"]) == 0
assert run(["--root", str(tmp_path), "--no-coverage", "notes.md"]) == 0
assert "No source files" in capsys.readouterr().out
class Result:
returncode = 0
stdout = ' M src/demo/core.clj\nR old.clj -> src/demo/core.clj\n?? "src/demo/core.clj"\n'
stderr = ""
monkeypatch.setattr("crapper.cli.subprocess.run", lambda *args, **kwargs: Result())
assert run(["--root", str(tmp_path), "--no-coverage", "--changed"]) == 0
class Failed:
returncode = 1
stdout = ""
stderr = "git failed"
monkeypatch.setattr("crapper.cli.subprocess.run", lambda *args, **kwargs: Failed())
assert run(["--root", str(tmp_path), "--no-coverage", "--changed"]) == 0
def test_run_asks_for_coverage_when_it_is_enabled(tmp_path, monkeypatch):
_write(tmp_path, "src/demo/core.clj", "(ns demo.core)\n\n(defn choose [x]\n x)\n")
monkeypatch.setattr("crapper.cli.run_coverage", lambda *args: None)
assert run(["--root", str(tmp_path), "--coverage-command", "true"]) == 0
+106
View File
@@ -0,0 +1,106 @@
from crapper.languages.clojure import (
cyclomatic_complexity,
extract_functions,
functions_in_source,
namespace_from_path,
)
def test_base_and_branch_forms():
assert cyclomatic_complexity("(defn foo [])") == 1
assert cyclomatic_complexity("(defn foo [x]\n (+ x 1))") == 1
assert cyclomatic_complexity("(defn foo [x]\n (if x 1 0))") == 2
assert cyclomatic_complexity("(defn foo [x]\n (if-not x 1 0))") == 2
assert cyclomatic_complexity("(defn foo [x]\n (if-let [y x] y 0))") == 2
assert cyclomatic_complexity("(defn foo [x]\n (when x 1))") == 2
assert cyclomatic_complexity("(defn foo [x]\n (when-first [y x] y))") == 2
assert cyclomatic_complexity("(defn foo [x y]\n (and x y))") == 2
assert cyclomatic_complexity("(defn foo [x y]\n (or x y))") == 2
assert cyclomatic_complexity("(defn foo [x]\n (loop [i 0] (recur (inc i))))") == 2
def test_cond_case_and_threading():
assert (
cyclomatic_complexity("(defn foo [x]\n (cond\n (= x 1) :one\n :else :other))")
== 3
)
assert cyclomatic_complexity("(defn foo [x]\n (condp = x\n 1 :one\n 2 :two))") == 3
assert cyclomatic_complexity("(defn foo [x]\n (case x\n 1 :one\n :other))") == 3
assert (
cyclomatic_complexity(
"(defn foo [x]\n (cond-> x\n (pos? x) inc\n (even? x) (* 2)))"
)
== 3
)
assert cyclomatic_complexity("(defn foo [x]\n (some-> x inc dec))") == 3
assert (
cyclomatic_complexity("(defn foo [x]\n (some->> x (map inc) (filter pos?)))") == 3
)
def test_ignores_decisions_in_strings_and_comments():
assert cyclomatic_complexity('(defn foo [x]\n "if when cond and or"\n x)') == 1
assert cyclomatic_complexity('(defn foo [x]\n "(if true 1 0)"\n x)') == 1
assert cyclomatic_complexity('(defn foo [x]\n ;; if when cond\n x)') == 1
assert cyclomatic_complexity('(defn foo [x]\n (if x :yes :no) ; and or when cond)') == 2
def test_map_literals_inside_cond_are_not_extra_clauses():
source = """(defn foo [cell]
(let [contents (:contents cell)]
(cond
(pred-a? contents)
{:type :army :mode :awake :owner (:owner contents) :aboard true}
(pred-b? contents)
{:type :fighter :mode :awake :owner (:owner contents) :fuel 20 :from-carrier true}
(pred-c? contents) contents
(pred-d? cell)
{:type :fighter :mode :awake :owner :player :fuel 20 :from-airport true}
:else nil)))"""
assert cyclomatic_complexity(source) == 6
def test_extracts_defn_names_lines_and_namespace():
source = "(ns demo.core)\n\n(defn choose [x]\n (if x 1 0))\n\n(defn- hidden [y]\n y)\n"
found = functions_in_source(source, "src/demo/core.clj", "/proj")
assert [(item.name, item.complexity, item.start_line) for item in found] == [
("choose", 2, 3),
("hidden", 1, 6),
]
assert {item.namespace for item in found} == {"demo.core"}
def test_top_level_def_does_not_attach_to_the_previous_function():
source = (
"(defn alpha [x]\n"
" (if x 1 0))\n\n"
"(def rules\n"
" [{:pred (fn [v]\n"
" (if v (when (pos? v) true) false))}])\n\n"
"(defn omega [y]\n"
" y)"
)
found = {item["name"]: item["complexity"] for item in extract_functions(source)}
assert found == {"alpha": 2, "omega": 1}
def test_namespace_falls_back_to_the_path():
assert namespace_from_path("src/demo/core_extra.clj", None) == "demo.core-extra"
assert namespace_from_path("/proj/src/demo/core.cljc", "/proj") == "demo.core"
assert namespace_from_path("pkg/src/demo/core.bb", None) == "demo.core"
found = functions_in_source("(defn foo [] 1)\n", "src/demo/core_extra.clj", "/proj")
assert found[0].namespace == "demo.core-extra"
def test_strings_comments_and_character_literals_stay_inside_one_form():
source = '(defn foo [x]\n "say \\"hi\\"\n (if x 1)"\n \\space\n ;; (defn hidden [])\n x)\n'
assert [item["name"] for item in extract_functions(source)] == ["foo"]
def test_character_literals_do_not_swallow_the_next_defn():
source = '(def x #{\\"})\n(defn foo [] 1)'
assert [item["name"] for item in extract_functions(source)] == ["foo"]
+137
View File
@@ -0,0 +1,137 @@
from crapper.coverage import (
CoverageBundle,
go_percent,
load_bundle,
parse_form_coverage,
parse_go_profile,
parse_jacoco_index,
parse_lcov,
)
from crapper.model import Function
def function(**kwargs):
defaults = dict(
name="choose",
namespace="demo.core",
complexity=2,
start_line=3,
end_line=5,
path="src/demo/core.clj",
language="clojure",
jacoco_class=None,
)
defaults.update(kwargs)
return Function(**defaults)
def test_cloverage_form_counts_and_lcov_lines():
html = '<span class="pc" title="1 out of 2 forms covered"> 4&nbsp;&nbsp;(if x 1)</span>'
assert parse_form_coverage(html) == {4: (1, 2)}
lcov = "SF:src/demo/core.clj\nDA:3,1\nDA:4,0\nDA:9,3\nend_of_record\n"
parsed = parse_lcov(lcov)
bundle = CoverageBundle(lcov=parsed, form_html={"src/demo/core.clj": {4: (1, 2)}})
# Form HTML wins over LCOV for Clojure, matching crap4clj.
assert bundle.percent_for(function(start_line=3, end_line=5)) == 50.0
def test_lcov_percentage_when_there_is_no_html():
lcov = parse_lcov("SF:src/demo/core.clj\nDA:3,1\nDA:4,0\nend_of_record\n")
bundle = CoverageBundle(lcov=lcov)
assert bundle.percent_for(function(start_line=3, end_line=4)) == 50.0
assert bundle.percent_for(function(path="src/missing.clj")) is None
def test_go_profile_matches_crap4go_statement_ranges():
profile = parse_go_profile(
"mode: set\ngithub.com/acme/demo/board.go:4.1,6.2 2 1\ngithub.com/acme/demo/board.go:8.1,9.2 1 0\n"
)
assert go_percent(profile, "board.go", 4, 6) == 100.0
assert go_percent(profile, "board.go", 8, 9) == 0.0
assert go_percent(profile, "other.go", 1, 3) is None
def test_jacoco_instruction_coverage_joins_on_class_and_method():
xml = """<?xml version="1.0"?>
<report>
<package>
<class name="demo/pkg/Board">
<method name="place" line="4">
<counter type="INSTRUCTION" missed="1" covered="3"/>
</method>
</class>
<class name="demo/pkg/Board$Inner">
<method name="tick" line="9">
<counter type="INSTRUCTION" missed="0" covered="2"/>
</method>
</class>
</package>
</report>
"""
bundle = CoverageBundle(jacoco=parse_jacoco_index(xml))
outer = function(
name="place",
namespace="demo.pkg.Board",
language="java",
jacoco_class="demo.pkg.Board",
start_line=4,
path="src/demo/pkg/Board.java",
)
inner = function(
name="tick",
namespace="demo.pkg.Board.Inner",
language="java",
jacoco_class="demo.pkg.Board$Inner",
start_line=9,
path="src/demo/pkg/Board.java",
)
assert bundle.percent_for(outer) == 75.0
assert bundle.percent_for(inner) == 100.0
def test_jacoco_skips_methods_without_an_instruction_counter():
xml = """<report>
<class name="demo/pkg/Board">
<method name="skip" line="2">
<counter type="LINE" missed="1" covered="0"/>
</method>
<method name="bad" line="nope">
<counter type="INSTRUCTION" missed="1" covered="1"/>
</method>
<counter type="INSTRUCTION" missed="0" covered="1"/>
</class>
</report>"""
found = parse_jacoco_index(xml)
assert list(found) == ["demo.pkg.Board#bad"]
assert found["demo.pkg.Board#bad"][0].line == 0
def test_missing_go_profile_is_na():
assert go_percent(None, "board.go", 1, 3) is None
def test_load_bundle_merges_each_report(tmp_path):
lcov = tmp_path / "target" / "coverage" / "lcov.info"
lcov.parent.mkdir(parents=True)
lcov.write_text("SF:src/demo/core.clj\nDA:3,1\nDA:4,0\nend_of_record\n", encoding="utf-8")
go = tmp_path / "target" / "coverage" / "go" / "coverage.out"
go.parent.mkdir(parents=True)
go.write_text("mode: set\nboard.go:1.1,2.2 1 1\n", encoding="utf-8")
jacoco = tmp_path / "target" / "site" / "jacoco" / "jacoco.xml"
jacoco.parent.mkdir(parents=True)
jacoco.write_text(
'<report><class name="demo/Board"><method name="place" line="4">'
'<counter type="INSTRUCTION" missed="0" covered="1"/></method></class></report>',
encoding="utf-8",
)
html = tmp_path / "target" / "coverage" / "src" / "demo" / "core.clj.html"
html.parent.mkdir(parents=True)
html.write_text(
'<span title="1 out of 2 forms covered"> 4&nbsp;&nbsp;(if x 1)</span>',
encoding="utf-8",
)
(tmp_path / "target" / "coverage" / "index.html").write_text("<html></html>", encoding="utf-8")
bundle = load_bundle(tmp_path)
assert bundle.percent_for(function(start_line=3, end_line=4)) == 50.0
assert bundle.go_profile is not None
assert bundle.jacoco is not None
+73
View File
@@ -0,0 +1,73 @@
from crapper.crap import crap_score, sort_entries
from crapper.metrics import render_edn
from crapper.model import Entry
from crapper.report import format_report
def entry(**kwargs):
defaults = dict(
name="score",
namespace="demo.core",
complexity=1,
coverage=100.0,
crap=1.0,
path="src/demo/core.clj",
language="clojure",
)
defaults.update(kwargs)
return Entry(**defaults)
def test_formula_matches_crap4clj():
assert crap_score(5, 100.0) == 5.0
assert crap_score(5, 0.0) == 30.0
assert crap_score(1, 100.0) == 1.0
assert abs(crap_score(8, 45.0) - 18.648) < 0.01
assert crap_score(3, None) is None
def test_unscored_rows_sort_after_scored_rows():
rows = sort_entries(
[
entry(name="missing", crap=None, coverage=None),
entry(name="mid", crap=5.0),
entry(name="worst", crap=30.0),
]
)
assert [row.name for row in rows] == ["worst", "mid", "missing"]
def test_unscored_rows_sort_by_complexity():
rows = sort_entries(
[
entry(name="small", crap=None, coverage=None, complexity=1),
entry(name="large", crap=None, coverage=None, complexity=8),
]
)
assert [row.name for row in rows] == ["large", "small"]
def test_report_shows_na_for_missing_coverage():
text = format_report([entry(coverage=None, crap=None, complexity=3)])
assert "CRAP Report" in text
assert "N/A" in text
assert "demo.core" in text
def test_edn_uses_crap4clj_keys():
text = render_edn(
[entry(name="layout", namespace="uml-viewer.layout", complexity=8, coverage=82.0, crap=3.4)]
)
assert text.startswith("{:entries [")
assert ':name "layout"' in text
assert ':namespace "uml-viewer.layout"' in text
assert ":complexity 8" in text
assert ":coverage 82.0" in text
assert ":crap 3.4" in text
assert text.endswith("}\n")
def test_edn_writes_nil_for_missing_coverage():
text = render_edn([entry(coverage=None, crap=None)])
assert ":coverage nil" in text
assert ":crap nil" in text
+62
View File
@@ -0,0 +1,62 @@
from crapper.languages.golang import functions_in_source
BRANCHY = """package sample
func Branchy(x int, ch chan int) int {
if x > 0 && x < 10 {
return 1
}
switch x {
case 1, 2:
return 2
default:
return 3
}
}
"""
def test_counts_the_same_decisions_as_crap4go():
functions = functions_in_source(BRANCHY, "sample.go")
assert [(fn.name, fn.namespace, fn.complexity) for fn in functions] == [
("Branchy", "sample", 5)
]
def test_methods_are_namespaced_by_receiver_type():
source = """package sample
type Widget struct{}
func (w *Widget) Run() {
for i := 0; i < 10; i++ {
}
for _, item := range items {
}
}
func Simple() int { return 1 }
"""
functions = functions_in_source(source, "widget.go")
assert [(fn.namespace, fn.name, fn.complexity) for fn in functions] == [
("sample.Widget", "Run", 3),
("sample", "Simple", 1),
]
def test_import_path_comes_from_go_mod(tmp_path):
(tmp_path / "go.mod").write_text("module github.com/acme/demo\n", encoding="utf-8")
source_path = tmp_path / "internal" / "game" / "board.go"
source_path.parent.mkdir(parents=True)
source_path.write_text(
"package game\n\ntype Board struct{}\n\nfunc (b *Board) Place() {\n\tif ready && ok {\n\t\treturn\n\t}\n}\n",
encoding="utf-8",
)
functions = functions_in_source(
source_path.read_text(encoding="utf-8"),
"internal/game/board.go",
str(tmp_path),
)
assert [(fn.namespace, fn.name, fn.complexity) for fn in functions] == [
("github.com/acme/demo/internal/game.Board", "Place", 3)
]
+134
View File
@@ -0,0 +1,134 @@
from crapper.languages.java import functions_in_source
SCORE = """
class Sample {
int score(boolean a, boolean b, int[] values) {
for (int i = 0; i < values.length; i++) {
}
for (int value : values) {
}
while (a) {
a = false;
}
do {
b = false;
} while (b);
if (a && b || values.length > 0) {
}
try {
return a ? 1 : 0;
} catch (RuntimeException ex) {
return 2;
}
}
}
"""
NESTED = """
class Sample {
int nested(boolean a, boolean b, int[] values) {
for (int i = 0; i < values.length; i++) {
if (a) {
}
}
for (int value : values) {
if (b) {
}
}
while (a) {
if (b) {
}
a = false;
}
do {
if (a) {
}
b = false;
} while (b);
try {
return a ? (b ? 1 : 0) : 2;
} catch (RuntimeException ex) {
if (values.length > 0) {
return values[0];
}
return 3;
}
}
int switched(int value) {
switch (value) {
case 1:
if (value > 0) {
return 1;
}
return 0;
default:
return 2;
}
}
}
"""
ANONYMOUS = """
class Sample {
int outer() {
Runnable runnable = new Runnable() {
@Override
public void run() {
if (true) {
}
}
};
return 1;
}
}
"""
def test_counts_the_same_decisions_as_crap4java():
methods = functions_in_source(SCORE, "Sample.java")
assert [(method.name, method.complexity) for method in methods] == [("score", 10)]
methods = functions_in_source(NESTED, "Sample.java")
assert [(method.name, method.complexity) for method in methods] == [
("nested", 13),
("switched", 4),
]
def test_anonymous_class_methods_are_not_entries_and_do_not_raise_complexity():
methods = functions_in_source(ANONYMOUS, "Sample.java")
assert [(method.name, method.complexity) for method in methods] == [("outer", 1)]
def test_ignores_constructors_and_names_the_class_for_uml_viewer():
source = """
package demo.pkg;
public class Board {
Board() { if (true) {} }
public void place(int x) { if (x > 0 && ready) return; }
class Inner { void tick() { return; } }
}
"""
methods = functions_in_source(source, "src/demo/pkg/Board.java")
assert [(method.namespace, method.name, method.complexity) for method in methods] == [
("demo.pkg.Board", "place", 3),
("demo.pkg.Board.Inner", "tick", 1),
]
assert methods[0].jacoco_class == "demo.pkg.Board"
assert methods[1].jacoco_class == "demo.pkg.Board$Inner"
def test_keywords_in_comments_and_strings_are_not_decisions():
source = """
class Sample {
int stable() {
String text = "if && || ? case default catch";
// if && || ? case default catch
/* if && || ? case default catch */
return 1;
}
}
"""
methods = functions_in_source(source, "Sample.java")
assert [(method.name, method.complexity) for method in methods] == [("stable", 1)]
+105
View File
@@ -0,0 +1,105 @@
from crapper.languages.python import functions_in_source
def test_function_method_and_nested_function():
source = """
def choose(x, ready):
if x > 0 and x < 10:
return 1
elif x == 0 or ready:
return 0
for i in xs:
if i == 2:
return i
while ready:
break
try:
return x
except ValueError:
return 0
except KeyError:
return -1
match x:
case 1:
return 1
case 2 | 3:
return 2
case _:
return 0
return x if x > 0 else 0
class Box:
def open(self, flag=False):
return 1 if flag else 0
@staticmethod
def shut(self):
xs = [n for n in range(3) if n]
return xs
def outer():
def inner(n):
if n:
return 1
return 0
return inner(1)
async def load():
if True:
return 1
"""
functions = functions_in_source(source, "src/demo/box.py", "/proj")
assert [(fn.namespace, fn.name, fn.complexity) for fn in functions] == [
("demo.box", "choose", 14),
("demo.box.Box", "open", 2),
("demo.box.Box", "shut", 2),
("demo.box", "outer", 2),
("demo.box", "load", 2),
]
def test_nested_class_method_is_its_own_entry():
source = """
class Outer:
class Inner:
def tick(self, ready, ok):
if ready and ok:
return 1
return 0
def factory():
class Hidden:
def secret(self):
if True:
return 1
return 0
return Hidden()
"""
functions = functions_in_source(source, "src/demo/sample.py", "/proj")
assert [(fn.namespace, fn.name, fn.complexity) for fn in functions] == [
("demo.sample.Outer.Inner", "tick", 3),
("demo.sample", "factory", 1),
("demo.sample.Hidden", "secret", 2),
]
def test_namespace_strips_roots_src_and_init():
source = "def f():\n return 1\n"
assert functions_in_source(source, "/proj/src/demo/app.py", "/proj")[0].namespace == "demo.app"
assert functions_in_source(source, "./src/demo/app.py", ".")[0].namespace == "demo.app"
assert functions_in_source(source, "pkg/src/demo/__init__.py", None)[0].namespace == "demo"
assert functions_in_source(source, "__init__.py", None)[0].namespace == "__init__"
def test_strings_comments_and_package_init_do_not_add_decisions():
source = '''
def literal():
"""if and or for while except"""
text = "if x and y"
# if x: return 1
return text
'''
functions = functions_in_source(source, "src/demo/__init__.py", "/proj")
assert [(fn.namespace, fn.name, fn.complexity) for fn in functions] == [
("demo", "literal", 1)
]
+252
View File
@@ -0,0 +1,252 @@
import json
from pathlib import Path
from crapper.discover import is_test_file
from crapper.runners import (
_clean_clojure,
_ensure_vitest_coverage,
_manifest_snapshot,
_python_executable,
_restore_manifests,
_rust_kind,
_vitest_version,
python_coverage_commands,
python_roots,
python_sources,
run_coverage,
rust_coverage_command,
rust_modules,
typescript_command,
typescript_packages,
uses_pytest,
)
def _write(root: Path, relative: str, source: str) -> Path:
path = root / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(source, encoding="utf-8")
return path
def test_vitest_uses_its_own_coverage_instead_of_c8(tmp_path):
_write(
tmp_path,
"package.json",
json.dumps({"scripts": {"test": "vitest run"}}),
)
_write(tmp_path, "vitest.config.ts", "export default {}\n")
source = _write(tmp_path, "src/book.ts", "export function plan() { return 1 }\n")
_write(tmp_path, "src/book.test.ts", "import { plan } from './book'\n")
report_dir = tmp_path / "target" / "coverage" / "typescript"
command = typescript_command(tmp_path, [source, tmp_path / "src" / "book.test.ts"], report_dir)
assert command is not None
assert "c8" not in command
assert "vitest run --coverage" in command
assert "--coverage.reporter=lcov" in command
assert f"--coverage.reportsDirectory={report_dir}" in command
assert "--coverage.include=src/book.ts" in command
assert "book.test.ts" not in command
def test_existing_coverage_script_is_left_alone(tmp_path):
_write(
tmp_path,
"package.json",
json.dumps({"scripts": {"coverage": "vitest run --coverage", "test": "vitest run"}}),
)
command = typescript_command(tmp_path, [], tmp_path / "coverage")
assert command == "npm run coverage"
def test_node_test_script_still_uses_c8(tmp_path):
_write(tmp_path, "package.json", json.dumps({"scripts": {"test": "node --test"}}))
report_dir = tmp_path / "target" / "coverage" / "typescript"
command = typescript_command(tmp_path, [], report_dir)
assert command == f"npx --yes c8 --reporter=lcov --reports-dir {report_dir} npm test"
def test_typescript_package_is_found_from_a_source_file(tmp_path):
_write(tmp_path, "web/package.json", json.dumps({"scripts": {"test": "vitest run"}}))
source = _write(tmp_path, "web/src/app.ts", "export const n = 1\n")
assert typescript_packages(tmp_path, [source]) == [(tmp_path / "web").resolve()]
def test_rust_module_is_the_nearest_cargo_package(tmp_path):
_write(tmp_path, "src-tauri/Cargo.toml", '[package]\nname = "bookwriter"\nversion = "0.1.0"\n')
lib = _write(tmp_path, "src-tauri/src/lib.rs", "pub fn read_text() {}\n")
assert rust_modules(tmp_path, [lib]) == [(tmp_path / "src-tauri").resolve()]
def test_manifests_are_restored_after_a_coverage_install(tmp_path):
package_json = tmp_path / "package.json"
lock = tmp_path / "package-lock.json"
package_json.write_bytes(b'{"name":"demo"}\n')
lock.write_bytes(b'{"lockfileVersion":3}\n')
snapshot = _manifest_snapshot(tmp_path)
package_json.write_bytes(b'{"name":"changed"}\n')
lock.unlink()
shrink = tmp_path / "npm-shrinkwrap.json"
shrink.write_text("{}\n", encoding="utf-8")
_restore_manifests(snapshot)
assert package_json.read_bytes() == b'{"name":"demo"}\n'
assert lock.read_bytes() == b'{"lockfileVersion":3}\n'
assert not shrink.exists()
def test_python_project_uses_pytest_and_coverage_lcov(tmp_path):
_write(tmp_path, "pyproject.toml", "[tool.pytest.ini_options]\ntestpaths = ['tests']\n")
source = _write(tmp_path, "src/demo/box.py", "def choose(x):\n return x\n")
_write(tmp_path, "tests/test_box.py", "def test_choose():\n assert True\n")
assert python_roots(tmp_path, [source]) == [tmp_path.resolve()]
assert uses_pytest(tmp_path)
assert python_sources(tmp_path, [source, tmp_path / "tests" / "test_box.py"]) == "src"
assert is_test_file(tmp_path / "tests" / "test_box.py")
data = tmp_path / "target" / "coverage" / "python" / ".coverage"
report = data.parent / "lcov.info"
run, lcov = python_coverage_commands("python3", "pytest", data, report, "src")
assert run == (
f"python3 -m coverage run --data-file={data} --source=src -m pytest"
)
assert lcov == f"python3 -m coverage lcov --data-file={data} -o {report}"
def test_python_package_without_pytest_uses_unittest(tmp_path):
_write(tmp_path, "setup.py", "from setuptools import setup\nsetup(name='demo')\n")
source = _write(tmp_path, "demo/app.py", "def run():\n return 1\n")
assert not uses_pytest(tmp_path)
run, _lcov = python_coverage_commands("python3", "unittest", Path("data"), Path("out"), "demo")
assert "-m unittest discover -s ." in run
assert python_roots(tmp_path, [source]) == [tmp_path.resolve()]
def test_rust_lcov_path_is_absolute():
report = Path("/tmp/bookwriter/target/coverage/rust/src-tauri/lcov.info")
command = rust_coverage_command("llvm-cov", report)
assert command == f"cargo llvm-cov --lcov --output-path {report}"
def test_vitest_version_and_provider_install(tmp_path, monkeypatch):
assert _vitest_version(tmp_path) is None
_write(tmp_path, "node_modules/vitest/package.json", "{not json")
assert _vitest_version(tmp_path) is None
_write(tmp_path, "node_modules/vitest/package.json", '{"version": "4.1.11"}\n')
assert _vitest_version(tmp_path) == "4.1.11"
provider = tmp_path / "node_modules" / "@vitest" / "coverage-v8"
provider.mkdir(parents=True)
assert _ensure_vitest_coverage(tmp_path) is True
provider.rmdir()
monkeypatch.setattr("crapper.runners.run_shell", lambda command, cwd: 1)
assert _ensure_vitest_coverage(tmp_path) is False
def test_clean_clojure_keeps_other_language_reports(tmp_path):
kept = tmp_path / "target" / "coverage" / "python"
kept.mkdir(parents=True)
(kept / "lcov.info").write_text("keep", encoding="utf-8")
stale = tmp_path / "target" / "coverage" / "lcov.info"
stale.write_text("old", encoding="utf-8")
_clean_clojure(tmp_path)
assert (kept / "lcov.info").is_file()
assert not stale.exists()
def test_python_executable_prefers_a_project_venv(tmp_path):
binary = tmp_path / ".venv" / "bin" / "python"
binary.parent.mkdir(parents=True)
binary.write_text("", encoding="utf-8")
assert _python_executable(tmp_path) == str(binary)
def test_rust_kind_installs_llvm_cov_when_it_is_missing(monkeypatch):
installed = {"llvm": False}
def which(name):
if name == "cargo-llvm-cov" and installed["llvm"]:
return "/bin/cargo-llvm-cov"
if name == "rustup":
return "/bin/rustup"
return None
def shell(command, cwd):
if "cargo install" in command:
installed["llvm"] = True
return 0
monkeypatch.setattr("crapper.runners.shutil.which", which)
monkeypatch.setattr("crapper.runners.run_shell", shell)
assert _rust_kind() == "llvm-cov"
installed["llvm"] = False
monkeypatch.setattr("crapper.runners.run_shell", lambda command, cwd: 1)
assert _rust_kind() is None
def test_coverage_commands_are_issued_per_language(tmp_path, monkeypatch):
commands: list[str] = []
def shell(command, cwd):
commands.append(command)
if "--data-file=" in command:
data = Path(command.split("--data-file=", 1)[1].split()[0])
data.parent.mkdir(parents=True, exist_ok=True)
data.write_text("x", encoding="utf-8")
return 1
if command.startswith("clj") or command.startswith("mvn") or command.startswith("go "):
return 1
if "vitest" in command or command.startswith("cargo "):
return 1
return 0
monkeypatch.setattr("crapper.runners.run_shell", shell)
monkeypatch.setattr(
"crapper.runners.shutil.which",
lambda name: "/bin/cargo-llvm-cov" if name == "cargo-llvm-cov" else None,
)
_write(tmp_path, "deps.edn", "{}\n")
_write(tmp_path, "src/demo/core.clj", "(ns demo.core)\n(defn x [])\n")
_write(tmp_path, "pom.xml", "<project></project>\n")
_write(tmp_path, "src/Board.java", "class Board { int place(){ return 1; } }\n")
(tmp_path / "target" / "jacoco.exec").parent.mkdir(parents=True)
(tmp_path / "target" / "jacoco.exec").write_text("x", encoding="utf-8")
_write(tmp_path, "go.mod", "module example.com/demo\n")
_write(tmp_path, "board.go", "package demo\nfunc Place() int { return 1 }\n")
profile = tmp_path / "target" / "coverage" / "go" / "coverage.out"
profile.parent.mkdir(parents=True)
profile.write_text("old", encoding="utf-8")
_write(tmp_path, "package.json", json.dumps({"scripts": {"test": "vitest run"}}))
_write(tmp_path, "src/ui.ts", "export function view(){ return 1 }\n")
(tmp_path / "node_modules" / "@vitest" / "coverage-v8").mkdir(parents=True)
_write(tmp_path, "pyproject.toml", "[tool.pytest.ini_options]\n")
_write(tmp_path, "src/app.py", "def run():\n return 1\n")
_write(tmp_path, "Cargo.toml", '[package]\nname = "demo"\nversion = "0.1.0"\n')
_write(tmp_path, "src/lib.rs", "pub fn open() {}\n")
files = [
tmp_path / "src/demo/core.clj",
tmp_path / "src/Board.java",
tmp_path / "board.go",
tmp_path / "src/ui.ts",
tmp_path / "src/app.py",
tmp_path / "src/lib.rs",
]
run_coverage(tmp_path, files, None)
text = "\n".join(commands)
assert "clj -M:cov --lcov" in text
assert "clj -M:cov\n" in text or text.endswith("clj -M:cov") or "clj -M:cov" in text
assert "mvn " in text
assert "go test" in text
assert "vitest" in text
assert "coverage run" in text
assert "cargo llvm-cov" in text
run_coverage(tmp_path, files, "echo custom")
assert commands[-1] == "echo custom"
def test_missing_build_files_skip_that_language(tmp_path, monkeypatch):
monkeypatch.setattr("crapper.runners.run_shell", lambda command, cwd: 0)
files = [
_write(tmp_path, "src/App.java", "class App {}\n"),
_write(tmp_path, "main.go", "package main\n"),
_write(tmp_path, "src/app.ts", "export const n = 1\n"),
_write(tmp_path, "src/lib.rs", "pub fn open() {}\n"),
]
run_coverage(tmp_path, files, None)
+95
View File
@@ -0,0 +1,95 @@
from crapper.languages.rust import functions_in_source
SOURCE = """
pub fn choose(x: i32) -> i32 {
if x > 0 && x < 10 {
return 1;
}
match x {
1 => 1,
2 => 2,
_ => 0,
}
}
impl Sample {
pub fn run(&self) -> i32 {
if self.ok { 1 } else { 0 }
}
}
impl Trait for Sample {
fn draw(&mut self) {}
}
mod tests {
fn hidden() {
if true {}
}
}
fn outer() {
fn inner() {
if true {}
}
}
"""
def test_functions_methods_and_skipped_tests(tmp_path):
(tmp_path / "Cargo.toml").write_text(
'[package]\nname = "demo-game"\nversion = "0.1.0"\n',
encoding="utf-8",
)
path = tmp_path / "src" / "lib.rs"
path.parent.mkdir()
path.write_text(SOURCE, encoding="utf-8")
functions = functions_in_source(SOURCE, "src/lib.rs", str(tmp_path))
assert [(fn.namespace, fn.name, fn.complexity) for fn in functions] == [
("demo_game", "choose", 6),
("demo_game::Sample", "run", 2),
("demo_game::Sample", "draw", 1),
("demo_game", "outer", 2),
]
def test_impl_for_a_nested_generic_type(tmp_path):
(tmp_path / "Cargo.toml").write_text(
'[package]\nname = "demo"\nversion = "0.1.0"\n',
encoding="utf-8",
)
source = "impl Trait for Vec<crate::Sample> {\n fn draw(&self) {}\n}\n"
functions = functions_in_source(source, str(tmp_path / "src" / "lib.rs"), str(tmp_path))
assert functions[0].name == "draw"
assert functions[0].namespace.endswith("Vec") or "Sample" in functions[0].namespace
def test_module_paths_outside_the_usual_layout(tmp_path):
(tmp_path / "Cargo.toml").write_text(
'[package]\nname = "demo"\nversion = "0.1.0"\n',
encoding="utf-8",
)
mod = "pub fn open() {}\n"
binary = "pub fn main_ok() {}\n"
functions_in_source(mod, str(tmp_path / "src" / "game" / "mod.rs"), str(tmp_path))
functions = functions_in_source(binary, str(tmp_path / "src" / "bin" / "tool.rs"), str(tmp_path))
assert functions[0].namespace == "tool"
outside = functions_in_source("pub fn loose() {}\n", "/tmp/loose.rs", str(tmp_path))
assert outside[0].name == "loose"
bare = functions_in_source("pub fn bare() {}\n", str(tmp_path / "notes.rs"), None)
assert bare[0].name == "bare"
def test_nested_module_path(tmp_path):
(tmp_path / "Cargo.toml").write_text(
'[package]\nname = "demo"\nversion = "0.1.0"\n',
encoding="utf-8",
)
path = tmp_path / "src" / "game" / "board.rs"
path.parent.mkdir(parents=True)
source = "pub fn place() -> i32 { if ready { 1 } else { 0 } }\n"
path.write_text(source, encoding="utf-8")
functions = functions_in_source(source, "src/game/board.rs", str(tmp_path))
assert [(fn.namespace, fn.name, fn.complexity) for fn in functions] == [
("demo::game::board", "place", 2)
]
+53
View File
@@ -0,0 +1,53 @@
from crapper.languages.typescript import functions_in_source
def test_function_class_method_and_arrow_are_separate_entries():
source = """
export function choose(x: number): number {
if (x > 0 && x < 10) return 1;
for (let i = 0; i < x; i++) {
if (i === 2) return i;
}
try {
return x;
} catch (e) {
return 0;
}
return x > 0 ? 1 : 0;
}
export class Box {
open(flag?: boolean): number {
return flag ? 1 : 0;
}
}
const arrow = (n: number) => (n > 0 ? n : 0);
function outer() {
function inner(n: number) {
if (n) return 1;
return 0;
}
return inner(1);
}
"""
functions = functions_in_source(source, "src/demo/box.ts", "/proj")
assert [(fn.namespace, fn.name, fn.complexity) for fn in functions] == [
("demo.box", "choose", 7),
("demo.box.Box", "open", 2),
("demo.box", "arrow", 2),
("demo.box", "outer", 2),
]
def test_tsx_counts_jsx_logic():
source = """
export function View(ok: boolean, ready: boolean) {
return ok && ready ? 1 : 0;
}
"""
functions = functions_in_source(source, "src/ui/view.tsx", "/proj")
assert [(fn.namespace, fn.name, fn.complexity) for fn in functions] == [
("ui.view", "View", 3)
]