#!/usr/bin/env python3 """Extract requirement traces from source and report coverage against a register. Ported from JellyTau's ``scripts/extract-traces.ts`` and written in stdlib Python so no repo needs a bun/node toolchain to check its source comments. **One implementation, parameterised.** This tool is shared by every JRay component repo — C++/Python extraction, C# plugin, Rust server — via the ``jray-project`` submodule. Nothing about a single repo is baked in: the requirement ID prefixes, the source file suffixes, the directories to scan, the register path and the system-spec path are all configuration. A second copy for "the other language" is how two implementations start drifting apart, so there is exactly one. Configuration comes from ``traceability.toml`` at the component repo root, from CLI flags, or both (flags win). ``--print-example-config`` emits the full annotated schema; the three required keys are:: requirement_types = ["AR", "DP", "IR", "GR", "VR"] languages = ["cpp", "python"] source_roots = ["src", "tests", "scripts"] Tag format (see the system CLAUDE.md and SPEC.md section 6). A pipe separates requirement *types*, a comma separates IDs within a type:: /// TRACES: AR-nnn, AR-mmm | SR-nnn struct TrackRegistry { ... }; Deliberate invariant exceptions carry their own tag and are reported separately — never silently folded into coverage:: // EXCEPTION: AR-nnn distributional check, not a match decision Usage:: python3 extract_traces.py --format coverage python3 extract_traces.py --format json > report.json python3 extract_traces.py --config path/to/traceability.toml The CI gate is ``traceability-gate.sh``, which wraps this. Three rules the gate exists to enforce. The first two are inherited from JellyTau's gate repair, both learned the hard way (``JellyTau/docs/specs/traceability-gate-repair.md``): 1. Coverage denominators are read out of the register at run time. Never hardcode them. JellyTau's gate divided by frozen literals while the register grew to 211 requirements; it reported 158% coverage, so its 50% threshold could never trip. A gate that cannot fail is worse than no gate, because it is trusted. 2. Coverage above 100% is a hard failure, not a pass. It means the computation is broken, and it is the signal that would have caught (1) immediately. 3. A misconfigured run refuses to report at all. Parsing zero requirements or scanning zero files prints a failure, not a plausible-looking 0% — the same family of error as (1), and the one a shared, parameterised tool makes easy to hit. The GPU-less-CI rule is configuration rather than code: requirements whose verification tiers all fall outside ``ci_executable_tiers`` are reported as *tagged but unexecuted* and excluded from the covered numerator. Counting a test that never runs is the same failure mode as the 158% bug. """ from __future__ import annotations import argparse import json import re import sys from dataclasses import dataclass, field, replace from datetime import datetime, timezone from pathlib import Path from typing import (Dict, FrozenSet, Iterable, List, Optional, Sequence, Set, Tuple) # -------------------------------------------------------------------------- # Defaults. Everything here is overridable; nothing here names a single repo. # -------------------------------------------------------------------------- #: Test identifiers. A separate taxonomy: a test ID is evidence *for* a #: requirement, not a requirement, so it never enters the coverage fraction. #: House-wide (SPEC.md section 6), hence a default rather than an argument. DEFAULT_TEST_TYPES: Tuple[str, ...] = ("UT", "IT") #: Defined in the system-level SPEC.md that ``jray-project`` owns. Recognised #: in tags everywhere, never counted in any component's fraction, and #: orphan-checked only when ``system_spec`` points at that file. DEFAULT_EXTERNAL_TYPES: Tuple[str, ...] = ("PR", "SR") #: Tiers a CI host can actually run. The extraction pipeline's CI is a GPU-less #: Intel N100, so its T4 is excluded; jRay numbers its live-only tier T4 for #: exactly this reason rather than renumbering the shared constant. DEFAULT_CI_EXECUTABLE_TIERS: FrozenSet[str] = frozenset({"T1", "T2", "T3", "static"}) DEFAULT_EXCLUDE_DIRS: FrozenSet[str] = frozenset({ ".git", "__pycache__", "external", "vendor", "build", "site", "dist", "node_modules", "target", "bin", "obj", "trt_cache", "ort_cache", ".venv", "venv", ".mypy_cache", ".pytest_cache", }) DEFAULT_REQUIREMENTS_PATH = "docs/requirements.md" DEFAULT_MATRIX_PATH = "docs/traceability.md" DEFAULT_JSON_PATH = "traces-report.json" CONFIG_FILENAME = "traceability.toml" #: Named suffix groups, so a repo declares "Rust" rather than remembering to #: spell every extension. Additive with an explicit ``source_suffixes`` list. LANGUAGE_SUFFIXES: Dict[str, FrozenSet[str]] = { "cpp": frozenset({".cpp", ".cc", ".cxx", ".hpp", ".hxx", ".h", ".cu", ".cuh"}), "python": frozenset({".py", ".pyi"}), "rust": frozenset({".rs"}), "csharp": frozenset({".cs"}), "javascript": frozenset({".js", ".mjs", ".cjs", ".jsx"}), "typescript": frozenset({".ts", ".tsx"}), "svelte": frozenset({".svelte"}), "go": frozenset({".go"}), "java": frozenset({".java"}), } # -------------------------------------------------------------------------- # Configuration # -------------------------------------------------------------------------- class ConfigError(Exception): """A configuration mistake, reported rather than papered over.""" @dataclass(frozen=True) class Config: """Everything that differs between component repos. The three required fields are exactly the three things that were hardcoded before this tool was shared: which ID prefixes the repo's register defines, which files count as source, and where those files live. A repo that gets any of them wrong parses zero requirements or scans zero files — and the gate refuses to report rather than printing a misleading 0%. """ requirement_types: Tuple[str, ...] source_suffixes: FrozenSet[str] source_roots: Tuple[str, ...] root: Path = Path(".") test_types: Tuple[str, ...] = DEFAULT_TEST_TYPES external_types: Tuple[str, ...] = DEFAULT_EXTERNAL_TYPES exclude_dirs: FrozenSet[str] = DEFAULT_EXCLUDE_DIRS ci_executable_tiers: FrozenSet[str] = DEFAULT_CI_EXECUTABLE_TIERS requirements_path: str = DEFAULT_REQUIREMENTS_PATH system_spec_path: Optional[str] = None matrix_path: Optional[str] = DEFAULT_MATRIX_PATH json_path: Optional[str] = DEFAULT_JSON_PATH min_coverage: float = 0.0 allow_orphans: bool = False source: str = "defaults" @property def known_types(self) -> Tuple[str, ...]: return (tuple(self.requirement_types) + tuple(self.test_types) + tuple(self.external_types)) def resolve(self, relative: Optional[str]) -> Optional[Path]: """A configured path, made absolute against the repo root.""" if relative is None: return None path = Path(relative) return path if path.is_absolute() else (self.root / path) def validate(self) -> None: problems: List[str] = [] if not self.requirement_types: problems.append( "requirement_types is empty - the register defines IDs with " "some prefix (AR/DP/IR/GR/VR in extraction, JR in jRay, UR/DR " "in the server) and the tool cannot guess it") for name, values in (("requirement_types", self.requirement_types), ("test_types", self.test_types), ("external_types", self.external_types)): for value in values: if not re.fullmatch(r"[A-Z]{2}", value): problems.append( f"{name}: {value!r} is not a two-letter uppercase " "prefix; IDs are PREFIX-nnn") overlaps = ((set(self.requirement_types) & set(self.test_types)) | (set(self.requirement_types) & set(self.external_types))) if overlaps: problems.append( f"prefixes {sorted(overlaps)} are declared both as " "requirement_types and as another taxonomy; a prefix counted " "in the fraction cannot also be excluded from it") if not self.source_suffixes: problems.append( 'no source suffixes - set `languages` (e.g. ["rust"]) or ' '`source_suffixes` (e.g. [".rs"])') for suffix in sorted(self.source_suffixes): if not suffix.startswith("."): problems.append(f"source suffix {suffix!r} must start with a dot") if not self.source_roots: problems.append( "no source roots - set `source_roots` to the directories " "holding this repo's code") if not 0.0 <= self.min_coverage <= 100.0: problems.append( f"min_coverage {self.min_coverage} is not a percentage") if problems: raise ConfigError("; ".join(problems)) CONFIG_KEYS = { "requirement_types", "test_types", "external_types", "languages", "source_suffixes", "source_roots", "exclude_dirs", "ci_executable_tiers", "requirements", "system_spec", "matrix", "json_report", "min_coverage", "allow_orphans", } def example_config() -> str: """The full schema, emitted by ``--print-example-config``.""" return """\ # traceability.toml - per-repo configuration for the shared trace extractor. # Lives at the component repo root; its directory is taken as the repo root. # REQUIRED. The ID prefixes this repo's register defines. These, and only # these, form the coverage fraction. # scene-actor-extraction: ["AR", "DP", "IR", "GR", "VR"] # jRay: ["JR"] # JRay-public-server: ["UR", "DR"] requirement_types = ["AR", "DP", "IR", "GR", "VR"] # REQUIRED (at least one of the two). `languages` names suffix groups; # `source_suffixes` adds anything else. Known groups: cpp, python, rust, # csharp, javascript, typescript, svelte, go, java. languages = ["cpp", "python"] # source_suffixes = [".inl"] # REQUIRED. Directories to scan, relative to the repo root. source_roots = ["src", "tests", "scripts"] # Optional; defaults shown. # test_types = ["UT", "IT"] # evidence for requirements, not counted # external_types = ["PR", "SR"] # owned by the system spec, not counted # requirements = "docs/requirements.md" # matrix = "docs/traceability.md" # json_report = "traces-report.json" # min_coverage = 0.0 # allow_orphans = false # exclude_dirs = ["external", "build", "target", "node_modules"] # Which verification tiers this repo's CI host can actually execute. A # requirement whose tiers all fall outside this set is reported as tagged but # unexecuted, and is never counted as covered. # ci_executable_tiers = ["T1", "T2", "T3", "static"] # The system spec defining PR/SR, vendored as a submodule in each component so # the check is runnable in CI. Omit to skip PR/SR orphan checking. # system_spec = "scripts/vendor/jray-project/SPEC.md" """ def find_config(start: Path) -> Optional[Path]: """Nearest ``traceability.toml`` at or above ``start``. The file's directory defines the repo root, which is what lets the tool be run from a subdirectory without silently scanning the wrong tree. """ current = start.resolve() for candidate in (current, *current.parents): path = candidate / CONFIG_FILENAME if path.is_file(): return path return None def _load_toml(path: Path) -> dict: try: import tomllib except ModuleNotFoundError as exc: # pragma: no cover - version dependent raise ConfigError( f"reading {path} needs Python 3.11+ for tomllib (this is " f"{sys.version_info.major}.{sys.version_info.minor}). Either " "upgrade, or pass --requirement-type/--language/--source-root on " "the command line instead of using a config file.") from exc with path.open("rb") as handle: return tomllib.load(handle) def config_from_dict(data: dict, root: Path, source: str) -> Config: unknown = sorted(set(data) - CONFIG_KEYS) if unknown: # A typo'd key would otherwise leave a required field empty, and the # user would be debugging "zero requirements" instead of a spelling. raise ConfigError( f"unknown key(s) in {source}: {', '.join(unknown)}. " f"Known keys: {', '.join(sorted(CONFIG_KEYS))}") suffixes: Set[str] = set() for language in data.get("languages", []): if language not in LANGUAGE_SUFFIXES: raise ConfigError( f"unknown language {language!r} in {source}. Known: " f"{', '.join(sorted(LANGUAGE_SUFFIXES))}") suffixes |= LANGUAGE_SUFFIXES[language] suffixes |= set(data.get("source_suffixes", [])) return Config( requirement_types=tuple(data.get("requirement_types", ())), source_suffixes=frozenset(suffixes), source_roots=tuple(data.get("source_roots", ())), root=root, test_types=tuple(data.get("test_types", DEFAULT_TEST_TYPES)), external_types=tuple(data.get("external_types", DEFAULT_EXTERNAL_TYPES)), exclude_dirs=frozenset(data.get("exclude_dirs", DEFAULT_EXCLUDE_DIRS)), ci_executable_tiers=frozenset( data.get("ci_executable_tiers", DEFAULT_CI_EXECUTABLE_TIERS)), requirements_path=data.get("requirements", DEFAULT_REQUIREMENTS_PATH), system_spec_path=data.get("system_spec"), matrix_path=data.get("matrix", DEFAULT_MATRIX_PATH), json_path=data.get("json_report", DEFAULT_JSON_PATH), min_coverage=float(data.get("min_coverage", 0.0)), allow_orphans=bool(data.get("allow_orphans", False)), source=source, ) def load_config(args: argparse.Namespace, cwd: Optional[Path] = None) -> Config: """Merge config file and CLI flags, per field. Flags win.""" cwd = (cwd or Path.cwd()).resolve() config_path: Optional[Path] = args.config if config_path is None and not args.no_config: config_path = find_config(cwd) if config_path is not None and not Path(config_path).is_file(): raise ConfigError(f"config file not found: {config_path}") if config_path is not None: config_path = Path(config_path).resolve() config = config_from_dict(_load_toml(config_path), config_path.parent, str(config_path)) else: config = Config(requirement_types=(), source_suffixes=frozenset(), source_roots=(), root=cwd, source="command line only") if args.root is not None: config = replace(config, root=Path(args.root).resolve()) if args.requirement_type: config = replace(config, requirement_types=tuple(args.requirement_type)) if args.test_type: config = replace(config, test_types=tuple(args.test_type)) if args.external_type: config = replace(config, external_types=tuple(args.external_type)) # Languages and suffixes are additive on top of whatever the file declared, # so `--language rust` extends rather than silently replacing. extra: Set[str] = set() for language in args.language or []: if language not in LANGUAGE_SUFFIXES: raise ConfigError( f"unknown language {language!r}. Known: " f"{', '.join(sorted(LANGUAGE_SUFFIXES))}") extra |= LANGUAGE_SUFFIXES[language] extra |= set(args.source_suffix or []) if extra: config = replace(config, source_suffixes=config.source_suffixes | extra) if args.source_root: config = replace(config, source_roots=tuple(args.source_root)) if args.exclude_dir: config = replace(config, exclude_dirs=config.exclude_dirs | set(args.exclude_dir)) if args.ci_executable_tier: config = replace(config, ci_executable_tiers=frozenset(args.ci_executable_tier)) if args.requirements is not None: config = replace(config, requirements_path=str(args.requirements)) if args.system_spec is not None: config = replace(config, system_spec_path=str(args.system_spec)) if args.markdown_out is not None: config = replace(config, matrix_path=str(args.markdown_out)) if args.json_out is not None: config = replace(config, json_path=str(args.json_out)) if args.no_write: config = replace(config, matrix_path=None, json_path=None) if args.min_coverage is not None: config = replace(config, min_coverage=float(args.min_coverage)) if args.allow_orphans: config = replace(config, allow_orphans=True) config.validate() return config # -------------------------------------------------------------------------- # Tag parsing # -------------------------------------------------------------------------- TRACES_RE = re.compile(r"TRACES:[ \t]*([^\n]*)") EXCEPTION_RE = re.compile(r"EXCEPTION:[ \t]*([A-Z]{2}-\d{3})[ \t]*([^\n]*)") REQ_ID_RE = re.compile(r"^([A-Z]{2})-(\d{3})$") LEADING_ID_RE = re.compile(r"^([A-Z]{2}-\d{3})\b(.*)$") #: Something that was *trying* to be a requirement ID. Keeps the malformed-tag #: diagnostic quiet when the word "TRACES" merely appears in prose or in this #: tool's own source, while still catching `AR-12`. ID_ATTEMPT_RE = re.compile(r"[A-Za-z]{2}-\d") #: Comment terminators that can trail a tag on the same line. COMMENT_TERMINATORS = ("*/", "-->", '"""', "'''") DECL_PATTERNS = [ re.compile(r"^\s*(?:async\s+)?def\s+\w+"), # Python re.compile(r"^\s*(?:pub(?:\([^)]*\))?\s+)?(?:async\s+)?" r"(?:unsafe\s+)?(?:extern\s+\"[^\"]*\"\s+)?fn\s+\w+"), # Rust re.compile(r"^\s*(?:pub(?:\([^)]*\))?\s+)?" r"(?:impl|trait|mod|type)\b"), # Rust re.compile(r"^\s*(?:template\s*<[^;]*>\s*)?" r"(?:struct|class|enum(?:\s+class)?|union|namespace|" r"interface|record)\s+\w+"), # C++/C# re.compile(r"^\s*(?:public|private|protected|internal|static|inline|" r"constexpr|virtual|explicit|friend|override|abstract|sealed|" r"partial|extern)\b"), re.compile(r"^\s*[A-Za-z_][\w:<>,\s\*&\.\[\]]*\s+[A-Za-z_~][\w:]*\s*\([^)]*\)"), ] #: Attribute/annotation lines: C# ``[HttpGet("x")]``, Rust ``#[derive(...)]``, #: Java ``@Override``. A tag above a decorated declaration must attribute to #: the declaration, not to its decoration. ATTRIBUTE_RE = re.compile(r"^\s*(?:\[[^\]]*\]|#\s*\[|@\w+)") @dataclass class TraceEntry: file: str line: int context: str requirements: List[str] def to_dict(self) -> dict: return { "file": self.file, "line": self.line, "context": self.context, "requirements": list(self.requirements), } @dataclass class ExceptionEntry: file: str line: int requirement: str reason: str context: str def to_dict(self) -> dict: return { "file": self.file, "line": self.line, "requirement": self.requirement, "reason": self.reason, "context": self.context, } @dataclass class Diagnostics: """Things wrong with the tags themselves, kept visible rather than dropped.""" malformed_tags: List[dict] = field(default_factory=list) mixed_type_groups: List[dict] = field(default_factory=list) unknown_id_types: List[dict] = field(default_factory=list) exceptions_without_reason: List[dict] = field(default_factory=list) def to_dict(self) -> dict: return { "malformedTags": self.malformed_tags, "mixedTypeGroups": self.mixed_type_groups, "unknownIdTypes": self.unknown_id_types, "exceptionsWithoutReason": self.exceptions_without_reason, } @property def empty(self) -> bool: return not (self.malformed_tags or self.mixed_type_groups or self.unknown_id_types or self.exceptions_without_reason) def parse_traces_tag(value: str) -> Tuple[List[List[str]], List[str]]: """Parse the text after ``TRACES:`` into per-type groups plus junk. Returns ``(groups, junk)``. ``groups`` is one list of IDs per pipe-separated segment, preserving the type grouping the format promises. ``junk`` holds anything that was not a bare ID, so a tag like ``AR-001 - see also AR-999`` yields ``AR-001`` and reports the rest rather than quietly harvesting an ID out of prose. """ text = value for term in COMMENT_TERMINATORS: idx = text.find(term) if idx != -1: text = text[:idx] text = text.strip() if not text: return [], [] groups: List[List[str]] = [] junk: List[str] = [] for raw_group in text.split("|"): ids: List[str] = [] for raw_item in raw_group.split(","): item = raw_item.strip() if not item: continue if REQ_ID_RE.match(item): ids.append(item) continue lead = LEADING_ID_RE.match(item) if lead: ids.append(lead.group(1)) remainder = lead.group(2).strip() if remainder: junk.append(remainder) else: junk.append(item) if ids: groups.append(ids) return groups, junk def find_context(lines: Sequence[str], index: int, window: int = 12) -> str: """Best-effort name of the declaration a tag belongs to. Searches both directions, because languages put the tag on opposite sides of the thing it describes: a C++/C#/Rust ``///`` tag sits *above* the declaration, while a Python tag usually sits *inside* the docstring, below the ``def``. Whichever declaration is nearer wins. Attribute lines (``[HttpGet]``, ``#[derive]``) are skipped so a tag above a decorated method attributes to the method rather than to its decoration. """ def matches(line: str) -> bool: if ATTRIBUTE_RE.match(line): return False return any(p.match(line) for p in DECL_PATTERNS) down: Optional[Tuple[int, str]] = None for offset in range(1, window + 1): i = index + offset if i >= len(lines): break if matches(lines[i]): down = (offset, lines[i]) break up: Optional[Tuple[int, str]] = None for offset in range(1, window + 1): i = index - offset if i < 0: break if matches(lines[i]): up = (offset, lines[i]) break if down and up: best = down if down[0] <= up[0] else up else: best = down or up if best is None: return "Unknown" return best[1].strip().rstrip("{").strip()[:120] or "Unknown" def iter_source_files(config: Config) -> List[Path]: """Every source file under the configured roots, by configured suffix.""" found: List[Path] = [] for rel in config.source_roots: base = config.root / rel if not base.is_dir(): continue for path in sorted(base.rglob("*")): if not path.is_file(): continue if path.suffix not in config.source_suffixes: continue if any(part in config.exclude_dirs for part in path.parts): continue found.append(path) return found @dataclass class ScanResult: files: List[Path] traces: List[TraceEntry] exceptions: List[ExceptionEntry] diagnostics: Diagnostics def scan_files(files: Sequence[Path], config: Config) -> ScanResult: traces: List[TraceEntry] = [] exceptions: List[ExceptionEntry] = [] diags = Diagnostics() root = config.root known = set(config.known_types) for path in files: try: content = path.read_text(encoding="utf-8", errors="replace") except OSError: continue lines = content.split("\n") rel = str(path.relative_to(root)) if path.is_relative_to(root) else str(path) for index, line in enumerate(lines): exc = EXCEPTION_RE.search(line) if exc: reason = exc.group(2) for term in COMMENT_TERMINATORS: cut = reason.find(term) if cut != -1: reason = reason[:cut] reason = reason.strip() exceptions.append(ExceptionEntry( file=rel, line=index + 1, requirement=exc.group(1), reason=reason, context=find_context(lines, index))) if not reason: # An exception is only *agreed* if the reason is recorded; # an unexplained one is an undocumented defect. diags.exceptions_without_reason.append( {"file": rel, "line": index + 1, "requirement": exc.group(1)}) continue match = TRACES_RE.search(line) if not match: continue groups, junk = parse_traces_tag(match.group(1)) ids = [i for g in groups for i in g] if not ids: # No IDs at all. Only worth reporting if something in the text # was trying to be one - otherwise every sentence containing # the word would be flagged, and a diagnostic nobody can act on # is how real diagnostics get ignored. if junk and ID_ATTEMPT_RE.search(match.group(1)): diags.malformed_tags.append( {"file": rel, "line": index + 1, "text": match.group(1).strip()}) continue if junk: diags.malformed_tags.append( {"file": rel, "line": index + 1, "ignored": junk}) for group in groups: types = {i.split("-")[0] for i in group} if len(types) > 1: # The pipe is what separates types; a mixed group means the # tag does not say what it looks like it says. diags.mixed_type_groups.append( {"file": rel, "line": index + 1, "group": list(group)}) for req in ids: if req.split("-")[0] not in known: diags.unknown_id_types.append( {"file": rel, "line": index + 1, "id": req}) seen: Set[str] = set() unique = [i for i in ids if not (i in seen or seen.add(i))] traces.append(TraceEntry(file=rel, line=index + 1, context=find_context(lines, index), requirements=unique)) return ScanResult(files=list(files), traces=traces, exceptions=exceptions, diagnostics=diags) # -------------------------------------------------------------------------- # The register: requirements.md is the authoritative denominator # -------------------------------------------------------------------------- @dataclass class Requirement: id: str text: str = "" traces_to: str = "" priority: str = "" status: str = "" tiers: Set[str] = field(default_factory=set) @property def tier_known(self) -> bool: return bool(self.tiers) @dataclass class Register: #: The ID prefixes this register defines, from configuration. types: Tuple[str, ...] = () #: Verification tiers the consuming repo's CI host can execute. ci_tiers: FrozenSet[str] = DEFAULT_CI_EXECUTABLE_TIERS requirements: Dict[str, Requirement] = field(default_factory=dict) withdrawn: Dict[str, Requirement] = field(default_factory=dict) @property def ids(self) -> Set[str]: return set(self.requirements) @property def total(self) -> int: return len(self.requirements) @property def uses_tiers(self) -> bool: """Whether this register assigns verification tiers at all. A repo with no tier tables is not 'missing' tiers; it does not use the mechanism. Warning about every requirement there would be noise that trains people to ignore the warning that matters. """ return any(r.tier_known for r in self.requirements.values()) def is_ci_executable(self, req_id: str) -> bool: """True when at least one verification tier can run on the CI host. A requirement with no tier recorded is *unknown*, not unexecutable - it is reported so the register gets fixed, and it is not penalised here. """ req = self.requirements[req_id] if not req.tiers: return True return bool(req.tiers & self.ci_tiers) def count(self, req_type: str) -> int: return sum(1 for i in self.requirements if i.startswith(req_type + "-")) def ci_executable_ids(self) -> Set[str]: return {i for i in self.requirements if self.is_ci_executable(i)} def unexecutable_ids(self) -> Set[str]: return {i for i in self.requirements if not self.is_ci_executable(i)} def tier_unknown_ids(self) -> Set[str]: return {i for i, r in self.requirements.items() if not r.tier_known} def parentless_ids(self) -> Set[str]: """Requirements whose ``Traces to`` cell names nothing. SPEC.md section 6 asks the gate to report these: a requirement serving no stated goal is scope creep, and it is invisible unless something looks. A cell naming a section (`§4`) counts as a parent - the point is that *something* was recorded, not that it was an ID. """ out = set() for req_id, req in self.requirements.items(): cell = req.traces_to.replace("*", "").strip() if not cell or cell in {"-", "—", "n/a", "N/A", "TBD"}: out.add(req_id) return out def _row_cells(line: str) -> List[str]: r"""Split a markdown table row, honouring escaped ``\|`` inside cells.""" placeholder = "\x00" stripped = line.strip().replace("\\|", placeholder) cells = stripped.strip("|").split("|") return [c.strip().replace(placeholder, "|") for c in cells] def _is_separator(cells: Sequence[str]) -> bool: return bool(cells) and all(re.fullmatch(r":?-{3,}:?", c) for c in cells) def _iter_table_rows(markdown: str): """Yield ``(header_cells, row_cells)`` for every markdown table data row. A table's header is the row immediately preceding its ``|---|`` separator; anything before that separator is not data. Any non-table line ends the current table. """ header: Optional[List[str]] = None previous: Optional[List[str]] = None for line in markdown.split("\n"): stripped = line.strip() if not stripped.startswith("|"): header = None previous = None continue cells = _row_cells(stripped) if _is_separator(cells): header = previous continue if header is not None: yield header, cells previous = cells def _is_register_header(header: Sequence[str]) -> bool: """A definition table: first column ``ID``, and a ``Requirement`` column. Deliberately narrow. The per-requirement verification plan is also keyed on ``ID`` but has no ``Requirement`` column, and the tier summary table's first column holds comma lists and ranges - neither defines requirements, and counting their rows would inflate the denominator. Prose mentions and ``Traces to`` references are excluded for the same reason: a naive scan for ``AR-\\d{3}`` over the whole file counts every reference as a definition. """ if not header: return False lowered = [c.lower() for c in header] return lowered[0] == "id" and "requirement" in lowered def _is_tier_header(header: Sequence[str]) -> bool: """Either of the two table shapes that assign verification tiers.""" if len(header) < 2: return False lowered = [c.lower() for c in header] return lowered[0] in ("id", "requirement") and lowered[1] == "tier" def _column(header: Sequence[str], name: str, cells: Sequence[str]) -> str: lowered = [c.lower() for c in header] if name in lowered: index = lowered.index(name) if index < len(cells): return cells[index] return "" def parse_tiers(cell: str) -> Set[str]: """Read a tier cell such as ``T1 + T4``, ``**T2**``, or ``Out of CI``.""" text = cell.replace("*", "").strip().lower() tiers = {"T" + m.group(1) for m in re.finditer(r"\bt(\d)\b", text)} if "out of ci" in text or "not in ci" in text: tiers.add("out-of-ci") if "manual" in text: tiers.add("manual") if "static" in text: tiers.add("static") return tiers def expand_id_spec(spec: str, defined: Set[str]) -> List[str]: """Expand the ID cell of a tier table into concrete, defined IDs. Handles every shape the registers actually use: ``AR-002``, ``AR-001, AR-005, AR-006``, ``AR-007 ... AR-017`` (with the unicode ellipsis), ``AR-009/010``, and ``DP-*``. Expansion is intersected with the defined set, so a range can never invent a requirement that does not exist. """ text = spec.replace("*", "").strip() # `DP-*` survives the bold-strip above as a bare `DP-`; both mean "every # requirement of this type". if re.fullmatch(r"[A-Z]{2}-", text): prefix = text[:2] return sorted(i for i in defined if i.startswith(prefix + "-")) out: List[str] = [] for part in text.split(","): part = part.strip() if not part: continue rng = re.fullmatch(r"([A-Z]{2})-(\d{3})\s*(?:…|\.\.\.)\s*([A-Z]{2})-(\d{3})", part) if rng and rng.group(1) == rng.group(3): prefix = rng.group(1) low, high = int(rng.group(2)), int(rng.group(4)) out.extend(sorted( i for i in defined if i.startswith(prefix + "-") and low <= int(i.split("-")[1]) <= high)) continue slash = re.fullmatch(r"([A-Z]{2})-(\d{3})/(\d{3})", part) if slash: prefix = slash.group(1) out.extend(f"{prefix}-{n}" for n in (slash.group(2), slash.group(3))) continue if REQ_ID_RE.match(part): out.append(part) return [i for i in out if i in defined] def parse_register(markdown: str, config: Config) -> Register: """Build the register from the repo's requirements.md. Two passes: definitions first, because tier rows use ranges and wildcards that can only be expanded against a known ID set. """ register = Register(types=tuple(config.requirement_types), ci_tiers=frozenset(config.ci_executable_tiers)) definable = set(config.requirement_types) | set(config.test_types) for header, cells in _iter_table_rows(markdown): if not _is_register_header(header) or not cells: continue first = cells[0].replace("*", "").strip() if not REQ_ID_RE.match(first): continue if first.split("-")[0] not in definable: continue req = Requirement( id=first, text=_column(header, "requirement", cells), traces_to=_column(header, "traces to", cells), priority=_column(header, "priority", cells), status=_column(header, "status", cells), ) if req.status.replace("*", "").strip().lower() == "withdrawn": # Permanently retired. Counting it in the denominator would depress # coverage forever for something that no longer needs implementing. register.withdrawn.setdefault(first, req) register.requirements.pop(first, None) continue if first in register.withdrawn: continue # An ID may legitimately appear in more than one definition table; the # first occurrence wins and duplicates never inflate the count. register.requirements.setdefault(first, req) defined = register.ids for header, cells in _iter_table_rows(markdown): if not _is_tier_header(header) or len(cells) < 2: continue tiers = parse_tiers(cells[1]) if not tiers: continue for req_id in expand_id_spec(cells[0], defined): register.requirements[req_id].tiers |= tiers return register def read_register(config: Config) -> Register: path = config.resolve(config.requirements_path) assert path is not None return parse_register(path.read_text(encoding="utf-8"), config) # -------------------------------------------------------------------------- # Coverage # -------------------------------------------------------------------------- @dataclass class Coverage: covered: List[str] total: int percent: float orphaned: List[str] unexecuted: List[str] ci_executable_total: int ci_percent: float tier_unknown: List[str] def to_dict(self) -> dict: return { "covered": len(self.covered), "coveredIds": self.covered, "total": self.total, "percent": self.percent, "orphaned": self.orphaned, "unexecuted": self.unexecuted, "ciExecutableTotal": self.ci_executable_total, "ciPercent": self.ci_percent, "tierUnknown": self.tier_unknown, } def compute_coverage(traced_ids: Iterable[str], register: Register) -> Coverage: """Coverage is ``|traced and defined and CI-executable| / |defined|``. Three exclusions, each of which is a way the number could otherwise lie: * Using the raw traced count as the numerator is what lets a ratio exceed 100% - a tag naming a deleted or mistyped requirement would count as covered. Those land in ``orphaned`` instead, so they get fixed rather than silently counted or silently dropped. * A requirement whose verification tiers all fall outside what the CI host can execute cannot be verified there at all. It is reported in ``unexecuted`` and is not covered: treating "has a test that never runs" as passing reports success the gate cannot substantiate. * Test IDs and system-level IDs are separate taxonomies with their own registers, so they neither count nor orphan here. """ traced = {i for i in traced_ids if i.split("-")[0] in register.types} defined = register.ids orphaned = sorted(traced - defined) matched = traced & defined unexecuted = sorted(i for i in matched if not register.is_ci_executable(i)) covered = sorted(matched - set(unexecuted)) total = len(defined) ci_total = len(register.ci_executable_ids()) percent = round(100.0 * len(covered) / total, 1) if total else 0.0 ci_percent = round(100.0 * len(covered) / ci_total, 1) if ci_total else 0.0 return Coverage( covered=covered, total=total, percent=percent, orphaned=orphaned, unexecuted=unexecuted, ci_executable_total=ci_total, ci_percent=ci_percent, tier_unknown=sorted(register.tier_unknown_ids()), ) # -------------------------------------------------------------------------- # Report assembly # -------------------------------------------------------------------------- @dataclass class Report: timestamp: str config: Config register: Register scan: ScanResult coverage: Coverage by_type: Dict[str, List[str]] requirement_map: Dict[str, List[TraceEntry]] external_orphans: List[str] = field(default_factory=list) def build_report(config: Config, register: Register, scan: ScanResult, system_ids: Optional[Set[str]] = None) -> Report: requirement_map: Dict[str, List[TraceEntry]] = {} seen_by_type: Dict[str, Set[str]] = {t: set() for t in config.known_types} seen_by_type["OTHER"] = set() for entry in scan.traces: for req in entry.requirements: requirement_map.setdefault(req, []).append(entry) prefix = req.split("-")[0] seen_by_type[prefix if prefix in seen_by_type else "OTHER"].add(req) coverage = compute_coverage(requirement_map.keys(), register) external_orphans: List[str] = [] if system_ids is not None: external_orphans = sorted( i for i in requirement_map if i.split("-")[0] in config.external_types and i not in system_ids) return Report( timestamp=datetime.now(timezone.utc).isoformat(timespec="seconds"), config=config, register=register, scan=scan, coverage=coverage, by_type={k: sorted(v) for k, v in seen_by_type.items()}, requirement_map=requirement_map, external_orphans=external_orphans, ) def parse_system_spec(markdown: str, types: Sequence[str] = DEFAULT_EXTERNAL_TYPES) -> Set[str]: """IDs of the system requirements defined in the umbrella SPEC.md. They appear as headings (``### SR-001 - ...``) and as bolded leading table cells (``| **PR-001** | ... |``), so accept both. """ alternation = "|".join(re.escape(t) for t in types) heading_re = re.compile(rf"^#{{1,6}}\s+\**((?:{alternation})-\d{{3}})\**\b") cell_re = re.compile(rf"(?:{alternation})-\d{{3}}") ids: Set[str] = set() for line in markdown.split("\n"): stripped = line.strip() heading = heading_re.match(stripped) if heading: ids.add(heading.group(1)) continue if stripped.startswith("|"): cells = _row_cells(stripped) if cells: first = cells[0].replace("*", "").strip() if cell_re.fullmatch(first): ids.add(first) return ids def per_type_stats(report: Report) -> Dict[str, Tuple[int, int, int]]: """``{type: (covered, unexecuted, defined)}``. Covered here means the same thing it means in the headline figure, so the per-type rows sum to it. Reporting "traced and defined" per type while the total excludes unexecuted requirements is how a breakdown quietly stops adding up to its own total. """ covered = set(report.coverage.covered) unexecuted = set(report.coverage.unexecuted) stats: Dict[str, Tuple[int, int, int]] = {} for req_type in report.register.types: prefix = req_type + "-" stats[req_type] = ( len([i for i in covered if i.startswith(prefix)]), len([i for i in unexecuted if i.startswith(prefix)]), report.register.count(req_type), ) return stats def _requirement_state(register: Register, req_id: str, tagged: bool) -> str: if not tagged: return "untagged" if not register.is_ci_executable(req_id): return "tagged-unexecuted" return "covered" def report_to_json_dict(report: Report) -> dict: register = report.register config = report.config defined = {t: register.count(t) for t in register.types} defined["total"] = register.total requirement_detail = {} for req_id, req in sorted(register.requirements.items()): entries = report.requirement_map.get(req_id, []) requirement_detail[req_id] = { "requirement": req.text, "tracesTo": req.traces_to, "priority": req.priority, "status": req.status, "tiers": sorted(req.tiers), "ciExecutable": register.is_ci_executable(req_id), "taggedIn": sorted({e.file for e in entries}), "state": _requirement_state(register, req_id, bool(entries)), } return { "timestamp": report.timestamp, # Echoed so a report can be read without guessing how it was produced - # a shared tool's output is ambiguous otherwise. "config": { "source": config.source, "root": str(config.root), "requirementTypes": list(config.requirement_types), "testTypes": list(config.test_types), "externalTypes": list(config.external_types), "sourceSuffixes": sorted(config.source_suffixes), "sourceRoots": list(config.source_roots), "requirements": config.requirements_path, "systemSpec": config.system_spec_path, "ciExecutableTiers": sorted(config.ci_executable_tiers), "minCoverage": config.min_coverage, }, "totalFiles": len(report.scan.files), "totalTraces": len(report.scan.traces), "totalExceptions": len(report.scan.exceptions), "requirements": {k: [e.to_dict() for e in v] for k, v in sorted(report.requirement_map.items())}, "byType": report.by_type, "defined": defined, "coverage": report.coverage.to_dict(), "unexecutableRequirements": sorted(register.unexecutable_ids()), "parentlessRequirements": sorted(register.parentless_ids()), "withdrawn": sorted(register.withdrawn), "exceptions": [e.to_dict() for e in report.scan.exceptions], "externalOrphans": report.external_orphans, "diagnostics": report.scan.diagnostics.to_dict(), "requirementDetail": requirement_detail, } # -------------------------------------------------------------------------- # Output formats # -------------------------------------------------------------------------- def generate_markdown(report: Report) -> str: register = report.register config = report.config cov = report.coverage out: List[str] = [] add = out.append req_link = Path(config.requirements_path).name add("# Requirements traceability matrix") add("") add("") add("") add("") add(f"**Generated:** {report.timestamp}") add("") add(f"Denominators are read from [`{req_link}`]({req_link}) at run time, " "never hardcoded. Coverage counts a requirement only when it is tagged " "in source **and** has a verification tier this repo's CI host can " f"execute (`{', '.join(sorted(config.ci_executable_tiers))}`).") add("") add("## Summary") add("") add("| Metric | Value |") add("|---|---|") add(f"| Source files scanned | {len(report.scan.files)} |") add(f"| TRACES tags found | {len(report.scan.traces)} |") add(f"| EXCEPTION tags found | {len(report.scan.exceptions)} |") add(f"| Requirements defined | {register.total} |") add(f"| Requirements covered | {len(cov.covered)} |") add(f"| **Coverage** | **{cov.percent}%** ({len(cov.covered)}/{cov.total}) |") add(f"| Coverage of CI-executable scope | {cov.ci_percent}% " f"({len(cov.covered)}/{cov.ci_executable_total}) |") add(f"| Tagged but unexecuted in CI | {len(cov.unexecuted)} |") add(f"| Orphan tags | {len(cov.orphaned)} |") add("") add("### By type") add("") add("| Type | Covered | Tagged but unexecuted | Defined |") add("|---|---|---|---|") for req_type, (covered, unexecuted, defined_n) in per_type_stats(report).items(): add(f"| {req_type} | {covered} | {unexecuted} | {defined_n} |") add("") for req_type in tuple(config.test_types) + tuple(config.external_types): tagged = report.by_type.get(req_type, []) if tagged: add(f"- **{req_type}** tags present (separate taxonomy, not counted " f"in coverage): {', '.join(tagged)}") add("") unexecutable = sorted(register.unexecutable_ids()) add("## Not executable in CI") add("") add("These requirements have no verification tier this repo's CI host can " "run, so a tag on them is evidence of *intent*, not of verification. " "They are never counted as covered.") add("") if unexecutable: add("| ID | Tiers | Tagged in source | Requirement |") add("|---|---|---|---|") for req_id in unexecutable: req = register.requirements[req_id] tagged = "yes" if req_id in report.requirement_map else "no" add(f"| {req_id} | {', '.join(sorted(req.tiers)) or '-'} | {tagged} " f"| {_truncate(req.text)} |") else: add("_None._") add("") if cov.unexecuted: add(f"**Tagged but unexecuted:** {', '.join(cov.unexecuted)} — a test " "exists and is tagged, but this CI host cannot run it. Report " "those runs separately.") add("") add("## Orphan tags") add("") add(f"A tag naming an ID `{req_link}` does not define. This is what " "renumbering produces, and what a typo produces.") add("") if cov.orphaned: add("| ID | Locations |") add("|---|---|") for req_id in cov.orphaned: where = ", ".join(f"`{e.file}:{e.line}`" for e in report.requirement_map[req_id]) add(f"| {req_id} | {where} |") else: add("_None._") add("") add("## Requirements tracing up to nothing") add("") add("A register row whose `Traces to` cell names no parent. Work serving no " "stated goal is how scope creeps in, and it is invisible unless " "something looks.") add("") parentless = sorted(register.parentless_ids()) add(", ".join(f"`{i}`" for i in parentless) if parentless else "_None._") add("") add("## Recorded exceptions") add("") add("Deliberate, documented departures from an invariant " "(`EXCEPTION: XX-nnn `). Reported separately and never counted " "as coverage — an exception is a decision to be reviewed, not evidence " "a requirement is met.") add("") if report.scan.exceptions: add("| Requirement | Location | Reason |") add("|---|---|---|") for exc in report.scan.exceptions: reason = _truncate(exc.reason, 90) or "**no reason recorded**" add(f"| {exc.requirement} | [`{exc.file}:{exc.line}`]" f"({_source_link(exc.file)}#L{exc.line}) | {reason} |") else: add("_None._") add("") add("## Register") add("") add("| ID | Status | Tier | Traces to | Trace state | Tagged in | Requirement |") add("|---|---|---|---|---|---|---|") order = {t: n for n, t in enumerate(register.types)} for req_id, req in sorted( register.requirements.items(), key=lambda kv: (order.get(kv[0].split("-")[0], 99), kv[0])): entries = report.requirement_map.get(req_id, []) state = {"covered": "covered", "tagged-unexecuted": "tagged, unexecuted", "untagged": "untagged"}[ _requirement_state(register, req_id, bool(entries))] files = ", ".join(f"`{f}`" for f in sorted({e.file for e in entries})) or "-" tiers = ", ".join(sorted(req.tiers)) or "unset" add(f"| {req_id} | {_truncate(req.status, 20) or '-'} | {tiers} " f"| {_truncate(req.traces_to, 40) or '-'} " f"| {state} | {files} | {_truncate(req.text)} |") add("") add("## Detailed mapping") add("") if not report.requirement_map: add("_No TRACES tags found yet. Tags are added as code is written; an " "empty matrix on a new tree is the correct reading, not a failure._") add("") for req_id in sorted(report.requirement_map): entries = report.requirement_map[req_id] add(f"### {req_id}") add("") add(f"**Locations:** {len(entries)}") add("") for entry in entries: add(f"- [`{entry.file}:{entry.line}`]" f"({_source_link(entry.file)}#L{entry.line}) — " f"`{_truncate(entry.context, 90)}`") add("") if not report.scan.diagnostics.empty: add("## Tag diagnostics") add("") diags = report.scan.diagnostics for label, items in ( ("Malformed tags", diags.malformed_tags), ("Groups mixing requirement types (pipe separates types)", diags.mixed_type_groups), ("Unrecognised ID prefixes", diags.unknown_id_types), ("Exceptions with no recorded reason", diags.exceptions_without_reason), ): if items: add(f"**{label}:**") add("") for item in items: detail = {k: v for k, v in item.items() if k not in ("file", "line")} add(f"- `{item.get('file')}:{item.get('line')}` — {detail}") add("") return "\n".join(out) + "\n" def _source_link(file_rel: str) -> str: """Link from the matrix (in docs/) back to a source file at the repo root.""" return "../" + file_rel def _truncate(text: str, limit: int = 70) -> str: """Collapse to one line, cap the length, and escape table-breaking pipes.""" text = " ".join(text.split()) if len(text) > limit: text = text[: limit - 1] + "…" return text.replace("|", "\\|") def format_coverage_report(report: Report) -> Tuple[str, int]: """Human-readable gate output plus the exit code it implies.""" register = report.register config = report.config cov = report.coverage lines: List[str] = [] add = lines.append failures: List[str] = [] add("Requirement traceability") add("=" * 72) add(f"Config : {config.source}") add(f"Repo root : {config.root}") add(f"Requirement types : {', '.join(config.requirement_types)}") add(f"Source files scanned : {len(report.scan.files)} " f"({', '.join(sorted(config.source_suffixes))} under " f"{', '.join(config.source_roots)})") add(f"TRACES tags found : {len(report.scan.traces)}") add(f"EXCEPTION tags found : {len(report.scan.exceptions)}") add("") # Self-checks. These are what make the gate mean something at a low # threshold, and what makes a misconfigured shared tool loud instead of # silent: a repo whose requirement_types or source_suffixes are wrong parses # nothing and scans nothing, and a plausible-looking 0% would hide it. if register.total == 0: failures.append( f"{config.requirements_path} parsed to ZERO requirements. Either " "the path is wrong, the register parser is broken, or " f"requirement_types ({', '.join(config.requirement_types) or 'none'}) " "does not match the prefixes it defines. Refusing to report " "coverage.") if not report.scan.files: failures.append( "no source files were scanned. Either source_roots " f"({', '.join(config.source_roots) or 'none'}) do not exist, or the " f"suffixes ({', '.join(sorted(config.source_suffixes)) or 'none'}) " "do not match this repo's languages. Refusing to report coverage " "against an empty tree.") add("Coverage by type (covered / defined):") for req_type, (covered, unexecuted, defined_n) in per_type_stats(report).items(): suffix = f" (+{unexecuted} tagged but unexecuted)" if unexecuted else "" add(f" {req_type}: {covered} / {defined_n}{suffix}") add("") add(f"Overall : {len(cov.covered)} / {cov.total} ({cov.percent}%)") add(f"CI scope : {len(cov.covered)} / {cov.ci_executable_total} " f"({cov.ci_percent}%) [excludes {cov.total - cov.ci_executable_total} " "requirement(s) this CI host cannot verify]") add("") unexecutable = sorted(register.unexecutable_ids()) if unexecutable: add("Not executable on this CI host (tiers outside " f"{', '.join(sorted(config.ci_executable_tiers))}): {len(unexecutable)}") add(f" {', '.join(unexecutable)}") if cov.unexecuted: add("") add("TAGGED BUT UNEXECUTED - a test exists and is tagged, but this CI " "host cannot run it.") add(" These are NOT counted as covered:") for req_id in cov.unexecuted: tiers = ", ".join(sorted(register.requirements[req_id].tiers)) add(f" {req_id} (tier {tiers})") add("") parentless = sorted(register.parentless_ids()) if parentless: add(f"WARNING: {len(parentless)} requirement(s) trace up to nothing - " "no parent recorded in the register's `Traces to` column. Work " "serving no stated goal is how scope creeps in:") add(f" {', '.join(parentless)}") add("") if cov.tier_unknown and register.uses_tiers: # Only meaningful where the register uses tiers at all; otherwise every # requirement is listed and the warning trains people to ignore it. add(f"WARNING: {len(cov.tier_unknown)} requirement(s) have no " "verification tier in the register; they are counted as " "CI-executable by default. Add them to the verification plan:") add(f" {', '.join(cov.tier_unknown)}") add("") elif register.total and not register.uses_tiers: add("Note: this register assigns no verification tiers, so nothing is " "excluded as unexecutable.") add("") if report.scan.exceptions: add(f"Recorded invariant exceptions: {len(report.scan.exceptions)}") for exc in report.scan.exceptions: reason = exc.reason or "(NO REASON RECORDED)" add(f" {exc.requirement} {exc.file}:{exc.line} {_truncate(reason, 80)}") add("") if cov.orphaned: add("ORPHAN TAGS - traced in source, not defined in " f"{config.requirements_path}:") for req_id in cov.orphaned: where = ", ".join(f"{e.file}:{e.line}" for e in report.requirement_map[req_id]) add(f" {req_id} ({where})") add(" Fix the tag, or add the requirement to the register.") add("") if report.external_orphans: add("ORPHAN SYSTEM TAGS - IDs the system spec does not define:") add(f" {', '.join(report.external_orphans)}") add("") diags = report.scan.diagnostics if not diags.empty: add("Tag diagnostics:") for item in diags.malformed_tags: add(f" malformed {item.get('file')}:{item.get('line')} {item}") for item in diags.mixed_type_groups: add(f" mixed types {item['file']}:{item['line']} {item['group']} " "(a pipe, not a comma, separates types)") for item in diags.unknown_id_types: add(f" unknown id {item['file']}:{item['line']} {item['id']}") for item in diags.exceptions_without_reason: add(f" exception without reason {item['file']}:{item['line']} " f"{item['requirement']}") add("") # A ratio above 100% means the computation is broken. This is the check # that would have caught JellyTau's frozen denominators immediately. if cov.percent > 100 or cov.ci_percent > 100: failures.append( f"coverage ({cov.percent}%) exceeds 100% - the gate is " "miscomputing. Do not trust this run.") if len(cov.covered) > cov.total: failures.append( f"covered ({len(cov.covered)}) exceeds defined ({cov.total}) - " "the gate is miscomputing.") if cov.orphaned and not config.allow_orphans: failures.append( f"{len(cov.orphaned)} orphan tag(s): {', '.join(cov.orphaned)}") if cov.percent < config.min_coverage: failures.append( f"coverage ({cov.percent}%) is below the minimum " f"({config.min_coverage}%)") if failures: add("FAILED:") for reason in failures: add(f" - {reason}") return "\n".join(lines) + "\n", 1 add(f"OK: coverage {cov.percent}% >= minimum {config.min_coverage}%, " f"{len(cov.orphaned)} orphan tag(s)") return "\n".join(lines) + "\n", 0 # -------------------------------------------------------------------------- # CLI # -------------------------------------------------------------------------- def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="extract_traces.py", description="Extract requirement traces and report coverage. Shared by " "every JRay component repo; configured per repo via " f"{CONFIG_FILENAME} or the flags below.", formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("--format", choices=("json", "markdown", "coverage"), default="coverage", help="output written to stdout (default: coverage)") parser.add_argument("--config", type=Path, default=None, help=f"path to {CONFIG_FILENAME} (default: the nearest " "one at or above the working directory)") parser.add_argument("--no-config", action="store_true", help="ignore any config file; use flags only") parser.add_argument("--print-example-config", action="store_true", help=f"print an annotated {CONFIG_FILENAME} and exit") parser.add_argument("--root", type=Path, default=None, help="repository root (default: the config file's " "directory, else the working directory)") group = parser.add_argument_group("what this repo defines") group.add_argument("--requirement-type", action="append", metavar="XX", help="ID prefix this repo's register defines; repeatable. " "Replaces the configured list") group.add_argument("--test-type", action="append", metavar="XX", help="test ID prefix, excluded from coverage; repeatable") group.add_argument("--external-type", action="append", metavar="XX", help="system-spec ID prefix, excluded from coverage; " "repeatable") group = parser.add_argument_group("what this repo scans") group.add_argument("--language", action="append", choices=sorted(LANGUAGE_SUFFIXES), help="named suffix group; repeatable, additive") group.add_argument("--source-suffix", action="append", metavar=".EXT", help="extra source suffix; repeatable, additive") group.add_argument("--source-root", action="append", metavar="DIR", help="directory to scan; repeatable. Replaces the " "configured list") group.add_argument("--exclude-dir", action="append", metavar="NAME", help="directory name to skip anywhere in the tree; " "repeatable, additive") group = parser.add_argument_group("paths") group.add_argument("--requirements", type=Path, default=None, help=f"register path (default: {DEFAULT_REQUIREMENTS_PATH})") group.add_argument("--system-spec", type=Path, default=None, help="SPEC.md defining the external types; enables " "orphan checking for them") group.add_argument("--json-out", type=Path, default=None, help=f"JSON report path (default: {DEFAULT_JSON_PATH})") group.add_argument("--markdown-out", type=Path, default=None, help=f"matrix path (default: {DEFAULT_MATRIX_PATH})") group.add_argument("--no-write", action="store_true", help="do not write report files, only print") group = parser.add_argument_group("policy") group.add_argument("--min-coverage", type=float, default=None, help="minimum coverage percent; below it the run fails") group.add_argument("--allow-orphans", action="store_true", help="report orphan tags without failing") group.add_argument("--ci-executable-tier", action="append", metavar="TIER", help="verification tier this CI host can run; " "repeatable. Replaces the configured set") return parser def main(argv: Optional[Sequence[str]] = None) -> int: args = build_arg_parser().parse_args(argv) if args.print_example_config: print(example_config(), end="") return 0 try: config = load_config(args) except ConfigError as exc: print(f"FAILED: {exc}", file=sys.stderr) return 2 req_path = config.resolve(config.requirements_path) if req_path is None or not req_path.is_file(): print(f"FAILED: requirements register not found at {req_path}", file=sys.stderr) return 2 register = read_register(config) files = iter_source_files(config) scan = scan_files(files, config) system_ids = None spec_path = config.resolve(config.system_spec_path) if spec_path is not None: if not spec_path.is_file(): print(f"FAILED: system spec not found at {spec_path}", file=sys.stderr) return 2 system_ids = parse_system_spec(spec_path.read_text(encoding="utf-8"), config.external_types) report = build_report(config, register, scan, system_ids) json_text = json.dumps(report_to_json_dict(report), indent=2, sort_keys=False) markdown_text = generate_markdown(report) for target, text in ((config.resolve(config.json_path), json_text), (config.resolve(config.matrix_path), markdown_text)): if target is not None: target.parent.mkdir(parents=True, exist_ok=True) target.write_text(text, encoding="utf-8") if args.format == "json": print(json_text) return 0 if args.format == "markdown": print(markdown_text, end="") return 0 text, code = format_coverage_report(report) print(text, end="") return code if __name__ == "__main__": sys.exit(main())