#!/usr/bin/env python3 """Validate that every declared rfgen entry point has one E2E disposition.""" from __future__ import annotations import ast import json import tomllib from pathlib import Path ROOT = Path(__file__).resolve().parents[2] MANIFEST_PATH = ROOT / "ci" / "integration" / "coverage_manifest.json" CLI_SOURCE_PATH = ROOT / "src" / "rfgen" / "cli.py" def _public_cli_commands() -> frozenset[str]: """Extract Typer's public command inventory from the canonical CLI module.""" module = ast.parse(CLI_SOURCE_PATH.read_text(encoding="utf-8")) commands: set[str] = set() for node in module.body: if not isinstance(node, ast.FunctionDef): continue for decorator in node.decorator_list: if not isinstance(decorator, ast.Call) or not isinstance(decorator.func, ast.Attribute): continue if not isinstance(decorator.func.value, ast.Name) or decorator.func.value.id != "app": continue if decorator.func.attr != "command": continue if not decorator.args: commands.add(node.name.replace("_", "-")) elif isinstance(decorator.args[0], ast.Constant) and isinstance(decorator.args[0].value, str): commands.add(decorator.args[0].value) return frozenset(commands) def _scenario_exists(scenario: str) -> bool: """Return whether a checked-in E2E test node exists without importing it.""" path_text, separator, node_name = scenario.partition("::") path = ROOT / path_text if not separator or not node_name or not path.is_file() or not path.is_relative_to( ROOT / "tests" / "integration" / "local" ): return False module = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) return any(isinstance(item, ast.FunctionDef) and item.name == node_name for item in module.body) def validate(manifest: dict[str, object], metadata: dict[str, object]) -> list[str]: """Return explicit manifest drift errors without mutating either artifact.""" errors: list[str] = [] covered = manifest.get("covered") excluded = manifest.get("excluded") optional_extras = manifest.get("optional_extras") scenarios = manifest.get("scenarios") entry_points = metadata.get("project", {}).get("entry-points", {}) if not isinstance(covered, dict) or not isinstance(excluded, dict): return ["manifest must contain object-valued covered and excluded sections"] if not isinstance(optional_extras, dict): return ["manifest must contain an object-valued optional_extras section"] if not isinstance(scenarios, dict): return ["manifest must contain an object-valued scenarios section"] if not isinstance(entry_points, dict): return ["pyproject has no entry-point mapping"] declared_extras = metadata.get("project", {}).get("optional-dependencies", {}) if not isinstance(declared_extras, dict): return ["pyproject has no optional-dependencies mapping"] declared_extra_names = set(declared_extras) manifest_extra_names = set(optional_extras) unknown_extras = manifest_extra_names - declared_extra_names if unknown_extras: errors.append(f"unknown optional extras: {sorted(unknown_extras)}") missing_extras = declared_extra_names - manifest_extra_names if missing_extras: errors.append(f"missing optional-extra dispositions: {sorted(missing_extras)}") allowed_dispositions = { "installed_exercised", "installed_failure_boundary", "external_manual", "provider_excluded", "not_retained_runtime", "test_tooling", } for name, entry in sorted(optional_extras.items()): if not isinstance(entry, dict): errors.append(f"optional_extras[{name!r}] must be an object") continue disposition = entry.get("disposition") reason = entry.get("reason") scenario = entry.get("scenario") if disposition not in allowed_dispositions: errors.append(f"optional_extras[{name!r}] has invalid disposition") if disposition in {"external_manual", "provider_excluded", "not_retained_runtime", "test_tooling", "installed_failure_boundary"} and not isinstance(reason, str): errors.append(f"optional_extras[{name!r}] requires a reason") if disposition in {"installed_exercised", "installed_failure_boundary"} and not isinstance(scenario, str): errors.append(f"optional_extras[{name!r}] requires an E2E scenario") elif isinstance(scenario, str) and not _scenario_exists(scenario): errors.append(f"optional_extras[{name!r}] references missing E2E scenario: {scenario}") all_groups = set(entry_points) | set(covered) | set(excluded) for group in sorted(all_groups): declared_raw = entry_points.get(group, {}) declared = ( set(_public_cli_commands()) if group == "rfgen.commands" else (set(declared_raw) if isinstance(declared_raw, dict) else set()) ) covered_raw = covered.get(group, []) excluded_raw = excluded.get(group, {}) if not isinstance(covered_raw, list) or not all(isinstance(item, str) for item in covered_raw): errors.append(f"covered[{group!r}] must be a string list") continue if not isinstance(excluded_raw, dict) or not all( isinstance(name, str) and isinstance(reason, str) and reason.strip() for name, reason in excluded_raw.items() ): errors.append(f"excluded[{group!r}] must map identifiers to non-empty reasons") continue covered_names = set(covered_raw) excluded_names = set(excluded_raw) if len(covered_names) != len(covered_raw): errors.append(f"duplicate covered identifier in {group}") overlap = covered_names & excluded_names if overlap: errors.append(f"ambiguous covered/excluded identifiers in {group}: {sorted(overlap)}") unknown = (covered_names | excluded_names) - declared if unknown: errors.append(f"unknown manifest identifiers in {group}: {sorted(unknown)}") missing = declared - covered_names - excluded_names if missing: errors.append(f"missing manifest identifiers in {group}: {sorted(missing)}") scenario_group = scenarios.get(group, {}) if not isinstance(scenario_group, dict): errors.append(f"scenarios[{group!r}] must map identifiers to scenario lists") continue missing_scenarios = covered_names - set(scenario_group) unknown_scenarios = set(scenario_group) - covered_names if missing_scenarios: errors.append(f"missing scenarios in {group}: {sorted(missing_scenarios)}") if unknown_scenarios: errors.append(f"unknown scenario identifiers in {group}: {sorted(unknown_scenarios)}") for name, scenario_list in scenario_group.items(): if not isinstance(scenario_list, list) or not scenario_list or not all( isinstance(scenario, str) and _scenario_exists(scenario) for scenario in scenario_list ): errors.append(f"invalid or stale scenario mapping for {group}.{name}") command_names = set(covered.get("rfgen.commands", [])) | set( excluded.get("rfgen.commands", {}) ) public_cli_commands = _public_cli_commands() if command_names != public_cli_commands: errors.append( "CLI command inventory drift: expected " f"{sorted(public_cli_commands)}, got {sorted(command_names)}" ) return errors def main() -> None: manifest = json.loads(MANIFEST_PATH.read_text(encoding="utf-8")) metadata = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8")) errors = validate(manifest, metadata) if errors: raise SystemExit("E2E coverage manifest failed:\n- " + "\n- ".join(errors)) print("E2E coverage manifest passed.") if __name__ == "__main__": main()