bj/crapper
1
0
Fork 0
mirror of https://github.com/unclebob/crapper.git synced 2026-11-06 18:13:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix the coverage, git, and Clojure bugs, and bring CRAP under 10.

Browse Source 
A path in a built-in coverage command is one argument, a failed custom coverage command does not read a stale report, and git failures exit with git's status. The Clojure reader honors discards, metadata, and character literals. Long functions are split so every score is under 10 except the option switch.
This commit is contained in:
Robert C. Martin
2026-10-03 17:27:59 -05:00
parent f0cb973110
commit 9f1bead298
20 changed files with 1287 additions and 513 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/crapper/analyze.py
+9
View File
@@ -25,12 +25,21 @@ def _coverage(bundle: CoverageBundle | None, function: Function) -> float | None
return found
def _source_path(file: Path, root: Path) -> str:
try:
return file.relative_to(root).as_posix()
except ValueError:
return file.as_posix()
def analyze_files(
files: list[Path],
project_root: Path,
bundle: CoverageBundle | None,
) -> list[Entry]:
root = project_root.resolve()
if bundle is not None:
bundle.bind_sources([_source_path(file.resolve(), root) for file in files])
entries: list[Entry] = []
for file in files:
file = file.resolve()
src/crapper/cli.py
+109 -39
View File
@@ -8,13 +8,34 @@ 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.coverage import CoverageBundle, load_bundle
from crapper.discover import (
CLOJURE_TEST_SUFFIXES,
GO_TEST_SUFFIX,
JS_TEST_SUFFIXES,
SKIP_DIRS,
TEST_DIRS,
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 = """\
_SKIPPED_DIRS = ", ".join(sorted(SKIP_DIRS | TEST_DIRS))
_TEST_PATTERNS = ", ".join(
[
f"*{GO_TEST_SUFFIX}",
*[f"*{suffix}" for suffix in CLOJURE_TEST_SUFFIXES],
*[f"*{suffix}" for suffix in JS_TEST_SUFFIXES],
"test_*.py",
"*_test.py",
"conftest.py",
]
)
HELP = f"""\
Usage: crapper [options] [path-or-filter ...]
Detect the language of each source file and score it with the CRAP metric
@@ -46,8 +67,7 @@ Arguments:
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.
named {_SKIPPED_DIRS} are skipped, as are {_TEST_PATTERNS}.
Coverage, when it is produced:
Clojure clj -M:cov --lcov, then Cloverage form counts or LCOV
@@ -144,26 +164,58 @@ def parse_args(argv: list[str] | None = None) -> Options:
return options
def _changed_files(root: Path) -> list[Path]:
result = subprocess.run(
["git", "status", "--porcelain"],
cwd=root,
class GitError(Exception):
"""git could not describe the working tree. `code` is git's status."""
def __init__(self, code: int, message: str) -> None:
super().__init__(message)
self.code = code
def _git(root: Path, args: list[str]) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["git", "-C", str(root), *args],
check=False,
capture_output=True,
text=True,
encoding="utf-8",
)
if result.returncode != 0:
print(result.stderr.strip() or "git status failed", file=sys.stderr)
return []
def _changed_files(root: Path) -> list[Path]:
"""Added and modified files from git, including files in new directories.
Paths come from `git status -z`, which does not quote non-ASCII names.
They are resolved from the repository root, then limited to `root`, so
`--root` can be a subdirectory. Deleted files are left out.
"""
top_result = _git(root, ["rev-parse", "--show-toplevel"])
if top_result.returncode != 0:
raise GitError(top_result.returncode, top_result.stderr.strip() or "git status failed")
status = _git(
root,
[
"status",
"--porcelain",
"-z",
"--no-renames",
"--untracked-files=all",
"--",
".",
],
)
if status.returncode != 0:
raise GitError(status.returncode, status.stderr.strip() or "git status failed")
top = Path(top_result.stdout.strip())
limit = root.resolve()
found: list[Path] = []
for line in result.stdout.splitlines():
if len(line) < 4:
for entry in status.stdout.split("\0"):
if len(entry) < 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())
path = (top / entry[3:]).resolve()
if path.is_file() and path.is_relative_to(limit):
found.append(path)
return found
@@ -222,38 +274,56 @@ def select_files(options: Options) -> list[Path]:
return sorted({path.resolve() for path in files}, key=lambda path: path.as_posix())
def _show_help(options: Options) -> int:
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
def _coverage_bundle(options: Options, root: Path, files: list[Path]):
if options.no_coverage:
return None
if options.use_existing_coverage:
return load_bundle(root)
status = run_coverage(root, files, options.coverage_command)
if options.coverage_command and status not in (0, None):
return CoverageBundle()
return load_bundle(root)
def _threshold_status(options: Options, entries) -> int:
if options.threshold is None:
return 0
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 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
return _show_help(options)
root = options.project_root.resolve()
files = select_files(options)
try:
files = select_files(options)
except GitError as exc:
print(str(exc), file=sys.stderr)
return exc.code
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)
entries = analyze_files(files, root, _coverage_bundle(options, root, files))
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
return _threshold_status(options, entries)
def main(argv: list[str] | None = None) -> None:
src/crapper/coverage.py
+209 -71
View File
@@ -60,6 +60,129 @@ def suffix_match(path: str, suffix: str) -> bool:
return path_parts[-len(suffix_parts) :] == suffix_parts
def _key_items(keys: list[str]) -> list[tuple[str, str]]:
return [(key, normalize_path(key)) for key in keys]
def _exact_key(items: list[tuple[str, str]], candidates: list[str]) -> str | None:
for candidate in candidates:
for original, norm in items:
if norm == candidate:
return original
return None
def _nonempty_parts(candidates: list[str]) -> list[tuple[str, list[str]]]:
found = []
for candidate in candidates:
parts = _segments(candidate)
if parts:
found.append((candidate, parts))
return found
def _key_has_suffix(key_parts: list[str], parts: list[str]) -> bool:
if len(key_parts) < len(parts):
return False
return key_parts[-len(parts) :] == parts
def _forward_matches(
items: list[tuple[str, str]],
parts_of: list[tuple[str, list[str]]],
source_parts: list[tuple[str, ...]],
) -> list[str]:
forward: list[tuple[int, str]] = []
for _candidate, parts in parts_of:
for original, norm in items:
key_parts = _segments(norm)
if not _key_has_suffix(key_parts, parts):
continue
if _longer_source_owns(key_parts, len(parts), source_parts):
continue
forward.append((len(key_parts) - len(parts), original))
if not forward:
return []
smallest = min(extra for extra, _key in forward)
return list(dict.fromkeys(key for extra, key in forward if extra == smallest))
def _key_is_proper_suffix(parts: list[str], key_parts: list[str]) -> bool:
if not key_parts or len(key_parts) >= len(parts):
return False
return parts[-len(key_parts) :] == key_parts
def _reverse_key(
items: list[tuple[str, str]],
parts_of: list[tuple[str, list[str]]],
source_parts: list[tuple[str, ...]],
) -> str | None:
own = {tuple(parts) for _candidate, parts in parts_of}
reverse: list[str] = []
for _candidate, parts in parts_of:
for original, norm in items:
key_parts = _segments(norm)
if not _key_is_proper_suffix(parts, key_parts):
continue
if _another_source_ends_with(key_parts, source_parts, own):
continue
reverse.append(original)
unique = list(dict.fromkeys(reverse))
if len(unique) == 1:
return unique[0]
return None
def _select_key(
keys: list[str], source_path: str, source_parts: list[tuple[str, ...]]
) -> str | None:
"""Report key for one source file.
An exact path wins. A report path may be the source path plus a prefix
(`proj/src/a.go` for `src/a.go`). It is not a match when another source
file is a longer suffix of that key, so `b/main.go` does not take
`a/b/main.go`.
"""
items = _key_items(keys)
candidates = _candidates(source_path)
exact = _exact_key(items, candidates)
if exact is not None:
return exact
parts_of = _nonempty_parts(candidates)
forward = _forward_matches(items, parts_of, source_parts)
if len(forward) == 1:
return forward[0]
if forward:
return None
return _reverse_key(items, parts_of, source_parts)
def _longer_source_owns(
key_parts: list[str], source_len: int, source_parts: list[tuple[str, ...]]
) -> bool:
for other in source_parts:
if len(other) <= source_len:
continue
if len(other) <= len(key_parts) and key_parts[-len(other) :] == list(other):
return True
return False
def _another_source_ends_with(
key_parts: list[str],
source_parts: list[tuple[str, ...]],
own: set[tuple[str, ...]],
) -> bool:
for other in source_parts:
if other in own or len(other) < len(key_parts):
continue
if list(other[-len(key_parts) :]) == key_parts:
return True
return False
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
@@ -119,6 +242,22 @@ def _add_branch(record: FileCoverage, line: int, taken: str) -> None:
record.branches[line] = (covered + (1 if _branch_hit(taken) else 0), total + 1)
def _store_lcov(out: dict[str, FileCoverage], current_file: str | None, current: FileCoverage) -> None:
if current_file is not None:
out[current_file] = current
def _read_lcov_line(current: FileCoverage, line: str) -> 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)
return
branch = _LCOV_BRDA.match(line)
if branch:
_add_branch(current, int(branch.group(1)), branch.group(4))
def parse_lcov(text: str) -> dict[str, FileCoverage]:
out: dict[str, FileCoverage] = {}
current_file = None
@@ -126,26 +265,16 @@ def parse_lcov(text: str) -> dict[str, FileCoverage]:
for raw in text.splitlines():
line = raw.strip()
if line.startswith("SF:"):
if current_file is not None:
out[current_file] = current
_store_lcov(out, current_file, current)
current_file = line[3:]
current = FileCoverage()
elif line == "end_of_record":
if current_file is not None:
out[current_file] = current
_store_lcov(out, current_file, current)
current_file = None
current = FileCoverage()
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)
continue
branch = _LCOV_BRDA.match(line)
if branch:
_add_branch(current, int(branch.group(1)), branch.group(4))
if current_file is not None:
out[current_file] = current
_read_lcov_line(current, line)
_store_lcov(out, current_file, current)
return out
@@ -189,18 +318,13 @@ def parse_go_profile(text: str) -> dict[str, list[tuple[int, int, int, int]]]:
return out
def _profile_segments(profile, path: str):
def _profile_segments(profile, path: str, source_parts: list[tuple[str, ...]] | None = None):
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
key = _select_key(list(profile), path, source_parts or [])
if key is None:
return None
return profile[key]
def go_percent(
@@ -208,8 +332,9 @@ def go_percent(
path: str,
start: int,
end: int,
source_parts: list[tuple[str, ...]] | None = None,
) -> float | None:
segments = _profile_segments(profile, path)
segments = _profile_segments(profile, path, source_parts)
if segments is None:
return None
total = 0
@@ -250,6 +375,36 @@ def _jacoco_from_index(classes, class_names, method_name, line) -> float | None:
return nearest.percent
def _instruction_counter(method):
for child in list(method):
if child.tag == "counter" and child.get("type") == "INSTRUCTION":
return child
return None
def _method_line(method) -> int:
try:
return int(method.get("line") or "0")
except ValueError:
return 0
def _record_jacoco_method(found: dict[str, list[JacocoMethod]], class_name: str, method) -> None:
if method.tag != "method":
return
counter = _instruction_counter(method)
if counter is None:
return
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=_method_line(method),
)
)
def parse_jacoco_index(text: str) -> dict[str, list[JacocoMethod]]:
"""Key methods as `binary.class.name#method`."""
@@ -259,30 +414,7 @@ def parse_jacoco_index(text: str) -> 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,
)
)
_record_jacoco_method(found, class_name, method)
return found
@@ -292,6 +424,20 @@ class CoverageBundle:
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)
source_parts: list[tuple[str, ...]] = field(default_factory=list)
def bind_sources(self, paths: list[str]) -> None:
"""Remember project files so a short path cannot take a longer file's report."""
found: list[tuple[str, ...]] = []
seen: set[tuple[str, ...]] = set()
for source in paths:
for candidate in _candidates(source):
parts = tuple(_segments(candidate))
if parts and parts not in seen:
seen.add(parts)
found.append(parts)
self.source_parts = found
def percent_for(self, function: Function) -> float | None:
if function.language == "java":
@@ -305,7 +451,11 @@ class CoverageBundle:
)
if function.language == "go":
return go_percent(
self.go_profile, function.path, function.start_line, function.end_line
self.go_profile,
function.path,
function.start_line,
function.end_line,
self.source_parts,
)
html = self._html_lines(function.path)
if html is not None:
@@ -319,23 +469,19 @@ class CoverageBundle:
return percent_for_range(record, function.start_line, function.end_line)
def _html_lines(self, path: str) -> dict[int, tuple[int, int]] | None:
return _lookup(self.form_html, path)
return _lookup(self.form_html, path, self.source_parts)
def _lcov_lines(self, path: str) -> dict[int, tuple[int, int]] | None:
return _lookup(self.lcov, path)
return _lookup(self.lcov, path, self.source_parts)
def _lookup(index: dict[str, dict], source_path: str):
def _lookup(index: dict[str, dict], source_path: str, source_parts: list[tuple[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
key = _select_key(list(index), source_path, source_parts)
if key is None:
return None
return index[key]
def _read(path: Path) -> str | None:
@@ -345,14 +491,7 @@ def _read(path: Path) -> str | None:
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 = list(root.glob("target/coverage/**/lcov.info"))
paths.extend(root.glob("coverage/**/lcov.info"))
return paths
@@ -414,8 +553,7 @@ def _merge_forms(root: Path) -> dict[str, dict[int, tuple[int, int]]]:
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")]
relative = relative[: -len(".html")]
found[relative] = parse_form_coverage(text)
return found
src/crapper/discover.py
+24 -20
View File
@@ -44,6 +44,27 @@ SKIP_DIRS = {
TEST_DIRS = {"test", "tests", "spec", "specs", "__tests__"}
GO_TEST_SUFFIX = "_test.go"
CLOJURE_TEST_SUFFIXES = ("_test.clj", "_test.cljc", "_test.cljs", "_test.bb")
JS_TEST_SUFFIXES = (
".test.ts",
".spec.ts",
".test.tsx",
".spec.tsx",
".test.mts",
".spec.mts",
".test.cts",
".spec.cts",
".test.js",
".spec.js",
".test.jsx",
".spec.jsx",
".test.mjs",
".spec.mjs",
".test.cjs",
".spec.cjs",
)
def language_of(path: str | Path) -> str | None:
file_path = Path(path)
@@ -56,28 +77,11 @@ def language_of(path: str | Path) -> str | None:
def is_test_file(path: str | Path) -> bool:
file_path = Path(path)
name = file_path.name
if name.endswith("_test.go"):
if name.endswith(GO_TEST_SUFFIX):
return True
if name.endswith(("_test.clj", "_test.cljc", "_test.cljs", "_test.bb")):
if name.endswith(CLOJURE_TEST_SUFFIXES):
return True
if name.endswith(
(
".test.ts",
".spec.ts",
".test.tsx",
".spec.tsx",
".test.mts",
".spec.mts",
".test.js",
".spec.js",
".test.jsx",
".spec.jsx",
".test.mjs",
".spec.mjs",
".test.cjs",
".spec.cjs",
)
):
if name.endswith(JS_TEST_SUFFIXES):
return True
if name == "conftest.py":
return True
src/crapper/languages/__init__.py
+17 -17
View File
@@ -1,27 +1,27 @@
"""Per-language function extraction."""
from crapper.languages.clojure import ClojureFactory
from crapper.languages.golang import GoFactory
from crapper.languages.java import JavaFactory
from crapper.languages.language import LanguageFactory, UnknownFactory
from crapper.languages.python import PythonFactory
from crapper.languages.rust import RustFactory
from crapper.languages.typescript import TypeScriptFactory
from crapper.languages.clojure import functions_in_source as clojure_functions
from crapper.languages.golang import functions_in_source as go_functions
from crapper.languages.java import functions_in_source as java_functions
from crapper.languages.python import functions_in_source as python_functions
from crapper.languages.rust import functions_in_source as rust_functions
from crapper.languages.typescript import functions_in_source as typescript_functions
from crapper.model import Function
_FACTORIES: dict[str, LanguageFactory] = {
"clojure": ClojureFactory(),
"java": JavaFactory(),
"go": GoFactory(),
"typescript": TypeScriptFactory(),
"python": PythonFactory(),
"rust": RustFactory(),
_READERS = {
"clojure": clojure_functions,
"java": java_functions,
"go": go_functions,
"typescript": typescript_functions,
"python": python_functions,
"rust": rust_functions,
}
_UNKNOWN = UnknownFactory()
def functions_in_file(
language: str, source: str, path: str, project_root: str
) -> list[Function]:
factory = _FACTORIES.get(language, _UNKNOWN)
return factory.create().functions(source, path, project_root)
reader = _READERS.get(language)
if reader is None:
return []
return reader(source, path, project_root)
src/crapper/languages/clojure.py
+216 -63
View File
@@ -7,14 +7,12 @@ variants, `and`, `or`, `loop`, `catch`, and each clause of `cond`, `condp`,
import re
from crapper.languages.language import Language, LanguageFactory
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(
@@ -33,38 +31,125 @@ _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 _blank_span(source: str, start: int, end: int, out: list[str]) -> None:
for ch in source[start:end]:
out.append("\n" if ch == "\n" else " ")
def _strip_comments(text: str) -> str:
return "\n".join(re.sub(r";.*", "", line) for line in text.splitlines())
def _advance_past_reader_token(source: str, index: int) -> int | None:
"""Index after a string, character, or comment, or None when `index` is code."""
ch = source[index]
if ch == '"':
return _consume_string(source, index, 1)[0]
if ch == "\\":
return _char_literal_end(source, index)
if ch != ";":
return None
while index < len(source) and source[index] != "\n":
index += 1
return index
def _balanced_end(source: str, index: int) -> int:
opener = source[index]
closer = {"(": ")", "[": "]", "{": "}"}[opener]
depth = 0
while index < len(source):
nxt = _advance_past_reader_token(source, index)
if nxt is not None:
index = nxt
continue
ch = source[index]
index += 1
if ch == opener:
depth += 1
elif ch == closer:
depth -= 1
if depth == 0:
return index
return index
def _skip_space_and_comment(source: str, index: int) -> int:
while index < len(source) and source[index].isspace():
index += 1
if index < len(source) and source[index] == ";":
while index < len(source) and source[index] != "\n":
index += 1
return index
def _skip_ignorable(source: str, index: int) -> int:
while True:
nxt = _skip_space_and_comment(source, index)
if nxt == index:
return index
index = nxt
def _end_of_atom(source: str, index: int) -> int:
while index < len(source) and not source[index].isspace() and source[index] not in "()[]{}\";":
index += 1
return index
def _end_of_one_form(source: str, index: int) -> int:
ch = source[index]
if ch in "([{":
return _balanced_end(source, index)
if ch == '"':
return _consume_string(source, index, 1)[0]
if ch == "\\":
return _char_literal_end(source, index)
if source.startswith("#_", index):
return _form_end(source, index + 2)
return _end_of_atom(source, index)
def _form_end(source: str, index: int) -> int:
"""Index just after the next form. `index` may sit on whitespace before it."""
index = _skip_ignorable(source, index)
if index >= len(source):
return index
return _end_of_one_form(source, index)
def without_strings_and_comments(source: str) -> str:
return _strip_comments(_strip_strings(source))
"""Blank strings, comments, character literals, and `#_` discarded forms.
Character literals are blanked before `"` or `;` can be read, so `\\"` does
not open a string and `\\;` does not open a comment.
"""
out: list[str] = []
index = 0
while index < len(source):
if source.startswith("#_", index):
end = _form_end(source, index + 2)
_blank_span(source, index, end, out)
index = end
continue
ch = source[index]
if ch == '"':
end = _consume_string(source, index, 1)[0]
_blank_span(source, index, end, out)
index = end
continue
if ch == "\\":
end = _char_literal_end(source, index)
_blank_span(source, index, end, out)
index = end
continue
if ch == ";":
while index < len(source) and source[index] != "\n":
out.append(" ")
index += 1
continue
out.append(ch)
index += 1
return "".join(out)
def _open_bracket(ch: str) -> bool:
@@ -75,6 +160,24 @@ def _close_bracket(ch: str) -> bool:
return ch in ")}]"
def _on_open_bracket(depth: int, in_form: bool, forms: int) -> tuple[int, bool, int]:
if depth == 1 and not in_form:
forms += 1
return depth + 1, True, forms
def _on_close_bracket(depth: int, in_form: bool, forms: int) -> tuple[int, bool, int]:
if depth == 1:
in_form = False
return depth - 1, in_form, forms
def _on_atom(depth: int, in_form: bool, forms: int) -> tuple[int, bool, int]:
if depth == 1 and not in_form:
forms += 1
return depth, True, forms
def _count_top_level_forms(text: str, start: int) -> int:
depth = 1
forms = 0
@@ -84,20 +187,13 @@ def _count_top_level_forms(text: str, start: int) -> int:
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
depth, in_form, forms = _on_open_bracket(depth, in_form, forms)
elif _close_bracket(ch):
if depth == 1:
in_form = False
depth -= 1
depth, in_form, forms = _on_close_bracket(depth, in_form, forms)
elif ch.isspace():
in_form = depth != 1
else:
if depth == 1 and not in_form:
forms += 1
in_form = True
depth, in_form, forms = _on_atom(depth, in_form, forms)
i += 1
return forms
@@ -128,7 +224,7 @@ def _count_clauses(text: str, form_type: str, match_start: int) -> int:
def _count_cond_decisions(clean: str) -> int:
total = 0
for match in _COND.finditer(clean):
form_type = _COND.match(match.group(0)).group(1)
form_type = match.group(1)
total += _count_clauses(clean, form_type, match.start())
return total
@@ -178,16 +274,73 @@ def _consume_string(source: str, index: int, line: int) -> tuple[int, int]:
return index, line
_NAME_STOP = " \t\n\r()[]{}\""
def _skip_space(form: str, index: int) -> int:
while index < len(form) and form[index].isspace():
index += 1
return index
def _keyword_ends(form: str, after: int) -> bool:
return after >= len(form) or form[after] in _NAME_STOP
def _defn_keyword_end(form: str, index: int) -> int | None:
for candidate in ("defn-", "defn"):
after = index + len(candidate)
if form.startswith(candidate, index) and _keyword_ends(form, after):
return after
return None
def _skip_metadata(form: str, index: int) -> int:
index += 1
if index < len(form) and form[index] == "{":
return _balanced_end(form, index)
while index < len(form) and form[index] not in _NAME_STOP:
index += 1
return index
def _skip_leading_metadata(form: str, index: int) -> int:
while True:
index = _skip_space(form, index)
if index < len(form) and form[index] == "^":
index = _skip_metadata(form, index)
continue
return index
def _read_name(form: str, index: int) -> str | None:
if index >= len(form) or form[index] in "()[]{}\";":
return None
start = index
while index < len(form) and form[index] not in _NAME_STOP:
index += 1
return form[start:index]
def _defn_name(form: str) -> str | None:
if not form.startswith("("):
return None
keyword_at = _defn_keyword_end(form, _skip_space(form, 1))
if keyword_at is None:
return None
return _read_name(form, _skip_leading_metadata(form, keyword_at))
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:
name = _defn_name(form_text)
if name is None:
return
forms.append(
{
"name": matched.group(1),
"name": name,
"start_line": start_line,
"end_line": line,
"text": form_text,
@@ -202,6 +355,20 @@ def _close_form(forms, source, depth, form_start, form_line, index, line):
return max(0, depth - 1), form_start, form_line
def _skip_reader(source: str, index: int, line: int) -> tuple[int, int] | None:
ch = source[index]
if ch == ";":
return _consume_comment(source, index, line)
if ch == '"':
return _consume_string(source, index, line)
if ch == "\\":
return _char_literal_end(source, index), line
if source.startswith("#_", index):
end = _form_end(source, index + 2)
return end, line + source[index:end].count("\n")
return None
def _extract_top_level_defns(source: str) -> list[dict]:
forms: list[dict] = []
index = 0
@@ -210,16 +377,11 @@ def _extract_top_level_defns(source: str) -> list[dict]:
form_start = None
form_line = None
while index < len(source):
skipped = _skip_reader(source, index, line)
if skipped is not None:
index, line = skipped
continue
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
@@ -250,8 +412,9 @@ def extract_functions(source: str) -> list[dict]:
def declared_namespace(source: str) -> str | None:
clean = without_strings_and_comments(source)
for pattern in (_NS, _IN_NS, _IN_NS_QUOTE):
match = pattern.search(source)
match = pattern.search(clean)
if match:
return match.group(1)
return None
@@ -286,13 +449,3 @@ def functions_in_source(
)
for item in extract_functions(source)
]
class Clojure(Language):
def functions(self, source: str, path: str, project_root: str) -> list[Function]:
return functions_in_source(source, path, project_root)
class ClojureFactory(LanguageFactory):
def create(self) -> Language:
return Clojure()
src/crapper/languages/golang.py
+1 -20
View File
@@ -10,6 +10,7 @@ uml-viewer makes between a class and its operations.
from pathlib import Path
from crapper.languages.treesitter import (
absolute_path as _absolute,
binary_logic,
child_of_type,
complexity,
@@ -19,7 +20,6 @@ from crapper.languages.treesitter import (
parse,
start_line,
)
from crapper.languages.language import Language, LanguageFactory
from crapper.model import Function
_DECISIONS = {
@@ -106,15 +106,6 @@ def package_namespace(path: str, package_name: str) -> str:
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]:
@@ -150,13 +141,3 @@ def functions_in_source(
)
)
return found
class Go(Language):
def functions(self, source: str, path: str, project_root: str) -> list[Function]:
return functions_in_source(source, path, project_root)
class GoFactory(LanguageFactory):
def create(self) -> Language:
return Go()
src/crapper/languages/java.py
+9 -19
View File
@@ -17,7 +17,6 @@ from crapper.languages.treesitter import (
parse,
start_line,
)
from crapper.languages.language import Language, LanguageFactory
from crapper.model import Function
_PACKAGE = re.compile(r"(?m)^\s*package\s+([a-zA-Z_][\w.]*)\s*;")
@@ -86,12 +85,16 @@ def _qualify(package: str | None, names: list[str], nested: str) -> str:
return body
def functions_in_source(source: str, path: str) -> list[Function]:
def functions_in_source(
source: str, path: str, project_root: str | None = None
) -> list[Function]:
del project_root
data, tree = parse(source, "java")
package = _package_name(source)
found: list[Function] = []
def visit(node) -> None:
stack = [tree.root_node]
while stack:
node = stack.pop()
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")
@@ -109,19 +112,6 @@ def functions_in_source(source: str, path: str) -> list[Function]:
jacoco_class=_qualify(package, names, "$"),
)
)
return
for child in node.children:
visit(child)
visit(tree.root_node)
continue
stack.extend(reversed(node.children))
return found
class Java(Language):
def functions(self, source: str, path: str, project_root: str) -> list[Function]:
return functions_in_source(source, path)
class JavaFactory(LanguageFactory):
def create(self) -> Language:
return Java()
src/crapper/languages/language.py
-27
View File
@@ -1,27 +0,0 @@
"""The language operation, and the factories that create language instances."""
from abc import ABC, abstractmethod
from crapper.model import Function
class Language(ABC):
@abstractmethod
def functions(self, source: str, path: str, project_root: str) -> list[Function]:
"""Functions and methods defined in one source file."""
class Unknown(Language):
def functions(self, source: str, path: str, project_root: str) -> list[Function]:
return []
class LanguageFactory(ABC):
@abstractmethod
def create(self) -> Language:
"""A language instance ready to read source files."""
class UnknownFactory(LanguageFactory):
def create(self) -> Language:
return Unknown()
src/crapper/languages/python.py
-11
View File
@@ -16,7 +16,6 @@ from crapper.languages.treesitter import (
parse,
start_line,
)
from crapper.languages.language import Language, LanguageFactory
from crapper.model import Function
_DECISIONS = {
@@ -132,13 +131,3 @@ def functions_in_source(
)
)
return found
class Python(Language):
def functions(self, source: str, path: str, project_root: str) -> list[Function]:
return functions_in_source(source, path, project_root)
class PythonFactory(LanguageFactory):
def create(self) -> Language:
return Python()
src/crapper/languages/rust.py
+30 -39
View File
@@ -11,6 +11,7 @@ import re
from pathlib import Path
from crapper.languages.treesitter import (
absolute_path as _absolute,
binary_logic,
child_of_type,
complexity,
@@ -20,7 +21,6 @@ from crapper.languages.treesitter import (
parse,
start_line,
)
from crapper.languages.language import Language, LanguageFactory
from crapper.model import Function
_DECISIONS = {
@@ -64,6 +64,15 @@ def _in_mod_named(data: bytes, node, name: str) -> bool:
return name in _ancestor_mods(data, node)
def _first_type_name(data: bytes, node) -> str | None:
if node.type == "type_identifier":
return node_text(data, node)
for item in descendants(node):
if item.type == "type_identifier":
return node_text(data, item)
return None
def _impl_type(data: bytes, impl) -> str | None:
saw_for = False
chosen = None
@@ -73,16 +82,7 @@ def _impl_type(data: bytes, impl) -> str | None:
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
chosen = _first_type_name(data, child)
if saw_for:
return chosen
return chosen
@@ -105,10 +105,27 @@ def _crate_name(path: str) -> tuple[str, Path | None]:
return "crate", None
def _rs_parts(relative: str) -> list[str]:
parts = relative.split("/")
if parts[-1] == "mod.rs":
return parts[:-1]
if parts[-1].endswith(".rs"):
parts[-1] = parts[-1][:-3]
return parts
def _with_crate(crate_name: str, parts: list[str]) -> list[str]:
if parts[:1] == ["bin"] and len(parts) >= 2:
return parts[1:]
return [crate_name, *parts]
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]
if file_path.stem in {"lib", "main"}:
return [crate_name]
return [crate_name, file_path.stem]
try:
relative = file_path.relative_to(crate_root).as_posix()
except ValueError:
@@ -117,14 +134,7 @@ def _file_modules(path: str, crate_name: str, crate_root: Path | None) -> list[s
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]
return _with_crate(crate_name, _rs_parts(relative))
def _namespace(modules: list[str], extra: list[str], type_name: str | None) -> str:
@@ -134,15 +144,6 @@ def _namespace(modules: list[str], extra: list[str], type_name: str | None) -> s
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":
@@ -187,13 +188,3 @@ def functions_in_source(
if recorded is not None:
found.append(recorded)
return found
class Rust(Language):
def functions(self, source: str, path: str, project_root: str) -> list[Function]:
return functions_in_source(source, path, project_root)
class RustFactory(LanguageFactory):
def create(self) -> Language:
return Rust()
src/crapper/languages/treesitter.py
+24 -9
View File
@@ -1,6 +1,7 @@
"""Shared tree-sitter helpers for Java, Go, TypeScript, Rust, and Python."""
from functools import lru_cache
from pathlib import Path
@lru_cache(maxsize=None)
@@ -47,18 +48,32 @@ def descendants(node):
stack.extend(reversed(current.children))
def absolute_path(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 complexity(node, is_decision, skip) -> int:
"""McCabe complexity of `node`: 1 plus each decision in the subtree."""
"""McCabe complexity of `node`: 1 plus each decision in the subtree.
def walk(current) -> int:
The walk is a stack, not recursion. A generated expression can be deeper
than Python's call limit.
"""
score = 1
stack = [node]
while stack:
current = stack.pop()
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)
continue
if is_decision(current):
score += 1
stack.extend(reversed(current.children))
return score
def binary_logic(node) -> bool:
src/crapper/languages/typescript.py
+24 -22
View File
@@ -21,7 +21,6 @@ from crapper.languages.treesitter import (
parse,
start_line,
)
from crapper.languages.language import Language, LanguageFactory
from crapper.model import Function
_DECISIONS = {
@@ -86,22 +85,35 @@ def _grammar(path: str) -> str:
return "typescript"
def _module_namespace(path: str, source_root: str | None) -> str:
relative = path.replace("\\", "/")
def _strip_root_prefix(relative: str, source_root: str | None) -> str:
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:]
return relative[len(root) + 1 :]
if relative.startswith("./"):
return relative[2:]
return relative
def _strip_src_dir(relative: str) -> str:
if relative.startswith("src/"):
relative = relative[4:]
elif "/src/" in relative:
relative = relative.split("/src/", 1)[1]
return relative[4:]
if "/src/" in relative:
return relative.split("/src/", 1)[1]
return relative
def _strip_script_suffix(relative: str) -> str:
for suffix in (".tsx", ".mts", ".cts", ".jsx", ".mjs", ".cjs", ".ts", ".js"):
if relative.endswith(suffix):
relative = relative[: -len(suffix)]
break
return relative.replace("/", ".")
return relative[: -len(suffix)]
return relative
def _module_namespace(path: str, source_root: str | None) -> str:
relative = path.replace("\\", "/")
relative = _strip_root_prefix(relative, source_root)
relative = _strip_src_dir(relative)
return _strip_script_suffix(relative).replace("/", ".")
def _inside_function(node) -> bool:
@@ -385,13 +397,3 @@ def functions_in_source(
_append_routes(found, routes, data, module, path)
found.sort(key=lambda item: item[0])
return [function for _, function in found]
class TypeScript(Language):
def functions(self, source: str, path: str, project_root: str) -> list[Function]:
return functions_in_source(source, path, project_root)
class TypeScriptFactory(LanguageFactory):
def create(self) -> Language:
return TypeScript()
src/crapper/runners.py
+163 -76
View File
@@ -15,22 +15,39 @@ 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"
)
_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}")
def _show(command: str | list[str]) -> str:
if isinstance(command, str):
return command
return " ".join(shlex.quote(part) for part in command)
def run_shell(command: str | list[str], cwd: Path) -> int:
"""Run a coverage command.
A list is an argument vector and does not go through a shell, so a path
with spaces or metacharacters stays one argument. A string is a command
the user typed in `--coverage-command` and runs in a shell.
"""
_warn(f"+ ({cwd}) {_show(command)}")
try:
completed = subprocess.run(command, cwd=cwd, shell=True)
if isinstance(command, str):
completed = subprocess.run(command, cwd=cwd, shell=True)
else:
completed = subprocess.run(list(command), cwd=cwd)
except OSError as exc:
_warn(f"Coverage command failed to start: {exc}")
return 127
@@ -53,18 +70,49 @@ def _clean_dir(path: Path) -> None:
shutil.rmtree(path)
_COVERAGE_DIRS = {"typescript", "rust", "go", "python"}
def _outside_language_dirs(coverage: Path, path: Path) -> bool:
relative = path.relative_to(coverage)
if path.name in _COVERAGE_DIRS:
return False
return not _COVERAGE_DIRS.intersection(relative.parts)
def _remove_cloverage_html(coverage: Path) -> None:
for path in coverage.rglob("*.html"):
if not _outside_language_dirs(coverage, path):
continue
path.unlink()
def _remove_empty_dirs(coverage: Path) -> None:
directories = [path for path in coverage.rglob("*") if path.is_dir()]
for path in sorted(directories, reverse=True):
if not _outside_language_dirs(coverage, path):
continue
try:
path.rmdir()
except OSError:
continue
def _clean_clojure(root: Path) -> None:
"""Remove Cloverage's HTML and top-level lcov without touching other reports.
Other languages keep their directories. Files that are not Cloverage HTML
stay, including anything a project keeps under target/coverage itself.
"""
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()
lcov = coverage / "lcov.info"
if lcov.is_file():
lcov.unlink()
_remove_cloverage_html(coverage)
_remove_empty_dirs(coverage)
_VITEST_CONFIGS = (
@@ -77,15 +125,21 @@ _VITEST_CONFIGS = (
)
class PackageJsonError(Exception):
"""package.json is present and cannot be read as an object."""
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
except json.JSONDecodeError as exc:
raise PackageJsonError(f"{path}: {exc.msg}") from exc
if not isinstance(data, dict):
raise PackageJsonError(f"{path}: expected a JSON object")
return data
def _uses_vitest(package: Path, scripts: dict) -> bool:
@@ -99,7 +153,8 @@ def _vitest_version(package: Path) -> str | None:
if installed.is_file():
try:
data = json.loads(installed.read_text(encoding="utf-8"))
except json.JSONDecodeError:
except json.JSONDecodeError as exc:
_warn(f"{installed}: {exc.msg}")
data = None
if isinstance(data, dict) and isinstance(data.get("version"), str):
return data["version"]
@@ -135,8 +190,13 @@ def _ensure_vitest_coverage(package: Path) -> bool:
"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)
try:
code = run_shell(
["npm", "install", "--no-save", "--no-package-lock", spec],
package,
)
finally:
_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 score 0%.")
@@ -155,8 +215,10 @@ def typescript_packages(root: Path, files: list[Path]) -> list[Path]:
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."""
def typescript_command(
package: Path, files: list[Path], report_dir: Path
) -> list[str] | None:
"""Argument vector that writes LCOV into `report_dir`, or None when there is no test script."""
data = _package_json(package)
if data is None:
@@ -165,13 +227,21 @@ def typescript_command(package: Path, files: list[Path], report_dir: Path) -> st
if not isinstance(scripts, dict):
scripts = {}
if "coverage" in scripts:
return "npm run coverage"
quoted_dir = shlex.quote(str(report_dir))
return ["npm", "run", "coverage"]
if _uses_vitest(package, scripts):
return _vitest_coverage_command(package, files, quoted_dir)
return _vitest_coverage_command(package, files, report_dir)
if "test" not in scripts:
return None
return f"npx --yes c8 --reporter=lcov --reports-dir {quoted_dir} npm test"
return [
"npx",
"--yes",
"c8",
"--reporter=lcov",
"--reports-dir",
str(report_dir),
"npm",
"test",
]
def _vitest_includes(package: Path, files: list[Path]) -> list[str]:
@@ -184,22 +254,28 @@ def _vitest_includes(package: Path, files: list[Path]) -> list[str]:
relative = Path(file).resolve().relative_to(package)
except ValueError:
continue
includes.append(shlex.quote(relative.as_posix()))
includes.append(relative.as_posix())
return includes
def _vitest_coverage_command(package: Path, files: list[Path], quoted_dir: str) -> str:
def _vitest_coverage_command(package: Path, files: list[Path], report_dir: Path) -> list[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",
]
if binary.is_file():
command = [str(binary)]
else:
command = ["npx", "vitest"]
command.extend(
[
"run",
"--coverage",
"--coverage.reporter=lcov",
f"--coverage.reportsDirectory={report_dir}",
"--coverage.reportOnFailure=true",
]
)
for include in _vitest_includes(package, files):
flags.append(f"--coverage.include={include}")
return f"{runner} run " + " ".join(flags)
command.append(f"--coverage.include={include}")
return command
def _coverage_report(root: Path, module: Path, language: str) -> Path:
@@ -212,15 +288,11 @@ def _coverage_report(root: Path, module: Path, language: str) -> Path:
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)
return [
module.resolve()
for module in _modules_with(files, ".rs", "Cargo.toml", root)
]
def _rust_kind() -> str | None:
@@ -230,10 +302,10 @@ def _rust_kind() -> str | None:
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())
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())
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 score 0%.")
@@ -262,7 +334,7 @@ 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 str(candidate)
return "python3"
@@ -295,38 +367,36 @@ def python_sources(package: Path, files: list[Path]) -> str:
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 ""
) -> tuple[list[str], list[str]]:
run = [py, "-m", "coverage", "run", f"--data-file={data_file}"]
if source:
run.append(f"--source={source}")
if kind == "pytest":
module = "pytest"
run.extend(["-m", "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}"
run.extend(["-m", "unittest", "discover", "-s", "."])
lcov = [py, "-m", "coverage", "lcov", f"--data-file={data_file}", "-o", str(report)]
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:
if run_shell([py, "-c", f"import {module}"], 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}",
[py, "-m", "pip", "install", "--disable-pip-version-check", module],
package,
)
return code == 0
def rust_coverage_command(kind: str, report: Path) -> str:
def rust_coverage_command(kind: str, report: Path) -> list[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))}"
return ["cargo", "llvm-cov", "--lcov", "--output-path", str(report)]
return ["cargo", "tarpaulin", "--out", "Lcov", "--output-dir", str(report.parent)]
def _languages_in(files: list[Path]) -> set[str]:
@@ -354,10 +424,10 @@ def _cover_clojure(root: Path) -> None:
_warn("No deps.edn or bb.edn; skipping Clojure coverage.")
return
_clean_clojure(root)
code = run_shell("clj -M:cov --lcov", 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)
code = run_shell(["clj", "-M:cov"], root)
if code != 0:
_warn(f"Clojure coverage exited {code}. Clojure coverage will score 0%.")
@@ -387,7 +457,7 @@ def _cover_go(root: Path, files: list[Path]) -> None:
profile.parent.mkdir(parents=True, exist_ok=True)
if profile.exists():
profile.unlink()
code = run_shell(f"go test ./... -coverprofile={profile}", module)
code = run_shell(["go", "test", "./...", f"-coverprofile={profile}"], module)
if code != 0:
_warn(f"Go coverage exited {code} in {module}. Go coverage will score 0%.")
@@ -405,7 +475,11 @@ def _cover_typescript(root: Path, files: list[Path]) -> None:
return
for package in packages:
report = _coverage_report(root, package, "typescript")
command = typescript_command(package, files, report.parent)
try:
command = typescript_command(package, files, report.parent)
except PackageJsonError as exc:
_warn(str(exc))
continue
if command is None:
_warn(f"No package.json test script in {package}; skipping TypeScript coverage.")
continue
@@ -417,8 +491,10 @@ def _cover_typescript(root: Path, files: list[Path]) -> None:
_warn(f"TypeScript coverage exited {code} in {package}. TypeScript coverage will score 0%.")
def _needs_vitest_provider(command: str) -> bool:
return command != "npm run coverage" and "vitest" in command
def _needs_vitest_provider(command: list[str]) -> bool:
if command[:3] == ["npm", "run", "coverage"]:
return False
return any(Path(part).name == "vitest" for part in command)
def _python_kind(py: str, package: Path) -> str:
@@ -430,7 +506,9 @@ def _python_kind(py: str, package: Path) -> str:
return "unittest"
def _record_python_lcov(package: Path, code: int, data_file: Path, lcov_cmd: str) -> None:
def _record_python_lcov(
package: Path, code: int, data_file: Path, lcov_cmd: str | list[str]
) -> None:
if not data_file.exists():
if code != 0:
_warn(f"Python coverage exited {code} in {package}. Python coverage will score 0%.")
@@ -476,15 +554,23 @@ def _cover_rust(root: Path, files: list[Path]) -> None:
_warn(f"Rust coverage exited {code} in {module}. Rust coverage will score 0%.")
def run_coverage(root: Path, files: list[Path], command: str | None) -> None:
"""Generate coverage reports for the languages present in `files`."""
def run_coverage(root: Path, files: list[Path], command: str | None) -> int:
"""Generate coverage reports for the languages present in `files`.
Returns the custom command's status. A non-zero status means the caller
must not read reports already on disk. Per-language runs return 0; a
failed tool is reported and that language scores 0% when it wrote nothing.
"""
root = root.resolve()
if command:
code = run_shell(command, root)
if code != 0:
_warn(f"Coverage command exited {code}. Coverage may score 0%.")
return
_warn(
f"Coverage command exited {code}. "
"Reports already on disk will not be read."
)
return code
languages = _languages_in(files)
if "clojure" in languages:
@@ -499,3 +585,4 @@ def run_coverage(root: Path, files: list[Path], command: str | None) -> None:
_cover_python(root, files)
if "rust" in languages:
_cover_rust(root, files)
return 0
tests/test_cli.py
+156 -49
View File
@@ -1,11 +1,12 @@
import os
import subprocess
import sys
from pathlib import Path
import pytest
from crapper.cli import _changed_files, _positionals, _take, main, parse_args, run
from crapper.discover import language_of
from crapper.cli import GitError, _changed_files, _positionals, _take, main, parse_args, run
from crapper.discover import is_test_file, language_of
def test_language_detection():
@@ -22,6 +23,11 @@ def test_language_detection():
assert language_of("src/crapper/cli.py") == "python"
assert language_of("types.d.ts") is None
assert language_of("notes.md") is None
assert language_of("app.cts") == "typescript"
assert language_of("app.test.cts") == "typescript"
assert is_test_file("app.test.cts") is True
assert is_test_file("app.spec.cts") is True
assert is_test_file("app.cts") is False
def _write(root: Path, relative: str, source: str) -> None:
@@ -30,6 +36,22 @@ def _write(root: Path, relative: str, source: str) -> None:
path.write_text(source, encoding="utf-8")
class _Completed:
def __init__(self, returncode: int, stdout: str, stderr: str = "") -> None:
self.returncode = returncode
self.stdout = stdout
self.stderr = stderr
def _git_status(root: Path, status: str):
def fake(args, **_kwargs):
if "rev-parse" in args:
return _Completed(0, f"{root.resolve()}\n")
return _Completed(0, status)
return fake
def test_project_snapshot_is_readable_as_crap4clj_edn(tmp_path):
_write(
tmp_path,
@@ -135,7 +157,10 @@ def test_threshold_exits_when_the_worst_score_is_too_high(tmp_path):
def test_help_and_unknown_option(capsys):
assert run(["--help"]) == 0
assert "uml-viewer" in capsys.readouterr().out
help_text = capsys.readouterr().out
assert "uml-viewer" in help_text
assert ".clj-kondo" in help_text
assert ".test.cts" in help_text
assert run(["--not-an-option"]) == 1
assert run(["--root"]) == 1
assert run(["--threshold", "nope"]) == 1
@@ -151,21 +176,21 @@ def test_selects_directories_filters_and_changed_files(tmp_path, monkeypatch, ca
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())
monkeypatch.setattr(
"crapper.cli.subprocess.run",
_git_status(tmp_path, " M src/demo/core.clj\0"),
)
assert run(["--root", str(tmp_path), "--no-coverage", "--changed"]) == 0
assert ':name "choose"' in (tmp_path / ".metrics" / "crap.edn").read_text(encoding="utf-8")
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
monkeypatch.setattr(
"crapper.cli.subprocess.run",
lambda *_args, **_kwargs: _Completed(128, "", "fatal: not a git repository"),
)
assert run(["--root", str(tmp_path), "--no-coverage", "--changed"]) == 128
captured = capsys.readouterr()
assert "not a git repository" in captured.err
assert "No source files" not in captured.out
def test_run_asks_for_coverage_when_it_is_enabled(tmp_path, monkeypatch):
@@ -201,43 +226,46 @@ def test_flags_that_only_store_a_boolean():
def test_changed_files_follow_git_status(tmp_path, monkeypatch, capsys):
seen = {}
named = tmp_path / "x"
named.write_text("x", encoding="utf-8")
keep = tmp_path / "sources" / "keep.py"
keep.parent.mkdir()
keep.write_text("x = 1\n", encoding="utf-8")
class Result:
returncode = 0
stdout = "\nM x\n?? sources/keep.py\n"
stderr = ""
def fake(*_args, **kwargs):
def fake(args, **kwargs):
seen["args"] = args
seen.update(kwargs)
return Result()
if "rev-parse" in args:
return _Completed(0, f"{tmp_path.resolve()}\n")
return _Completed(0, " M x\0?? sources/keep.py\0")
monkeypatch.setattr("crapper.cli.subprocess.run", fake)
found = _changed_files(tmp_path)
assert seen["check"] is False
assert seen["capture_output"] is True
assert seen["text"] is True
assert found == [
(tmp_path / "x").resolve(),
(tmp_path / "sources" / "keep.py").resolve(),
]
assert seen["encoding"] == "utf-8"
assert "-z" in seen["args"]
assert "--no-renames" in seen["args"]
assert "--untracked-files=all" in seen["args"]
assert found == [named.resolve(), keep.resolve()]
class Failed:
returncode = 1
stdout = "?? sources/keep.py\n"
stderr = "git failed"
monkeypatch.setattr(
"crapper.cli.subprocess.run",
lambda *_a, **_k: _Completed(1, "", "git failed"),
)
with pytest.raises(GitError, match="git failed") as failed:
_changed_files(tmp_path)
assert failed.value.code == 1
assert capsys.readouterr().err == ""
monkeypatch.setattr("crapper.cli.subprocess.run", lambda *_a, **_k: Failed())
assert _changed_files(tmp_path) == []
assert capsys.readouterr().err.strip() == "git failed"
class Silent:
returncode = 1
stdout = ""
stderr = ""
monkeypatch.setattr("crapper.cli.subprocess.run", lambda *_a, **_k: Silent())
assert _changed_files(tmp_path) == []
assert capsys.readouterr().err.strip() == "git status failed"
monkeypatch.setattr(
"crapper.cli.subprocess.run",
lambda *_a, **_k: _Completed(128, "", ""),
)
with pytest.raises(GitError, match="git status failed") as silent:
_changed_files(tmp_path)
assert silent.value.code == 128
def test_relative_paths_are_resolved_under_the_project_root(tmp_path):
@@ -255,12 +283,10 @@ def test_changed_selection_ignores_files_git_did_not_name(tmp_path, monkeypatch,
_write(tmp_path, "src/kept.clj", "(ns demo.kept)\n\n(defn kept [] 1)\n")
_write(tmp_path, "src/other.clj", "(ns demo.other)\n\n(defn other [] 1)\n")
class Result:
returncode = 0
stdout = "?? src/kept.clj\n"
stderr = ""
monkeypatch.setattr("crapper.cli.subprocess.run", lambda *_a, **_k: Result())
monkeypatch.setattr(
"crapper.cli.subprocess.run",
_git_status(tmp_path, "?? src/kept.clj\0"),
)
assert run(["--root", str(tmp_path), "--no-coverage", "--changed"]) == 0
text = (tmp_path / ".metrics" / "crap.edn").read_text(encoding="utf-8")
assert ':name "kept"' in text
@@ -284,6 +310,87 @@ def test_threshold_with_no_scores_does_not_call_max_on_an_empty_list(tmp_path):
assert run(["--root", str(tmp_path), "--no-coverage", "--threshold", "1"]) == 0
def test_a_failed_coverage_command_ignores_a_stale_report(tmp_path, monkeypatch):
_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")
monkeypatch.setattr("crapper.cli.run_coverage", lambda *_args, **_kwargs: 3)
code = run(
["--root", str(tmp_path), "--coverage-command", "exit 3", "src/demo/core.clj"]
)
assert code == 0
text = (tmp_path / ".metrics" / "crap.edn").read_text(encoding="utf-8")
assert ":coverage 0.0" in text
assert ":crap 6.0" in text
monkeypatch.setattr("crapper.cli.run_coverage", lambda *_args, **_kwargs: 0)
assert run(
["--root", str(tmp_path), "--coverage-command", "true", "src/demo/core.clj"]
) == 0
text = (tmp_path / ".metrics" / "crap.edn").read_text(encoding="utf-8")
assert ":coverage 100.0" in text
def test_changed_files_from_a_real_git_status(tmp_path):
_git = ["git", "-C", str(tmp_path)]
subprocess.run([*_git, "init"], check=True, capture_output=True)
subprocess.run([*_git, "config", "user.email", "[email protected]"], check=True)
subprocess.run([*_git, "config", "user.name", "t"], check=True)
keep = tmp_path / "keep.go"
keep.write_text("package demo\nfunc Keep() int { return 1 }\n", encoding="utf-8")
subprocess.run([*_git, "add", "keep.go"], check=True)
subprocess.run([*_git, "commit", "-m", "init"], check=True, capture_output=True)
keep.unlink()
cafe = tmp_path / "café.go"
cafe.write_text("package demo\nfunc Cafe() int { return 1 }\n", encoding="utf-8")
fresh = tmp_path / "fresh"
fresh.mkdir()
(fresh / "new.go").write_text("package demo\nfunc New() int { return 1 }\n", encoding="utf-8")
assert run(["--root", str(tmp_path), "--no-coverage", "--changed"]) == 0
text = (tmp_path / ".metrics" / "crap.edn").read_text(encoding="utf-8")
assert ':name "Cafe"' in text
assert ':name "New"' in text
assert "Keep" not in text
pkg = tmp_path / "pkg"
pkg.mkdir()
(pkg / "inside.go").write_text("package pkg\nfunc Inside() int { return 1 }\n", encoding="utf-8")
(tmp_path / "outside.go").write_text(
"package demo\nfunc Outside() int { return 1 }\n",
encoding="utf-8",
)
assert run(["--root", str(pkg), "--no-coverage", "--changed"]) == 0
nested = (pkg / ".metrics" / "crap.edn").read_text(encoding="utf-8")
assert ':name "Inside"' in nested
assert "Outside" not in nested
assert "Cafe" not in nested
assert "New" not in nested
def test_changed_outside_a_repository_returns_gits_status(tmp_path, capsys):
_write(tmp_path, "src/app.py", "def run():\n return 1\n")
code = run(["--root", str(tmp_path), "--no-coverage", "--changed"])
assert code != 0
captured = capsys.readouterr()
assert "No source files" not in captured.out
assert captured.err.strip()
def test_uml_loader_rejects_a_missing_viewer(tmp_path):
script = Path(__file__).resolve().parents[1] / "uml"
completed = subprocess.run(
[str(script)],
env={**os.environ, "UML_VIEWER_ROOT": str(tmp_path / "missing")},
capture_output=True,
text=True,
check=False,
)
assert completed.returncode == 1
assert "uml-viewer checkout not found" in completed.stderr
def test_main_exits_with_the_status(monkeypatch):
monkeypatch.setattr("crapper.cli.run", lambda _argv=None: 4)
try:
tests/test_clojure.py
+62
View File
@@ -101,6 +101,68 @@ def test_strings_comments_and_character_literals_stay_inside_one_form():
assert [item["name"] for item in extract_functions(source)] == ["foo"]
def test_discards_and_metadata_cover_each_reader_form():
source = r"""
#_
; gone
(defn ghost [] (if x 1 0))
#_ "string"
#_ \x
#_ symbol
#_ #_(defn inner [] (if x 1 0))
#_(defn boxed []
"str"
\;
; comment
(if x 1 0))
(defn ^{:doc "hi" :private true} kept [] 1)
(defn ^{:flag \; :private true} also [] 1)
(defn ^{:private true ; note
} third [] 1)
( defn spaced [] 1)
(defn)
(defn ^)
"""
assert [item["name"] for item in extract_functions(source)] == [
"kept",
"also",
"third",
"spaced",
]
assert extract_functions("(defn ^{:private kept [] 1)") == []
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"]
def test_reader_keeps_the_real_namespace_name_and_decisions():
source = r""";; (ns demo.wrong)
(ns demo.core)
#_(defn dropped [x]
(if x 1 0))
(defn ^:private kept [x]
#_(if x 1 0)
1)
(defn ^{:private true} also [] 1)
(defn quoted []
\"
(if x 1 0))
(defn semi []
\;
(if true 1 2))
"""
found = functions_in_source(source, "src/demo/core.clj", "/proj")
assert [(item.name, item.complexity) for item in found] == [
("kept", 1),
("also", 1),
("quoted", 2),
("semi", 2),
]
assert {item.namespace for item in found} == {"demo.core"}
assert cyclomatic_complexity(r'(defn semi [] \; (if true 1 2))') == 2
assert cyclomatic_complexity(r'(defn quoted [] \" (if x 1 0))') == 2
assert extract_functions("#_(defn dropped [] (if x 1 0))\n(defn kept [] 1)\n") == [
{"name": "kept", "start_line": 2, "end_line": 2, "complexity": 1}
]
tests/test_edges.py
+75
View File
@@ -123,6 +123,52 @@ def test_go_coverage_is_read_from_the_bundle():
assert bundle.percent_for(function()) == 100.0
def test_the_most_specific_prefix_wins_and_a_tie_matches_nothing():
specific = parse_go_profile(
"mode: set\n"
"other/proj/src/a.go:1.1,2.2 1 0\n"
"proj/src/a.go:1.1,2.2 1 1\n"
)
assert go_percent(specific, "src/a.go", 1, 2) == 100.0
tied = parse_go_profile(
"mode: set\nproj/a.go:1.1,2.2 1 1\nother/a.go:1.1,2.2 1 0\n"
)
assert go_percent(tied, "a.go", 1, 2) is None
assert go_percent(specific, "/", 1, 2) is None
def test_exact_profile_key_wins_over_a_suffix():
profile = parse_go_profile(
"mode: set\n"
"a/b/main.go:1.1,2.2 1 1\n"
"b/main.go:1.1,2.2 1 0\n"
)
assert go_percent(profile, "b/main.go", 1, 2) == 0.0
assert go_percent(profile, "a/b/main.go", 1, 2) == 100.0
def test_a_shorter_file_does_not_take_a_longer_files_coverage():
bundle = CoverageBundle(
go_profile=parse_go_profile("mode: set\na/b/main.go:1.1,2.2 1 1\n")
)
bundle.bind_sources(["b/main.go", "a/b/main.go"])
short = function(path="b/main.go", start_line=1, end_line=2)
long = function(path="a/b/main.go", start_line=1, end_line=2)
assert bundle.percent_for(short) is None
assert bundle.percent_for(long) == 100.0
def test_a_short_report_key_matches_one_longer_source():
bundle = CoverageBundle(
go_profile=parse_go_profile("mode: set\nboard.go:4.1,6.2 2 1\n")
)
bundle.bind_sources(["cmd/board.go"])
assert bundle.percent_for(function(path="cmd/board.go")) == 100.0
bundle.bind_sources(["cmd/board.go", "other/board.go"])
assert bundle.percent_for(function(path="cmd/board.go")) is None
assert bundle.percent_for(function(path="other/board.go")) is None
def test_suffix_lookup_finds_a_report_keyed_by_a_longer_path():
bundle = CoverageBundle(lcov={"proj/src/demo/core.clj": {3: (1, 1), 4: (0, 1)}})
fn = function(
@@ -264,6 +310,35 @@ def test_python_function_without_a_name_is_skipped():
assert [fn.name for fn in python_functions(source, "src/app.py", "/proj")] == ["choose"]
def test_a_deep_expression_does_not_recurse_off_the_stack():
expr = " + ".join(["1"] * 1200)
python = python_functions(f"def huge():\n return {expr}\n", "huge.py", None)
assert [fn.name for fn in python] == ["huge"]
java = java_functions(
f"class Huge {{\n int huge() {{ return {expr}; }}\n}}\n",
"Huge.java",
)
assert [fn.name for fn in java] == ["huge"]
def test_analyze_does_not_give_a_short_path_the_longer_files_coverage(tmp_path):
body = "package demo\nfunc Place() int { return 1 }\n"
short = tmp_path / "b" / "main.go"
long = tmp_path / "a" / "b" / "main.go"
short.parent.mkdir(parents=True)
long.parent.mkdir(parents=True)
short.write_text(body, encoding="utf-8")
long.write_text(body, encoding="utf-8")
bundle = CoverageBundle(
go_profile=parse_go_profile("mode: set\na/b/main.go:1.1,2.2 1 1\n")
)
entries = analyze_files([short, long], tmp_path, bundle)
assert [(entry.name, entry.path, entry.coverage) for entry in entries] == [
("Place", "b/main.go", 0.0),
("Place", "a/b/main.go", 100.0),
]
def test_rust_relative_path_and_a_trait_method(tmp_path):
(tmp_path / "Cargo.toml").write_text(
'[package]\nname = "demo"\nversion = "0.1.0"\n',
tests/test_runners.py
+140 -29
View File
@@ -1,8 +1,11 @@
import json
from pathlib import Path
import pytest
from crapper.discover import is_test_file
from crapper.runners import (
PackageJsonError,
_clean_clojure,
_clean_dir,
_coverage_report,
@@ -29,6 +32,12 @@ from crapper.runners import (
)
def _as_text(command) -> str:
if isinstance(command, list):
return " ".join(command)
return command
def _write(root: Path, relative: str, source: str) -> Path:
path = root / relative
path.parent.mkdir(parents=True, exist_ok=True)
@@ -48,12 +57,13 @@ def test_vitest_uses_its_own_coverage_instead_of_c8(tmp_path):
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
text = _as_text(command)
assert "c8" not in text
assert "vitest run --coverage" in text
assert "--coverage.reporter=lcov" in text
assert f"--coverage.reportsDirectory={report_dir}" in text
assert "--coverage.include=src/book.ts" in text
assert "book.test.ts" not in text
def test_existing_coverage_script_is_left_alone(tmp_path):
@@ -63,14 +73,23 @@ def test_existing_coverage_script_is_left_alone(tmp_path):
json.dumps({"scripts": {"coverage": "vitest run --coverage", "test": "vitest run"}}),
)
command = typescript_command(tmp_path, [], tmp_path / "coverage")
assert command == "npm run 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"
assert command == [
"npx",
"--yes",
"c8",
"--reporter=lcov",
"--reports-dir",
str(report_dir),
"npm",
"test",
]
def test_typescript_package_is_found_from_a_source_file(tmp_path):
@@ -112,10 +131,25 @@ def test_python_project_uses_pytest_and_coverage_lcov(tmp_path):
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}"
assert run == [
"python3",
"-m",
"coverage",
"run",
f"--data-file={data}",
"--source=src",
"-m",
"pytest",
]
assert lcov == [
"python3",
"-m",
"coverage",
"lcov",
f"--data-file={data}",
"-o",
str(report),
]
def test_python_package_without_pytest_uses_unittest(tmp_path):
@@ -123,20 +157,21 @@ def test_python_package_without_pytest_uses_unittest(tmp_path):
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 _as_text(run).endswith("-m unittest discover -s .")
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}"
assert command == ["cargo", "llvm-cov", "--lcov", "--output-path", str(report)]
def test_vitest_version_and_provider_install(tmp_path, monkeypatch):
def test_vitest_version_and_provider_install(tmp_path, monkeypatch, capsys):
assert _vitest_version(tmp_path) is None
_write(tmp_path, "node_modules/vitest/package.json", "{not json")
assert _vitest_version(tmp_path) is None
assert "package.json" in capsys.readouterr().err
_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"
@@ -176,7 +211,7 @@ def test_rust_kind_installs_llvm_cov_when_it_is_missing(monkeypatch):
return None
def shell(command, cwd):
if "cargo install" in command:
if "cargo install" in _as_text(command):
installed["llvm"] = True
return 0
@@ -189,18 +224,19 @@ def test_rust_kind_installs_llvm_cov_when_it_is_missing(monkeypatch):
def test_coverage_commands_are_issued_per_language(tmp_path, monkeypatch):
commands: list[str] = []
commands: list[str | list[str]] = []
def shell(command, cwd):
commands.append(command)
if "--data-file=" in command:
data = Path(command.split("--data-file=", 1)[1].split()[0])
text = _as_text(command)
if "--data-file=" in text:
data = Path(text.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 "):
if text.startswith("clj") or text.startswith("mvn") or text.startswith("go "):
return 1
if "vitest" in command or command.startswith("cargo "):
if "vitest" in text or text.startswith("cargo "):
return 1
return 0
@@ -236,7 +272,7 @@ def test_coverage_commands_are_issued_per_language(tmp_path, monkeypatch):
tmp_path / "src/lib.rs",
]
run_coverage(tmp_path, files, None)
text = "\n".join(commands)
text = "\n".join(_as_text(command) for command in 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
@@ -271,13 +307,22 @@ def test_run_shell_returns_the_child_status_and_reports_a_failed_start(tmp_path,
def test_clean_removes_a_stale_directory_and_ignores_a_missing_one(tmp_path):
stale = tmp_path / "target" / "coverage" / "old"
coverage = tmp_path / "target" / "coverage"
stale = coverage / "old"
stale.mkdir(parents=True)
(stale / "lcov.info").write_text("x", encoding="utf-8")
kept = tmp_path / "target" / "coverage" / "python"
(stale / "index.html").write_text("x", encoding="utf-8")
(coverage / "notes.txt").write_text("keep", encoding="utf-8")
nested = coverage / "notes"
nested.mkdir()
(nested / "keep.txt").write_text("x", encoding="utf-8")
kept = coverage / "python"
kept.mkdir()
(kept / "index.html").write_text("py", encoding="utf-8")
_clean_clojure(tmp_path)
assert not stale.exists()
assert (coverage / "notes.txt").is_file()
assert (nested / "keep.txt").is_file()
assert (kept / "index.html").is_file()
assert kept.is_dir()
_clean_dir(tmp_path / "target" / "coverage")
assert not (tmp_path / "target" / "coverage").exists()
@@ -288,9 +333,11 @@ def test_clean_removes_a_stale_directory_and_ignores_a_missing_one(tmp_path):
def test_package_json_without_a_test_script_has_no_coverage_command(tmp_path):
assert typescript_command(tmp_path, [], tmp_path / "cov") is None
_write(tmp_path, "package.json", "{")
assert typescript_command(tmp_path, [], tmp_path / "cov") is None
with pytest.raises(PackageJsonError, match="package.json"):
typescript_command(tmp_path, [], tmp_path / "cov")
_write(tmp_path, "package.json", "[]")
assert typescript_command(tmp_path, [], tmp_path / "cov") is None
with pytest.raises(PackageJsonError, match="expected a JSON object"):
typescript_command(tmp_path, [], tmp_path / "cov")
_write(tmp_path, "package.json", '{"scripts": ["nope"]}')
assert typescript_command(tmp_path, [], tmp_path / "cov") is None
@@ -344,13 +391,13 @@ def test_missing_python_module_is_installed(tmp_path, monkeypatch):
def shell(command, _cwd):
commands.append(command)
if command.startswith("python3 -c"):
if _as_text(command).startswith("python3 -c"):
return 1
return 0
monkeypatch.setattr("crapper.runners.run_shell", shell)
assert _ensure_python_module("python3", tmp_path, "coverage") is True
assert any("pip install" in command for command in commands)
assert any("pip install" in _as_text(command) for command in commands)
commands.clear()
def already(_command, _cwd):
@@ -431,3 +478,67 @@ def test_python_coverage_module_missing_is_reported(tmp_path, monkeypatch, capsy
source = _write(tmp_path, "src/app.py", "def run():\n return 1\n")
run_coverage(tmp_path, [source], None)
assert "coverage is missing" in capsys.readouterr().err
def test_an_argument_vector_is_not_a_shell(tmp_path, monkeypatch):
seen = []
def fake(command, cwd, shell=False):
seen.append((command, shell))
class Done:
returncode = 0
return Done()
monkeypatch.setattr("crapper.runners.subprocess.run", fake)
assert run_shell("exit 3", tmp_path) == 0
marker = tmp_path / "pwned"
assert run_shell(["true", f"a;touch {marker}"], tmp_path) == 0
assert seen[0] == ("exit 3", True)
assert seen[1][1] is False
assert seen[1][0] == ["true", f"a;touch {marker}"]
def test_an_argument_vector_cannot_run_a_second_command(tmp_path):
marker = tmp_path / "pwned"
assert run_shell(["true", f"a;touch {marker}"], tmp_path) == 0
assert not marker.exists()
def test_go_coverprofile_stays_one_argument(tmp_path, monkeypatch):
root = tmp_path / "my module"
root.mkdir()
_write(root, "go.mod", "module example.com/demo\n")
source = _write(root, "board.go", "package demo\nfunc Place() int { return 1 }\n")
seen = []
monkeypatch.setattr("crapper.runners.run_shell", lambda command, cwd: seen.append(command) or 0)
run_coverage(root, [source], None)
command = seen[0]
assert command[:3] == ["go", "test", "./..."]
assert command[3].startswith("-coverprofile=")
assert " " in command[3]
assert len(command) == 4
def test_invalid_package_json_is_not_a_missing_test_script(tmp_path, monkeypatch, capsys):
monkeypatch.setattr("crapper.runners.run_shell", lambda *_args: 0)
_write(tmp_path, "package.json", "{ not json")
source = _write(tmp_path, "src/app.ts", "export const n = 1\n")
run_coverage(tmp_path, [source], None)
err = capsys.readouterr().err
assert "package.json" in err
assert "No package.json test script" not in err
def test_vitest_install_restores_manifests_when_install_changes_them(tmp_path, monkeypatch):
package_json = tmp_path / "package.json"
package_json.write_bytes(b'{"name":"demo"}\n')
def shell(command, cwd):
package_json.write_bytes(b'{"name":"changed"}\n')
return 1
monkeypatch.setattr("crapper.runners.run_shell", shell)
assert _ensure_vitest_coverage(tmp_path) is False
assert package_json.read_bytes() == b'{"name":"demo"}\n'
tests/test_rust.py
+7 -1
View File
@@ -1,4 +1,6 @@
from crapper.languages.rust import functions_in_source
from pathlib import Path
from crapper.languages.rust import _file_modules, functions_in_source
SOURCE = """
pub fn choose(x: i32) -> i32 {
@@ -80,6 +82,10 @@ def test_module_paths_outside_the_usual_layout(tmp_path):
assert bare[0].name == "bare"
def test_a_file_outside_the_crate_root_uses_the_crate_name():
assert _file_modules("/tmp/loose.rs", "demo", Path("/projects/demo")) == ["demo"]
def test_nested_module_path(tmp_path):
(tmp_path / "Cargo.toml").write_text(
'[package]\nname = "demo"\nversion = "0.1.0"\n',
uml
+12 -1
View File
@@ -5,7 +5,18 @@ set -euo pipefail
ROOT="$(cd "$(dirname "$0")" && pwd)"
cd "$ROOT"
DEPS='{:deps {uml-viewer/uml-viewer {:local/root "/Users/unclebob/projects/uml-viewer"} quil/quil {:mvn/version "4.3.1563"}}}'
if [[ -n "${UML_VIEWER_ROOT:-}" ]]; then
VIEWER="$UML_VIEWER_ROOT"
else
VIEWER="$ROOT/../uml-viewer"
fi
if [[ ! -d "$VIEWER" ]]; then
echo "uml-viewer checkout not found at $VIEWER" >&2
echo "Set UML_VIEWER_ROOT to the checkout." >&2
exit 1
fi
VIEWER="$(cd "$VIEWER" && pwd)"
DEPS="{:deps {uml-viewer/uml-viewer {:local/root \"$VIEWER\"} quil/quil {:mvn/version \"4.3.1563\"}}}"
edn_path() {
local name base f