diff --git a/.aiwg/architecture/sketch-testing-quality-program.md b/.aiwg/architecture/sketch-testing-quality-program.md index 0f4c1fe..59341db 100644 --- a/.aiwg/architecture/sketch-testing-quality-program.md +++ b/.aiwg/architecture/sketch-testing-quality-program.md @@ -16,7 +16,7 @@ produce retained evidence and never silently convert a required failure into a success. Tool versions remain project-controlled. The default gate must not access external services, inherit user model caches, or require credentials. -Coverage policy evolves by wave: +The initial coverage waves are complete: - measurement wave: preserve the 49% line floor and establish branch baseline; - boundary wave: repository line coverage at least 55%, changed lines at least @@ -24,3 +24,10 @@ Coverage policy evolves by wave: - integration wave: repository line coverage at least 60%; - mature CPU-testable target: at least 80% line and 75% branch coverage, with exclusions limited to documented conditional environment code. + +Wave A locks that baseline by generating exact-base coverage in the same CI run, +rejecting per-module line or branch regressions, validating a machine-readable +source-to-test and conditional-gate graph, retaining normalized trend evidence +for 90 days, and enforcing owner/issue/expiry requirements for repeated flaky +tests. The exact thresholds and exclusion graph remain owned by +`ci/test-quality-policy.json`; `ci/test-risk-map.json` owns test responsibility. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5c0cd0c..2f007ca 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -203,9 +203,11 @@ jobs: scripts/check_mutation_score.py scripts/check_quality_policy.py scripts/check_conditional_policy.py + scripts/check_test_risk_map.py scripts/conditional_gate_summary.py scripts/run_conditional_gate.py scripts/run_repeat_gate.py + scripts/write_test_evidence.py scripts/check_supply_chain_policy.py scripts/gemma4_12b_recursive_loop.py @@ -217,9 +219,11 @@ jobs: scripts/check_mutation_score.py scripts/check_quality_policy.py scripts/check_conditional_policy.py + scripts/check_test_risk_map.py scripts/conditional_gate_summary.py scripts/run_conditional_gate.py scripts/run_repeat_gate.py + scripts/write_test_evidence.py scripts/check_supply_chain_policy.py scripts/gemma4_12b_recursive_loop.py || true @@ -288,26 +292,57 @@ jobs: --cov-report="xml:test-results/coverage-py${{ matrix.python-version }}.xml" \ --cov-report="json:test-results/coverage-py${{ matrix.python-version }}.json" + - name: Generate exact-base coverage for module regression comparison + if: matrix.python-version == '3.12' + env: + BASE_TEST_ENV: /tmp/obliteratus-base-test-env + BASE_WORKTREE: /tmp/obliteratus-base-worktree + COVERAGE_BASE: ${{ github.event.pull_request.base.sha || github.event.before }} + run: | + git rev-parse --verify "${COVERAGE_BASE}^{commit}" + git worktree add --detach "$BASE_WORKTREE" "$COVERAGE_BASE" + ( + cd "$BASE_WORKTREE" + UV_PROJECT_ENVIRONMENT="$BASE_TEST_ENV" \ + uv sync --locked --no-default-groups --extra dev --no-editable + COVERAGE_FILE="$RUNNER_TEMP/.coverage-base" \ + "$BASE_TEST_ENV/bin/python" -m pytest \ + -m "not slow and not gpu and not mps and not mlx and not network and not download and not remote and not operator_ui" \ + --cov-branch \ + --cov-fail-under=0 \ + --cov-report="json:$GITHUB_WORKSPACE/test-results/base-coverage-py3.12.json" + ) | tee test-results/base-tests-py3.12.log + - name: Enforce line and branch coverage floors env: COVERAGE_BASE: ${{ github.event.pull_request.base.sha || github.event.before }} - run: >- - "$TEST_ENV/bin/python" scripts/check_coverage_thresholds.py - "test-results/coverage-py${{ matrix.python-version }}.json" - --min-line 60 - --min-branch 42 - --min-file obliteratus/device.py=70 - --min-file obliteratus/models/loader.py=70 - --min-file obliteratus/architecture_profiles.py=70 - --min-file obliteratus/cli.py=70 - --min-file obliteratus/mlx_backend.py=70 - --min-file obliteratus/evaluation/metrics.py=70 - --min-file obliteratus/evaluation/advanced_metrics.py=70 - --min-file obliteratus/reporting/report.py=70 - --min-file obliteratus/community.py=70 - --min-file obliteratus/telemetry.py=70 - --min-changed 90 - --base-ref "$COVERAGE_BASE" + run: | + module_args=() + if [ "${{ matrix.python-version }}" = "3.12" ]; then + module_args=( + --base-report test-results/base-coverage-py3.12.json + --touched-module-no-regression + --new-module-min-line 80 + --new-module-min-branch 75 + ) + fi + "$TEST_ENV/bin/python" scripts/check_coverage_thresholds.py \ + "test-results/coverage-py${{ matrix.python-version }}.json" \ + --min-line 60 \ + --min-branch 42 \ + --min-file obliteratus/device.py=70 \ + --min-file obliteratus/models/loader.py=70 \ + --min-file obliteratus/architecture_profiles.py=70 \ + --min-file obliteratus/cli.py=70 \ + --min-file obliteratus/mlx_backend.py=70 \ + --min-file obliteratus/evaluation/metrics.py=70 \ + --min-file obliteratus/evaluation/advanced_metrics.py=70 \ + --min-file obliteratus/reporting/report.py=70 \ + --min-file obliteratus/community.py=70 \ + --min-file obliteratus/telemetry.py=70 \ + --min-changed 90 \ + --base-ref "$COVERAGE_BASE" \ + "${module_args[@]}" - name: Enforce mature CPU-scope coverage and immutable quality policy run: | @@ -315,6 +350,25 @@ jobs: --policy ci/test-quality-policy.json \ --coverage "test-results/coverage-py${{ matrix.python-version }}.json" "$TEST_ENV/bin/python" scripts/check_conditional_policy.py + "$TEST_ENV/bin/python" scripts/check_test_risk_map.py + + - name: Write normalized test trend evidence + if: always() + env: + COVERAGE_BASE: ${{ github.event.pull_request.base.sha || github.event.before }} + run: | + base_args=() + if [ -f test-results/base-coverage-py3.12.json ]; then + base_args=(--base-coverage test-results/base-coverage-py3.12.json) + fi + "$TEST_ENV/bin/python" scripts/write_test_evidence.py \ + --coverage "test-results/coverage-py${{ matrix.python-version }}.json" \ + --junit "test-results/junit-py${{ matrix.python-version }}.xml" \ + --head-sha "$GITHUB_SHA" \ + --base-sha "$COVERAGE_BASE" \ + --python-version "${{ matrix.python-version }}" \ + --output "test-results/test-trend-py${{ matrix.python-version }}.json" \ + "${base_args[@]}" - name: Upload test and coverage evidence if: always() @@ -323,7 +377,7 @@ jobs: name: test-evidence-py${{ matrix.python-version }} path: test-results/ if-no-files-found: error - retention-days: 14 + retention-days: 90 quality-depth: name: Quality depth @@ -375,6 +429,25 @@ jobs: "$QUALITY_ENV/bin/python" scripts/check_mutation_score.py \ quality-evidence/mutation-stats.json --minimum 70 + - name: Write normalized quality trend evidence + if: always() + env: + COVERAGE_BASE: ${{ github.event.pull_request.base.sha || github.event.before }} + run: | + evidence_args=() + if [ -f quality-evidence/repeat-gate.json ]; then + evidence_args+=(--repeat quality-evidence/repeat-gate.json) + fi + if [ -f quality-evidence/mutation-stats.json ]; then + evidence_args+=(--mutation quality-evidence/mutation-stats.json) + fi + "$QUALITY_ENV/bin/python" scripts/write_test_evidence.py \ + --head-sha "$GITHUB_SHA" \ + --base-sha "$COVERAGE_BASE" \ + --python-version "3.12" \ + --output quality-evidence/quality-trend-py3.12.json \ + "${evidence_args[@]}" + - name: Upload quality-depth evidence if: always() uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 @@ -382,7 +455,7 @@ jobs: name: quality-depth-py3.12 path: quality-evidence/ if-no-files-found: error - retention-days: 14 + retention-days: 90 supply-chain: name: Supply chain diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 34c205a..c61de0c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -40,6 +40,10 @@ modules. New changes should raise these floors rather than consume the existing margin. A separately measured mature CPU-only scope must remain at or above 80% statement and 75% branch coverage; its environment-bound exclusions and their conditional-test ownership are versioned in `ci/test-quality-policy.json`. +Every changed production module is also compared with coverage generated from +the exact base commit. Line and branch coverage may not regress independently, +and new modules start at 80% line / 75% branch coverage in addition to the 90% +changed-line requirement. The quality-depth job repeats the highest-consequence pure tests three times in different file orders and with different deterministic hash seeds. It also runs @@ -54,11 +58,13 @@ mutmut export-cicd-stats python scripts/check_mutation_score.py mutants/mutmut-cicd-stats.json --minimum 70 ``` -On the calibration runner, the repeat gate took approximately 23 seconds and -the 141-mutant selection took approximately 11 seconds after environment setup. -CI retains repeat timing/output, mutation timing, test selection, score, and -survivor evidence for 14 days. Thresholds may move downward only through a -time-bounded exception linked to a reviewed repository issue. +The repeat and mutation selections are intentionally bounded; on the Wave A +calibration runner each completed in under one minute after environment setup. +CI retains normalized coverage/JUnit trends and repeat/mutation evidence for 90 +days. Flake observations and time-bounded quarantines are governed in +`ci/test-quality-policy.json`; two observations within 30 days require an +owner, reason, repository issue, and expiry. Thresholds may move downward only +through a time-bounded exception linked to a reviewed repository issue. Hardware, model-download, network-service, operator-UI, and remote-provider tests run separately so the pull-request baseline stays offline and credential-free. See @@ -68,8 +74,15 @@ freshness. Validate the mapping between those gates and CPU coverage exclusions ```bash python scripts/check_conditional_policy.py +python scripts/check_test_risk_map.py ``` +The machine-readable source-to-test ownership graph lives in +`ci/test-risk-map.json`. Each behavior-changing PR must add a focused regression +or contract test, relevant negative/boundary coverage, and propagation/runtime +evidence for public options. Hardware or service changes require both a +deterministic boundary test and their mapped conditional gate. + ## Code Style We use [ruff](https://docs.astral.sh/ruff/) for linting and formatting: diff --git a/README.md b/README.md index 9d8c360..ac7d99a 100644 --- a/README.md +++ b/README.md @@ -761,8 +761,8 @@ pip install -e ".[dev]" pytest ``` -The mandatory CPU suite currently contains 1,130 tests across 54 test files, -including a repository-owned synthetic model that exercises the offline pipeline, +The mandatory CPU suite contains more than 1,100 tests, including a +repository-owned synthetic model that exercises the offline pipeline, installed-wheel CLI, study runner, transactional checkpoint recovery, and resumable auto-obliteration state. The suite also covers model/device/quantization/MLX boundaries, all analysis modules, architecture detection, visualization sanitization, @@ -770,7 +770,10 @@ community contributions, edge cases, and evaluation metrics. CI enforces at leas 60% repository statement coverage, 42% repository branch coverage, 90% changed-line coverage, and 80% statement / 75% branch coverage for the documented mature CPU-only scope. Deterministic property, order-repeat, and selective mutation -gates provide additional depth for numerical and policy-critical behavior. +gates provide additional depth for numerical and policy-critical behavior. CI +also rejects line or branch regressions in each touched production module by +measuring the exact base commit, retains normalized trend evidence for 90 days, +and validates the source-to-test ownership graph in `ci/test-risk-map.json`. Eight environment-bound test files run through the separately documented conditional workflow for model downloads, network services, operator UI, CUDA, bitsandbytes, MPS, MLX, and least-privileged remote execution. diff --git a/WORKSPACE.md b/WORKSPACE.md index 1858888..8cdde33 100644 --- a/WORKSPACE.md +++ b/WORKSPACE.md @@ -42,10 +42,14 @@ accelerator, or remote-execution credentials. Canonical required checks: -- `python -m ruff check --select F app.py obliteratus tests scripts/check_coverage_thresholds.py scripts/check_supply_chain_policy.py scripts/gemma4_12b_recursive_loop.py` -- `python -m pytest` (includes the measured 49% coverage floor) -- CI additionally enforces the measured 36% branch-coverage floor from its - retained coverage JSON report. +- the exact Ruff F and actionlint command set in [.github/workflows/ci.yml](.github/workflows/ci.yml); +- `python -m pytest` with at least 60% repository line coverage and 42% branch + coverage; +- at least 90% changed-line coverage plus no line or branch regression in any + touched production module, compared with coverage from the exact base commit; +- at least 80% line and 75% branch coverage for the documented mature + CPU-testable scope, plus a 70% selective mutation score and zero unexpected + warnings; - `python -m build --sdist --wheel` - `python -c 'import obliteratus; print(obliteratus.__version__)'` - `python -m obliteratus --help` @@ -55,6 +59,13 @@ in an independent environment outside the checkout, exercises both CLI entry paths, and retains the distributions plus evidence. Immutable CI action/tool pins are recorded in [ci/digests.txt](ci/digests.txt). +The source-to-test ownership graph is versioned in +[ci/test-risk-map.json](ci/test-risk-map.json). Coverage, JUnit, repeat, and +mutation trends are normalized into project-owned JSON and retained for 90 +days. Flake history and time-bounded quarantines are governed by +[ci/test-quality-policy.json](ci/test-quality-policy.json); a test observed +flaking twice in 30 days requires an active owner/issue-linked quarantine. + Python CI resolution is locked by `uv.lock`, including the official CPU-only PyTorch source for Linux and Windows. The required Supply chain job scans all supported Python versions for known vulnerabilities, scans the checkout for @@ -62,8 +73,9 @@ secrets with fully redacted evidence, enforces the packaged-dependency license allow list, and binds a CycloneDX SBOM to the built wheel. Exception and update rules are documented in [docs/SUPPLY_CHAIN_POLICY.md](docs/SUPPLY_CHAIN_POLICY.md). -GPU, MPS, model-download, network, and remote-execution checks are conditional -release or risk-surface gates, not part of the default CPU job. +GPU, MPS, MLX, model-download, external-evaluation, network, operator-UI, and +remote-execution checks are conditional release or risk-surface gates, not part +of the default CPU job. Use [.aiwg/bt6-maintainer.yaml](.aiwg/bt6-maintainer.yaml) and the project-local `bt6-maintainer` bundle for issue, pull-request, provider, and merge-train work. diff --git a/ci/test-quality-policy.json b/ci/test-quality-policy.json index 565b3cc..16190ea 100644 --- a/ci/test-quality-policy.json +++ b/ci/test-quality-policy.json @@ -9,6 +9,18 @@ "mutation_score": 70.0, "warning_budget": 0 }, + "critical_cpu_paths": [ + "obliteratus/device.py", + "obliteratus/models/loader.py", + "obliteratus/architecture_profiles.py", + "obliteratus/cli.py", + "obliteratus/mlx_backend.py", + "obliteratus/evaluation/metrics.py", + "obliteratus/evaluation/advanced_metrics.py", + "obliteratus/reporting/report.py", + "obliteratus/community.py", + "obliteratus/telemetry.py" + ], "mature_cpu_scope": { "description": "All source modules except boundaries that intrinsically require a real model runtime, an external service, an interactive UI, or remote/hardware execution.", "exclusions": [ @@ -140,5 +152,12 @@ } ] }, + "test_evidence": { + "retention_days": 90, + "flake_window_days": 30, + "maximum_quarantine_days": 30, + "flake_history": [], + "quarantines": [] + }, "threshold_exceptions": [] } diff --git a/ci/test-risk-map.json b/ci/test-risk-map.json new file mode 100644 index 0000000..f4ad80e --- /dev/null +++ b/ci/test-risk-map.json @@ -0,0 +1,214 @@ +{ + "schema_version": 1, + "owner": "OBLITERATUS maintainers", + "modules": [ + { + "path": "obliteratus/cli.py", + "risk_class": "cpu-contract", + "risk": "public parsing, validation, dispatch, and local/remote option propagation", + "required_tests": ["tests/test_cli.py", "tests/test_cli_boundaries.py"], + "conditional_gates": [] + }, + { + "path": "obliteratus/config.py", + "risk_class": "cpu-contract", + "risk": "configuration defaults, normalization, validation, and serialization", + "required_tests": ["tests/test_config.py", "tests/test_config_properties.py"], + "conditional_gates": [] + }, + { + "path": "obliteratus/architecture_profiles.py", + "risk_class": "cpu-contract", + "risk": "architecture detection and projection-path contracts", + "required_tests": ["tests/test_architecture_profiles.py", "tests/test_gemma4_support.py"], + "conditional_gates": [] + }, + { + "path": "obliteratus/community.py", + "risk_class": "cpu-contract", + "risk": "contribution schema, aggregation, and atomic persistence", + "required_tests": ["tests/test_community.py"], + "conditional_gates": [] + }, + { + "path": "obliteratus/telemetry.py", + "risk_class": "cpu-contract", + "risk": "research telemetry schema, aggregation, and filesystem behavior", + "required_tests": ["tests/test_telemetry.py"], + "conditional_gates": [] + }, + { + "path": "obliteratus/reporting/report.py", + "risk_class": "cpu-contract", + "risk": "report schema, output paths, serialization, and plotting contracts", + "required_tests": ["tests/test_report.py"], + "conditional_gates": [] + }, + { + "path": "obliteratus/evaluation/metrics.py", + "risk_class": "cpu-contract", + "risk": "research metric semantics and numerical invariants", + "required_tests": ["tests/test_metrics.py", "tests/test_property_contracts.py"], + "conditional_gates": [] + }, + { + "path": "obliteratus/evaluation/advanced_metrics.py", + "risk_class": "cpu-contract", + "risk": "refusal detection, confidence intervals, and robustness metrics", + "required_tests": ["tests/test_advanced_metrics.py", "tests/test_property_contracts.py"], + "conditional_gates": [] + }, + { + "path": "obliteratus/device.py", + "risk_class": "mixed-runtime", + "risk": "device and dtype selection across CPU, CUDA, and Apple backends", + "required_tests": ["tests/test_device_boundaries.py"], + "conditional_gates": ["cuda-runtime", "mps-runtime"] + }, + { + "path": "obliteratus/models/loader.py", + "risk_class": "mixed-runtime", + "risk": "model loading, cache, architecture, quantization, and device-map boundaries", + "required_tests": ["tests/test_loader_boundaries.py", "tests/test_offline_integration.py"], + "conditional_gates": ["cuda-runtime", "bitsandbytes-runtime"] + }, + { + "path": "obliteratus/mlx_backend.py", + "risk_class": "mixed-runtime", + "risk": "MLX model discovery, tensor placement, mutation, and persistence", + "required_tests": ["tests/test_mlx_backend_boundaries.py"], + "conditional_gates": ["mlx-runtime"] + }, + { + "path": "obliteratus/abliterate.py", + "risk_class": "mixed-runtime", + "risk": "core model mutation, checkpoint, evaluation, and save pipeline", + "required_tests": [ + "tests/test_abliterate.py", + "tests/test_abliterate_extended.py", + "tests/test_checkpoint_atomicity.py", + "tests/test_offline_integration.py" + ], + "conditional_gates": ["model-download-runtime"] + }, + { + "path": "obliteratus/auto_obliterate.py", + "risk_class": "mixed-runtime", + "risk": "automated search state, retry, scoring, and checkpoint behavior", + "required_tests": ["tests/test_auto_obliterate.py"], + "conditional_gates": ["model-download-runtime"] + }, + { + "path": "obliteratus/bayesian_optimizer.py", + "risk_class": "conditional-runtime", + "risk": "optional optimizer trials over repeated live model mutation and evaluation", + "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_model_download_runtime.py"], + "conditional_gates": ["model-download-runtime"] + }, + { + "path": "obliteratus/evaluation/baselines.py", + "risk_class": "mixed-runtime", + "risk": "control baseline mutation and comparative evaluation", + "required_tests": ["tests/test_evaluator.py", "tests/conditional/test_model_download_runtime.py"], + "conditional_gates": ["model-download-runtime"] + }, + { + "path": "obliteratus/evaluation/evaluator.py", + "risk_class": "mixed-runtime", + "risk": "dataset bounds, causal/classifier evaluation, and result semantics", + "required_tests": ["tests/test_evaluator.py", "tests/test_offline_integration.py"], + "conditional_gates": ["model-download-runtime"] + }, + { + "path": "obliteratus/informed_pipeline.py", + "risk_class": "mixed-runtime", + "risk": "multi-stage pipeline orchestration and stage-result contracts", + "required_tests": ["tests/test_informed_pipeline.py", "tests/test_offline_integration.py"], + "conditional_gates": ["model-download-runtime"] + }, + { + "path": "obliteratus/lora_ablation.py", + "risk_class": "conditional-runtime", + "risk": "optional adapter construction and validation against live projections", + "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_model_download_runtime.py"], + "conditional_gates": ["model-download-runtime"] + }, + { + "path": "obliteratus/sweep.py", + "risk_class": "conditional-runtime", + "risk": "parameter sweeps over repeated mutation and evaluation", + "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_model_download_runtime.py"], + "conditional_gates": ["model-download-runtime"] + }, + { + "path": "obliteratus/evaluation/heretic_eval.py", + "risk_class": "mixed-runtime", + "risk": "publication evaluation prompts, classifiers, scoring, and external adapters", + "required_tests": ["tests/test_heretic_eval.py", "tests/conditional/test_external_evaluation_runtime.py"], + "conditional_gates": ["external-evaluation"] + }, + { + "path": "obliteratus/evaluation/lm_eval_integration.py", + "risk_class": "conditional-runtime", + "risk": "lm-evaluation-harness integration and benchmark result translation", + "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_external_evaluation_runtime.py"], + "conditional_gates": ["external-evaluation"] + }, + { + "path": "obliteratus/tourney.py", + "risk_class": "conditional-runtime", + "risk": "multi-model tournament mutation, comparison, and optional publication", + "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_external_evaluation_runtime.py"], + "conditional_gates": ["external-evaluation"] + }, + { + "path": "obliteratus/bestiary_sync.py", + "risk_class": "conditional-runtime", + "risk": "external catalog synchronization and malformed service responses", + "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_network_services.py"], + "conditional_gates": ["network-services"] + }, + { + "path": "obliteratus/models_client.py", + "risk_class": "conditional-runtime", + "risk": "catalog resolution across operator files and network services", + "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_network_services.py"], + "conditional_gates": ["network-services"] + }, + { + "path": "obliteratus/watchtower.py", + "risk_class": "conditional-runtime", + "risk": "scheduled scans, queue state, and live service responses", + "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_network_services.py"], + "conditional_gates": ["network-services"] + }, + { + "path": "obliteratus/interactive.py", + "risk_class": "conditional-runtime", + "risk": "interactive terminal prompts and operator decision flow", + "required_tests": ["tests/test_cli.py", "tests/conditional/test_operator_ui.py"], + "conditional_gates": ["operator-ui"] + }, + { + "path": "obliteratus/local_ui.py", + "risk_class": "mixed-runtime", + "risk": "Gradio construction, launch configuration, authentication, and signals", + "required_tests": ["tests/test_cli.py", "tests/conditional/test_operator_ui.py"], + "conditional_gates": ["operator-ui"] + }, + { + "path": "obliteratus/ui_watchtower.py", + "risk_class": "conditional-runtime", + "risk": "service-backed UI tabs and scheduler controls", + "required_tests": ["tests/test_module_imports.py", "tests/conditional/test_operator_ui.py"], + "conditional_gates": ["operator-ui"] + }, + { + "path": "obliteratus/remote.py", + "risk_class": "mixed-runtime", + "risk": "SSH discovery, quoting, execution, cancellation, and result synchronization", + "required_tests": ["tests/test_remote_boundaries.py", "tests/conditional/test_remote_runtime.py"], + "conditional_gates": ["remote-execution"] + } + ] +} diff --git a/scripts/check_coverage_thresholds.py b/scripts/check_coverage_thresholds.py index 1d63db2..fd0f7c2 100644 --- a/scripts/check_coverage_thresholds.py +++ b/scripts/check_coverage_thresholds.py @@ -7,10 +7,28 @@ import json import math import re import subprocess +from dataclasses import dataclass from pathlib import Path from typing import Any +@dataclass(frozen=True) +class ChangedModule: + """One live production module changed between the base and head commits.""" + + status: str + base_path: str | None + head_path: str + + +@dataclass(frozen=True) +class ModuleCoverage: + """Unrounded line and branch percentages for one measured module.""" + + line: float + branch: float | None + + def validate_coverage( report: dict[str, Any], *, min_line: float, min_branch: float, file_floors: dict[str, float] | None = None, @@ -114,6 +132,166 @@ def validate_changed_coverage( return [] +def parse_changed_modules( + raw: bytes, *, roots: tuple[str, ...] = ("obliteratus",), +) -> list[ChangedModule]: + """Parse ``git diff --name-status -M -z`` for live Python modules.""" + + tokens = raw.decode("utf-8", "surrogateescape").split("\0") + if tokens and tokens[-1] == "": + tokens.pop() + + normalized_roots = tuple(root.rstrip("/") for root in roots if root.rstrip("/")) + + def included(path: str) -> bool: + return path.endswith(".py") and any( + path.startswith(f"{root}/") for root in normalized_roots + ) + + changes: list[ChangedModule] = [] + index = 0 + while index < len(tokens): + status = tokens[index] + index += 1 + if not status: + raise ValueError("changed-module diff contains an empty status") + kind = status[0] + if kind in {"R", "C"}: + if index + 1 >= len(tokens): + raise ValueError(f"changed-module diff is truncated after {status!r}") + old_path, new_path = tokens[index], tokens[index + 1] + index += 2 + if kind == "R" and included(old_path) and not included(new_path): + changes.append(ChangedModule("D", old_path, old_path)) + continue + if not included(new_path): + continue + base_path = old_path if kind == "R" and included(old_path) else None + changes.append(ChangedModule(status, base_path, new_path)) + continue + + if index >= len(tokens): + raise ValueError(f"changed-module diff is truncated after {status!r}") + path = tokens[index] + index += 1 + if not included(path): + continue + if kind == "D": + changes.append(ChangedModule(status, path, path)) + continue + if kind not in {"A", "M", "T"}: + raise ValueError(f"unsupported changed-module status {status!r}") + changes.append( + ChangedModule(status, None if kind == "A" else path, path), + ) + return changes + + +def module_coverage( + report: dict[str, Any], path: str, +) -> tuple[ModuleCoverage | None, str | None]: + """Return exact per-module line/branch coverage or a validation failure.""" + + files = report.get("files") + entry = files.get(path) if isinstance(files, dict) else None + if not isinstance(entry, dict): + return None, f"coverage report is missing touched module {path}" + summary = entry.get("summary") + if not isinstance(summary, dict): + return None, f"coverage report is missing summary for touched module {path}" + + values: dict[str, int] = {} + for key in ("covered_lines", "num_statements", "covered_branches", "num_branches"): + value = summary.get(key) + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + return None, f"coverage report has invalid {key} for touched module {path}" + values[key] = value + + if values["num_statements"] == 0: + return None, f"coverage report measures no statements for touched module {path}" + if values["covered_lines"] > values["num_statements"]: + return None, f"coverage report overcounts covered lines for touched module {path}" + if values["covered_branches"] > values["num_branches"]: + return None, f"coverage report overcounts covered branches for touched module {path}" + + line = values["covered_lines"] / values["num_statements"] * 100 + branch = ( + values["covered_branches"] / values["num_branches"] * 100 + if values["num_branches"] + else None + ) + return ModuleCoverage(line=line, branch=branch), None + + +def validate_touched_module_regression( + head_report: dict[str, Any], + base_report: dict[str, Any], + changes: list[ChangedModule], + *, + line_tolerance: float = 0.0, + branch_tolerance: float = 0.0, + new_module_min_line: float = 80.0, + new_module_min_branch: float = 75.0, +) -> list[str]: + """Reject line or branch regressions in each touched production module.""" + + def regressed(head: float, base: float, tolerance: float) -> bool: + return base - head > tolerance + 1e-9 + + failures: list[str] = [] + for change in changes: + if change.status.startswith("D"): + failures.append( + f"deleted production module {change.base_path} requires an explicit " + "reviewed coverage-policy exception", + ) + continue + head, head_failure = module_coverage(head_report, change.head_path) + if head_failure: + failures.append(head_failure) + continue + assert head is not None + + if change.base_path is None: + if head.line < new_module_min_line: + failures.append( + f"new module {change.head_path} line coverage {head.line:.2f}% " + f"is below the {new_module_min_line:.2f}% floor", + ) + if head.branch is not None and head.branch < new_module_min_branch: + failures.append( + f"new module {change.head_path} branch coverage {head.branch:.2f}% " + f"is below the {new_module_min_branch:.2f}% floor", + ) + continue + + base, base_failure = module_coverage(base_report, change.base_path) + if base_failure: + failures.append(f"base {base_failure}") + continue + assert base is not None + + if regressed(head.line, base.line, line_tolerance): + failures.append( + f"touched module {change.head_path} line coverage regressed " + f"from {base.line:.2f}% to {head.line:.2f}%", + ) + if base.branch is None and head.branch is not None: + if head.branch < new_module_min_branch: + failures.append( + f"touched module {change.head_path} added branches at " + f"{head.branch:.2f}% coverage, below the " + f"{new_module_min_branch:.2f}% floor", + ) + elif base.branch is not None and head.branch is not None: + if regressed(head.branch, base.branch, branch_tolerance): + failures.append( + f"touched module {change.head_path} branch coverage regressed " + f"from {base.branch:.2f}% to {head.branch:.2f}%", + ) + return failures + + def _file_floor(value: str) -> tuple[str, float]: try: path, minimum = value.rsplit("=", 1) @@ -142,6 +320,26 @@ def _parser() -> argparse.ArgumentParser: "--base-ref", help="git base commit/ref used to calculate changed executable lines", ) + parser.add_argument( + "--base-report", + type=Path, + help="coverage.py JSON report generated from the base commit", + ) + parser.add_argument( + "--touched-module-no-regression", + action="store_true", + help="compare each touched production module against --base-report", + ) + parser.add_argument( + "--module-root", + action="append", + default=[], + help="production module root for touched-module comparison (default: obliteratus)", + ) + parser.add_argument("--module-line-tolerance", type=float, default=0.0) + parser.add_argument("--module-branch-tolerance", type=float, default=0.0) + parser.add_argument("--new-module-min-line", type=float, default=80.0) + parser.add_argument("--new-module-min-branch", type=float, default=75.0) return parser @@ -180,6 +378,48 @@ def main() -> int: failures.extend( validate_changed_coverage(report, changed, minimum=args.min_changed), ) + touched_count: int | None = None + if args.touched_module_no_regression: + if not args.base_ref or args.base_report is None: + failures.append( + "touched-module comparison requires --base-ref and --base-report", + ) + elif args.module_line_tolerance < 0 or args.module_branch_tolerance < 0: + failures.append("touched-module tolerances must be non-negative") + else: + try: + base_report = json.loads(args.base_report.read_text(encoding="utf-8")) + pathspecs = [ + f":(glob){root.rstrip('/')}/**/*.py" + for root in (args.module_root or ["obliteratus"]) + ] + raw_changes = subprocess.run( + [ + "git", "diff", "--name-status", "-M", "-z", + args.base_ref, "HEAD", "--", *pathspecs, + ], + check=True, + capture_output=True, + ).stdout + module_changes = parse_changed_modules( + raw_changes, + roots=tuple(args.module_root or ["obliteratus"]), + ) + except (OSError, json.JSONDecodeError, subprocess.CalledProcessError, ValueError) as exc: + failures.append(f"cannot compare touched-module coverage: {exc}") + else: + touched_count = len(module_changes) + failures.extend( + validate_touched_module_regression( + report, + base_report, + module_changes, + line_tolerance=args.module_line_tolerance, + branch_tolerance=args.module_branch_tolerance, + new_module_min_line=args.new_module_min_line, + new_module_min_branch=args.new_module_min_branch, + ), + ) if failures: for failure in failures: print(f"coverage gate failed: {failure}") @@ -196,6 +436,8 @@ def main() -> int: print( f"changed-line gate passed: {percentage:.2f}% ({covered}/{executable})", ) + if touched_count is not None: + print(f"touched-module coverage gate passed: {touched_count} module(s)") return 0 diff --git a/scripts/check_quality_policy.py b/scripts/check_quality_policy.py index 7d95b1c..041bb8c 100644 --- a/scripts/check_quality_policy.py +++ b/scripts/check_quality_policy.py @@ -6,6 +6,8 @@ from __future__ import annotations import argparse import json import math +import re +from datetime import date, timedelta from pathlib import Path from typing import Any @@ -41,7 +43,125 @@ def _valid_exception(exceptions: Any, name: str, current: float) -> bool: return False -def validate_policy(policy: dict[str, Any]) -> list[str]: +def _policy_date(value: Any, label: str, failures: list[str]) -> date | None: + if not isinstance(value, str): + failures.append(f"{label} must be an ISO date") + return None + try: + return date.fromisoformat(value) + except ValueError: + failures.append(f"{label} must be an ISO date") + return None + + +def _validate_test_evidence( + policy: dict[str, Any], *, today: date, +) -> list[str]: + failures: list[str] = [] + evidence = policy.get("test_evidence") + if not isinstance(evidence, dict): + return ["quality policy is missing the test_evidence object"] + + numeric: dict[str, int] = {} + for key, minimum in ( + ("retention_days", 30), + ("flake_window_days", 30), + ("maximum_quarantine_days", 1), + ): + value = evidence.get(key) + if isinstance(value, bool) or not isinstance(value, int) or value < minimum: + failures.append(f"test evidence {key} must be an integer at least {minimum}") + else: + numeric[key] = value + + history = evidence.get("flake_history") + quarantines = evidence.get("quarantines") + if not isinstance(history, list): + failures.append("test evidence flake_history must be a list") + history = [] + if not isinstance(quarantines, list): + failures.append("test evidence quarantines must be a list") + quarantines = [] + + observations: dict[str, list[date]] = {} + history_keys: set[tuple[str, str, str]] = set() + for index, entry in enumerate(history): + label = f"flake history {index}" + if not isinstance(entry, dict): + failures.append(f"{label} must be an object") + continue + nodeid = entry.get("nodeid") + head_sha = entry.get("head_sha") + gate = entry.get("gate") + if not isinstance(nodeid, str) or not nodeid.strip(): + failures.append(f"{label} requires a non-empty nodeid") + continue + if not isinstance(head_sha, str) or re.fullmatch(r"[0-9a-f]{40}", head_sha) is None: + failures.append(f"{label} requires a 40-character head_sha") + if not isinstance(gate, str) or not gate.strip(): + failures.append(f"{label} requires a non-empty gate") + observed = _policy_date(entry.get("observed_on"), f"{label} observed_on", failures) + if observed is not None: + if observed > today: + failures.append(f"{label} observed_on cannot be in the future") + observations.setdefault(nodeid, []).append(observed) + key = (nodeid, str(entry.get("observed_on")), str(head_sha)) + if key in history_keys: + failures.append(f"duplicate flake history entry for {nodeid}") + history_keys.add(key) + + active_quarantines: set[str] = set() + quarantine_nodeids: set[str] = set() + max_days = numeric.get("maximum_quarantine_days", 30) + for index, entry in enumerate(quarantines): + label = f"test quarantine {index}" + if not isinstance(entry, dict): + failures.append(f"{label} must be an object") + continue + nodeid = entry.get("nodeid") + if not isinstance(nodeid, str) or not nodeid.strip(): + failures.append(f"{label} requires a non-empty nodeid") + continue + if nodeid in quarantine_nodeids: + failures.append(f"duplicate test quarantine for {nodeid}") + quarantine_nodeids.add(nodeid) + owner = entry.get("owner") + if not isinstance(owner, str) or not owner.startswith("@"): + failures.append(f"{label} requires an @owner") + if not isinstance(entry.get("reason"), str) or not entry["reason"].strip(): + failures.append(f"{label} requires a non-empty reason") + issue = entry.get("issue") + if not isinstance(issue, str) or not issue.startswith( + "https://github.com/elder-plinius/OBLITERATUS/issues/", + ): + failures.append(f"{label} requires an OBLITERATUS issue URL") + opened = _policy_date(entry.get("opened"), f"{label} opened", failures) + expires = _policy_date(entry.get("expires"), f"{label} expires", failures) + if opened is not None and expires is not None: + if opened > today: + failures.append(f"{label} cannot open in the future") + if expires <= opened: + failures.append(f"{label} must expire after it opens") + elif expires - opened > timedelta(days=max_days): + failures.append(f"{label} exceeds the {max_days}-day maximum") + elif expires < today: + failures.append(f"{label} expired on {expires.isoformat()}") + else: + active_quarantines.add(nodeid) + + window = numeric.get("flake_window_days", 30) + cutoff = today - timedelta(days=window - 1) + for nodeid, dates in observations.items(): + recent = [observed for observed in dates if cutoff <= observed <= today] + if len(recent) >= 2 and nodeid not in active_quarantines: + failures.append( + f"test {nodeid} flaked {len(recent)} times in {window} days " + "without an active quarantine", + ) + return failures + + +def validate_policy(policy: dict[str, Any], *, today: date | None = None) -> list[str]: """Validate policy structure, immutable floors, and exclusion traceability.""" failures: list[str] = [] minimums = policy.get("minimums") @@ -63,6 +183,19 @@ def validate_policy(policy: dict[str, Any]) -> list[str]: "without an explicit reviewed exception", ) + critical_paths = policy.get("critical_cpu_paths") + if not isinstance(critical_paths, list) or not critical_paths: + failures.append("quality policy requires non-empty critical_cpu_paths") + else: + seen_critical: set[str] = set() + for path in critical_paths: + if not isinstance(path, str) or not path.startswith("obliteratus/"): + failures.append(f"invalid critical CPU path: {path!r}") + elif path in seen_critical: + failures.append(f"critical CPU path is duplicated: {path}") + else: + seen_critical.add(path) + scope = policy.get("mature_cpu_scope") exclusions = scope.get("exclusions") if isinstance(scope, dict) else None if not isinstance(exclusions, list): @@ -88,6 +221,7 @@ def validate_policy(policy: dict[str, Any]) -> list[str]: issue = exclusion.get("conditional_issue") if issue != "https://github.com/elder-plinius/OBLITERATUS/issues/71": failures.append(f"{label} must link the conditional-test issue #71") + failures.extend(_validate_test_evidence(policy, today=today or date.today())) return failures diff --git a/scripts/check_test_risk_map.py b/scripts/check_test_risk_map.py new file mode 100644 index 0000000..87ed03f --- /dev/null +++ b/scripts/check_test_risk_map.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +"""Validate the source-to-test risk map and conditional coverage graph.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + + +RISK_CLASSES = {"cpu-contract", "mixed-runtime", "conditional-runtime"} + + +def _load_object(path: Path, label: str, errors: list[str]) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + errors.append(f"cannot read {label}: {exc}") + return {} + if not isinstance(value, dict): + errors.append(f"{label} root must be an object") + return {} + return value + + +def validate(risk_path: Path, quality_path: Path, conditional_path: Path) -> list[str]: + """Return structural and cross-policy failures for the risk map.""" + + errors: list[str] = [] + risk = _load_object(risk_path, "test risk map", errors) + quality = _load_object(quality_path, "quality policy", errors) + conditional = _load_object(conditional_path, "conditional policy", errors) + if errors: + return errors + + if risk.get("schema_version") != 1: + errors.append("test risk map schema_version must be 1") + if not isinstance(risk.get("owner"), str) or not risk["owner"].strip(): + errors.append("test risk map requires a non-empty owner") + + gates = conditional.get("gates") + gate_by_id = { + gate.get("id"): gate + for gate in gates if isinstance(gate, dict) and isinstance(gate.get("id"), str) + } if isinstance(gates, list) else {} + + modules = risk.get("modules") + if not isinstance(modules, list) or not modules: + return errors + ["test risk map modules must be a non-empty list"] + + module_by_path: dict[str, dict[str, Any]] = {} + for index, module in enumerate(modules): + label = f"risk module {index}" + if not isinstance(module, dict): + errors.append(f"{label} must be an object") + continue + path = module.get("path") + if not isinstance(path, str) or not path: + errors.append(f"{label} requires a non-empty path") + continue + if path in module_by_path: + errors.append(f"duplicate risk module path: {path}") + module_by_path[path] = module + if not Path(path).is_file(): + errors.append(f"risk module maps missing source path: {path}") + if module.get("risk_class") not in RISK_CLASSES: + errors.append(f"risk module {path} has invalid risk_class") + if not isinstance(module.get("risk"), str) or not module["risk"].strip(): + errors.append(f"risk module {path} requires a non-empty risk") + + tests = module.get("required_tests") + if not isinstance(tests, list) or not tests: + errors.append(f"risk module {path} requires at least one test") + else: + for test_path in tests: + if not isinstance(test_path, str) or not test_path.startswith("tests/"): + errors.append(f"risk module {path} has invalid test path: {test_path!r}") + elif not Path(test_path).is_file(): + errors.append(f"risk module {path} maps missing test path: {test_path}") + + module_gates = module.get("conditional_gates") + if not isinstance(module_gates, list): + errors.append(f"risk module {path} conditional_gates must be a list") + continue + if len(module_gates) != len(set(module_gates)): + errors.append(f"risk module {path} has duplicate conditional gates") + for gate_id in module_gates: + gate = gate_by_id.get(gate_id) + if gate is None: + errors.append(f"risk module {path} references unknown gate {gate_id}") + elif path not in gate.get("coverage_paths", []): + errors.append(f"risk module {path} is not covered by gate {gate_id}") + + exclusions = quality.get("mature_cpu_scope", {}).get("exclusions", []) + if not isinstance(exclusions, list): + errors.append("quality policy exclusions must be a list") + exclusions = [] + for exclusion in exclusions: + if not isinstance(exclusion, dict): + errors.append("quality policy exclusion must be an object") + continue + path = exclusion.get("path") + gate_id = exclusion.get("conditional_gate") + module = module_by_path.get(path) + if module is None: + errors.append(f"CPU exclusion is missing from test risk map: {path}") + elif gate_id not in module.get("conditional_gates", []): + errors.append(f"CPU exclusion {path} is missing conditional gate {gate_id}") + + critical_paths = quality.get("critical_cpu_paths", []) + if not isinstance(critical_paths, list): + errors.append("quality policy critical_cpu_paths must be a list") + critical_paths = [] + for path in critical_paths: + if path not in module_by_path: + errors.append(f"critical CPU path is missing from test risk map: {path}") + + for gate_id, gate in gate_by_id.items(): + for path in gate.get("coverage_paths", []): + module = module_by_path.get(path) + if module is None: + errors.append(f"conditional path is missing from test risk map: {path}") + elif gate_id not in module.get("conditional_gates", []): + errors.append(f"risk module {path} is missing conditional gate {gate_id}") + return errors + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--risk-map", type=Path, default=Path("ci/test-risk-map.json")) + parser.add_argument("--quality", type=Path, default=Path("ci/test-quality-policy.json")) + parser.add_argument( + "--conditional", + type=Path, + default=Path("ci/conditional-test-policy.json"), + ) + args = parser.parse_args() + errors = validate(args.risk_map, args.quality, args.conditional) + if errors: + for error in errors: + print(f"ERROR: {error}") + return 1 + print("test risk map: valid") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/run_repeat_gate.py b/scripts/run_repeat_gate.py index b83409b..dfe77f0 100644 --- a/scripts/run_repeat_gate.py +++ b/scripts/run_repeat_gate.py @@ -11,6 +11,7 @@ import sys import time from pathlib import Path from typing import Any, Sequence +from xml.etree import ElementTree DEFAULT_TESTS = ( @@ -32,6 +33,26 @@ def test_orders(paths: Sequence[str]) -> list[list[str]]: return [forward, reverse, interleaved] +def junit_snapshot(path: Path) -> dict[str, Any]: + """Return counts and stable failed/skipped node IDs for one repeat pass.""" + + root = ElementTree.parse(path).getroot() + cases = list(root.iter("testcase")) + failed: list[str] = [] + skipped: list[str] = [] + for case in cases: + nodeid = f"{case.attrib.get('classname', '')}::{case.attrib.get('name', '')}" + if case.find("failure") is not None or case.find("error") is not None: + failed.append(nodeid) + if case.find("skipped") is not None: + skipped.append(nodeid) + return { + "tests": len(cases), + "failed_nodeids": failed, + "skipped_nodeids": skipped, + } + + def run_repeat_gate( paths: Sequence[str], *, output: Path, python: str = sys.executable, ) -> int: @@ -44,7 +65,10 @@ def run_repeat_gate( zip(test_orders(paths), HASH_SEEDS, strict=True), start=1, ): - command = [python, "-m", "pytest", "--no-cov", "-q", *order] + junit = output.parent / f"repeat-pass-{index}.xml" + command = [ + python, "-m", "pytest", "--no-cov", "-q", f"--junitxml={junit}", *order, + ] environment = os.environ.copy() environment["PYTHONHASHSEED"] = hash_seed pass_started = time.monotonic() @@ -56,28 +80,60 @@ def run_repeat_gate( check=False, ) duration = time.monotonic() - pass_started - results.append({ + try: + counts = junit_snapshot(junit) + junit_error = None + except (OSError, ElementTree.ParseError) as exc: + counts = {"tests": 0, "failed_nodeids": [], "skipped_nodeids": []} + junit_error = str(exc) + result = { "pass": index, "python_hash_seed": hash_seed, "tests": order, "duration_seconds": round(duration, 3), "return_code": completed.returncode, + "junit": str(junit), + **counts, "stdout": completed.stdout[-4000:], "stderr": completed.stderr[-4000:], - }) - if completed.returncode != 0: + } + if junit_error is not None: + result["junit_error"] = junit_error + results.append(result) + if completed.returncode != 0 and exit_code == 0: exit_code = completed.returncode - break + if junit_error is not None and exit_code == 0: + exit_code = 1 + + occurrences: dict[str, int] = {} + for result in results: + for nodeid in result["failed_nodeids"]: + occurrences[nodeid] = occurrences.get(nodeid, 0) + 1 + flake_candidates = [ + {"nodeid": nodeid, "occurrences": count} + for nodeid, count in sorted(occurrences.items()) + if count < len(results) + ] + consistent_failures = [ + {"nodeid": nodeid, "occurrences": count} + for nodeid, count in sorted(occurrences.items()) + if count == len(results) + ] evidence = { "schema_version": 1, "status": "passed" if exit_code == 0 else "failed", "total_duration_seconds": round(time.monotonic() - started, 3), "passes": results, + "flake_candidates": flake_candidates, + "consistent_failures": consistent_failures, } output.write_text(json.dumps(evidence, indent=2) + "\n", encoding="utf-8") if exit_code: - failed = results[-1] + failed = next( + result for result in results + if result["return_code"] != 0 or result.get("junit_error") is not None + ) print( f"repeat gate failed on pass {failed['pass']} " f"with PYTHONHASHSEED={failed['python_hash_seed']}", diff --git a/scripts/write_test_evidence.py b/scripts/write_test_evidence.py new file mode 100644 index 0000000..e1bf710 --- /dev/null +++ b/scripts/write_test_evidence.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python3 +"""Normalize coverage, JUnit, repeat, and mutation results into retained evidence.""" + +from __future__ import annotations + +import argparse +import json +import math +import os +import platform +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from xml.etree import ElementTree + + +def _read_object(path: Path, label: str) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise ValueError(f"cannot read {label}: {exc}") from exc + if not isinstance(value, dict): + raise ValueError(f"{label} root must be an object") + return value + + +def _percentage(summary: dict[str, Any], covered: str, total: str, path: str) -> float | None: + covered_value = summary.get(covered) + total_value = summary.get(total) + for key, value in ((covered, covered_value), (total, total_value)): + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise ValueError(f"coverage report has invalid {key} for {path}") + if covered_value > total_value: + raise ValueError(f"coverage report overcounts {covered} for {path}") + return covered_value / total_value * 100 if total_value else None + + +def coverage_snapshot( + report: dict[str, Any], + risk_map: dict[str, Any] | None = None, + *, + allow_missing_risk_modules: bool = False, +) -> dict[str, Any]: + """Return repository and risk-module coverage metrics.""" + + totals = report.get("totals") + files = report.get("files") + if not isinstance(totals, dict) or not isinstance(files, dict): + raise ValueError("coverage report requires totals and files objects") + line = totals.get("percent_statements_covered") + branch = totals.get("percent_branches_covered") + for label, value in (("line", line), ("branch", branch)): + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + ): + raise ValueError(f"coverage report requires numeric {label} coverage") + + modules: dict[str, dict[str, float | None]] = {} + if risk_map is not None: + risk_modules = risk_map.get("modules") + if not isinstance(risk_modules, list): + raise ValueError("test risk map modules must be a list") + for item in risk_modules: + path = item.get("path") if isinstance(item, dict) else None + entry = files.get(path) if isinstance(path, str) else None + summary = entry.get("summary") if isinstance(entry, dict) else None + if not isinstance(path, str): + raise ValueError("test risk map module requires a path") + if not isinstance(summary, dict) and allow_missing_risk_modules: + modules[path] = {"line_percent": None, "branch_percent": None} + continue + if not isinstance(summary, dict): + raise ValueError(f"coverage report is missing risk module {path}") + modules[path] = { + "line_percent": _percentage(summary, "covered_lines", "num_statements", path), + "branch_percent": _percentage( + summary, "covered_branches", "num_branches", path, + ), + } + return { + "line_percent": float(line), + "branch_percent": float(branch), + "risk_modules": modules, + } + + +def junit_snapshot(path: Path) -> dict[str, Any]: + """Return stable counts, node IDs, and slow-test evidence from JUnit XML.""" + + try: + root = ElementTree.parse(path).getroot() + except (OSError, ElementTree.ParseError) as exc: + raise ValueError(f"cannot read JUnit XML: {exc}") from exc + cases = list(root.iter("testcase")) + failures: list[str] = [] + errors: list[str] = [] + skipped: list[str] = [] + durations: list[dict[str, str | float]] = [] + for case in cases: + nodeid = f"{case.attrib.get('classname', '')}::{case.attrib.get('name', '')}" + try: + duration = float(case.attrib.get("time", "0")) + except ValueError as exc: + raise ValueError(f"JUnit testcase {nodeid} has invalid time") from exc + durations.append({"nodeid": nodeid, "seconds": duration}) + if case.find("failure") is not None: + failures.append(nodeid) + if case.find("error") is not None: + errors.append(nodeid) + if case.find("skipped") is not None: + skipped.append(nodeid) + failed = set(failures) | set(errors) | set(skipped) + durations.sort(key=lambda item: (-float(item["seconds"]), str(item["nodeid"]))) + return { + "total": len(cases), + "passed": len(cases) - len(failed), + "failures": failures, + "errors": errors, + "skipped": skipped, + "duration_seconds": round(sum(float(item["seconds"]) for item in durations), 3), + "slowest": durations[:20], + } + + +def mutation_snapshot(stats: dict[str, Any]) -> dict[str, int | float]: + killed = stats.get("killed") + total = stats.get("total") + if ( + isinstance(killed, bool) + or not isinstance(killed, int) + or killed < 0 + or isinstance(total, bool) + or not isinstance(total, int) + or total <= 0 + or killed > total + ): + raise ValueError("mutation statistics require valid killed and total counts") + return {"killed": killed, "total": total, "score_percent": killed / total * 100} + + +def build_evidence( + *, + head_sha: str, + base_sha: str, + python_version: str, + coverage: dict[str, Any] | None = None, + base_coverage: dict[str, Any] | None = None, + risk_map: dict[str, Any] | None = None, + junit: dict[str, Any] | None = None, + repeat: dict[str, Any] | None = None, + mutation: dict[str, Any] | None = None, + generated_at: str | None = None, +) -> dict[str, Any]: + """Build the project-owned evidence schema from already-parsed inputs.""" + + evidence: dict[str, Any] = { + "schema_version": 1, + "generated_at": generated_at or datetime.now(timezone.utc).isoformat(), + "repository": os.environ.get("GITHUB_REPOSITORY", "elder-plinius/OBLITERATUS"), + "head_sha": head_sha, + "base_sha": base_sha, + "python": python_version, + "platform": platform.platform(), + } + if coverage is not None: + current = coverage_snapshot(coverage, risk_map) + coverage_evidence: dict[str, Any] = {"current": current} + if base_coverage is not None: + base = coverage_snapshot( + base_coverage, + risk_map, + allow_missing_risk_modules=True, + ) + coverage_evidence["base"] = base + coverage_evidence["delta"] = { + "line_percentage_points": current["line_percent"] - base["line_percent"], + "branch_percentage_points": current["branch_percent"] - base["branch_percent"], + } + evidence["coverage"] = coverage_evidence + if junit is not None: + evidence["tests"] = junit + if repeat is not None: + if repeat.get("schema_version") != 1: + raise ValueError("repeat evidence schema_version must be 1") + evidence["repeat"] = repeat + if mutation is not None: + evidence["mutation"] = mutation_snapshot(mutation) + if not any(key in evidence for key in ("coverage", "tests", "repeat", "mutation")): + raise ValueError("at least one test evidence input is required") + return evidence + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--coverage", type=Path) + parser.add_argument("--base-coverage", type=Path) + parser.add_argument("--junit", type=Path) + parser.add_argument("--repeat", type=Path) + parser.add_argument("--mutation", type=Path) + parser.add_argument("--risk-map", type=Path, default=Path("ci/test-risk-map.json")) + parser.add_argument("--head-sha", default=os.environ.get("GITHUB_SHA", "local")) + parser.add_argument("--base-sha", default="") + parser.add_argument("--python-version", default=platform.python_version()) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + try: + coverage = _read_object(args.coverage, "coverage report") if args.coverage else None + base_coverage = ( + _read_object(args.base_coverage, "base coverage report") + if args.base_coverage else None + ) + risk_map = _read_object(args.risk_map, "test risk map") if coverage else None + junit = junit_snapshot(args.junit) if args.junit else None + repeat = _read_object(args.repeat, "repeat evidence") if args.repeat else None + mutation = _read_object(args.mutation, "mutation evidence") if args.mutation else None + evidence = build_evidence( + head_sha=args.head_sha, + base_sha=args.base_sha, + python_version=args.python_version, + coverage=coverage, + base_coverage=base_coverage, + risk_map=risk_map, + junit=junit, + repeat=repeat, + mutation=mutation, + ) + except ValueError as exc: + print(f"test evidence failed: {exc}") + return 1 + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(evidence, indent=2, sort_keys=True) + "\n", encoding="utf-8") + print(f"test evidence written: {args.output}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_ci_policy.py b/tests/test_ci_policy.py index 07da397..409fe47 100644 --- a/tests/test_ci_policy.py +++ b/tests/test_ci_policy.py @@ -69,3 +69,21 @@ def test_ci_requires_the_committed_lock_and_strict_policy_gate(): assert "scripts/check_supply_chain_policy.py licenses" in workflow supply_chain_job = workflow.split(" supply-chain:\n", maxsplit=1)[1] assert "|| true" not in supply_chain_job + + +def test_ci_enforces_exact_base_module_regression_and_risk_mapping(): + workflow = WORKFLOW.read_text(encoding="utf-8") + + assert "Generate exact-base coverage for module regression comparison" in workflow + assert 'git worktree add --detach "$BASE_WORKTREE" "$COVERAGE_BASE"' in workflow + assert "--touched-module-no-regression" in workflow + assert "--base-report test-results/base-coverage-py3.12.json" in workflow + assert "scripts/check_test_risk_map.py" in workflow + + +def test_ci_retains_normalized_test_and_quality_trends_for_ninety_days(): + workflow = WORKFLOW.read_text(encoding="utf-8") + + assert "test-trend-py${{ matrix.python-version }}.json" in workflow + assert "quality-trend-py3.12.json" in workflow + assert workflow.count("retention-days: 90") >= 2 diff --git a/tests/test_coverage_thresholds.py b/tests/test_coverage_thresholds.py index e1d3289..b39feff 100644 --- a/tests/test_coverage_thresholds.py +++ b/tests/test_coverage_thresholds.py @@ -165,3 +165,142 @@ def test_changed_line_gate_accepts_exact_floor(): assert MODULE.validate_changed_coverage( _report(), changed, minimum=75.0, ) == [] + + +def _module_report( + path: str = "obliteratus/example.py", + *, + covered_lines: int = 8, + statements: int = 10, + covered_branches: int = 3, + branches: int = 4, +) -> dict[str, object]: + return { + "files": { + path: { + "summary": { + "covered_lines": covered_lines, + "num_statements": statements, + "covered_branches": covered_branches, + "num_branches": branches, + }, + }, + }, + } + + +def test_parse_changed_modules_handles_nul_renames_and_filters_nonproduction(): + raw = ( + b"M\0obliteratus/current.py\0" + b"A\0tests/test_current.py\0" + b"D\0obliteratus/deleted.py\0" + b"R100\0obliteratus/old.py\0obliteratus/new.py\0" + b"R100\0obliteratus/removed.py\0docs/removed.py\0" + b"C100\0obliteratus/source.py\0obliteratus/copied.py\0" + ) + + assert MODULE.parse_changed_modules(raw) == [ + MODULE.ChangedModule("M", "obliteratus/current.py", "obliteratus/current.py"), + MODULE.ChangedModule("D", "obliteratus/deleted.py", "obliteratus/deleted.py"), + MODULE.ChangedModule("R100", "obliteratus/old.py", "obliteratus/new.py"), + MODULE.ChangedModule("D", "obliteratus/removed.py", "obliteratus/removed.py"), + MODULE.ChangedModule("C100", None, "obliteratus/copied.py"), + ] + + +def test_parse_changed_modules_rejects_truncated_and_unknown_statuses(): + import pytest + + with pytest.raises(ValueError, match="truncated"): + MODULE.parse_changed_modules(b"R100\0obliteratus/old.py\0") + with pytest.raises(ValueError, match="unsupported"): + MODULE.parse_changed_modules(b"U\0obliteratus/example.py\0") + + +def test_touched_module_gate_reports_line_and_branch_regressions_separately(): + base = _module_report(covered_lines=9, covered_branches=4) + head = _module_report(covered_lines=8, covered_branches=3) + changes = [MODULE.ChangedModule("M", "obliteratus/example.py", "obliteratus/example.py")] + + assert MODULE.validate_touched_module_regression(head, base, changes) == [ + "touched module obliteratus/example.py line coverage regressed from 90.00% to 80.00%", + "touched module obliteratus/example.py branch coverage regressed from 100.00% to 75.00%", + ] + + +def test_touched_module_gate_applies_tolerances_to_existing_module(): + base = _module_report(covered_lines=801, statements=1000, covered_branches=751, branches=1000) + head = _module_report(covered_lines=800, statements=1000, covered_branches=750, branches=1000) + changes = [MODULE.ChangedModule("M", "obliteratus/example.py", "obliteratus/example.py")] + + assert MODULE.validate_touched_module_regression( + head, + base, + changes, + line_tolerance=0.1, + branch_tolerance=0.1, + ) == [] + assert len(MODULE.validate_touched_module_regression( + head, + base, + changes, + line_tolerance=0.09, + branch_tolerance=0.09, + )) == 2 + + +def test_touched_module_gate_enforces_new_module_floors(): + head = _module_report(covered_lines=7, statements=10, covered_branches=2, branches=4) + changes = [MODULE.ChangedModule("A", None, "obliteratus/example.py")] + + assert MODULE.validate_touched_module_regression(head, {}, changes) == [ + "new module obliteratus/example.py line coverage 70.00% is below the 80.00% floor", + "new module obliteratus/example.py branch coverage 50.00% is below the 75.00% floor", + ] + + +def test_touched_module_gate_requires_floor_when_existing_module_adds_branches(): + base = _module_report(covered_branches=0, branches=0) + head = _module_report(covered_branches=2, branches=4) + changes = [MODULE.ChangedModule("M", "obliteratus/example.py", "obliteratus/example.py")] + + assert MODULE.validate_touched_module_regression(head, base, changes) == [ + "touched module obliteratus/example.py added branches at 50.00% coverage, " + "below the 75.00% floor", + ] + + +def test_touched_module_gate_accepts_branch_removal_and_rename(): + base = _module_report("obliteratus/old.py", covered_branches=3, branches=4) + head = _module_report("obliteratus/new.py", covered_branches=0, branches=0) + changes = [MODULE.ChangedModule("R100", "obliteratus/old.py", "obliteratus/new.py")] + + assert MODULE.validate_touched_module_regression(head, base, changes) == [] + + +def test_touched_module_gate_rejects_missing_and_malformed_measurements(): + changes = [MODULE.ChangedModule("M", "obliteratus/example.py", "obliteratus/example.py")] + assert MODULE.validate_touched_module_regression({}, _module_report(), changes) == [ + "coverage report is missing touched module obliteratus/example.py", + ] + + malformed = _module_report() + malformed["files"]["obliteratus/example.py"]["summary"]["covered_lines"] = True + assert MODULE.validate_touched_module_regression(malformed, _module_report(), changes) == [ + "coverage report has invalid covered_lines for touched module obliteratus/example.py", + ] + + +def test_touched_module_gate_rejects_missing_base_measurement(): + changes = [MODULE.ChangedModule("M", "obliteratus/example.py", "obliteratus/example.py")] + assert MODULE.validate_touched_module_regression(_module_report(), {}, changes) == [ + "base coverage report is missing touched module obliteratus/example.py", + ] + + +def test_touched_module_gate_requires_reviewed_exception_for_deletion(): + changes = [MODULE.ChangedModule("D", "obliteratus/example.py", "obliteratus/example.py")] + assert MODULE.validate_touched_module_regression({}, _module_report(), changes) == [ + "deleted production module obliteratus/example.py requires an explicit reviewed " + "coverage-policy exception", + ] diff --git a/tests/test_quality_gate_scripts.py b/tests/test_quality_gate_scripts.py index b015688..a973bfd 100644 --- a/tests/test_quality_gate_scripts.py +++ b/tests/test_quality_gate_scripts.py @@ -43,6 +43,11 @@ def test_repeat_gate_records_each_pass(monkeypatch, tmp_path): return SimpleNamespace(returncode=0, stdout="2 passed\n", stderr="") monkeypatch.setattr(run_repeat_gate.subprocess, "run", fake_run) + monkeypatch.setattr( + run_repeat_gate, + "junit_snapshot", + lambda _path: {"tests": 2, "failed_nodeids": [], "skipped_nodeids": []}, + ) output = tmp_path / "repeat.json" assert run_repeat_gate.run_repeat_gate( ["first.py", "second.py"], output=output, python="python-fixture", @@ -53,18 +58,61 @@ def test_repeat_gate_records_each_pass(monkeypatch, tmp_path): "0", "1", "8675309", ] assert calls[0][0] == [ - "python-fixture", "-m", "pytest", "--no-cov", "-q", "first.py", "second.py", + "python-fixture", + "-m", + "pytest", + "--no-cov", + "-q", + f"--junitxml={tmp_path / 'repeat-pass-1.xml'}", + "first.py", + "second.py", ] -def test_repeat_gate_stops_and_preserves_failure_output(monkeypatch, tmp_path): +def test_repeat_gate_completes_all_passes_and_preserves_failure_output(monkeypatch, tmp_path): def fake_run(*_args, **_kwargs): return SimpleNamespace(returncode=3, stdout="failed output", stderr="failure detail") monkeypatch.setattr(run_repeat_gate.subprocess, "run", fake_run) + monkeypatch.setattr( + run_repeat_gate, + "junit_snapshot", + lambda _path: { + "tests": 1, + "failed_nodeids": ["tests.test_example::test_failure"], + "skipped_nodeids": [], + }, + ) output = tmp_path / "repeat.json" assert run_repeat_gate.run_repeat_gate(["test.py"], output=output) == 3 evidence = json.loads(output.read_text()) assert evidence["status"] == "failed" - assert len(evidence["passes"]) == 1 + assert len(evidence["passes"]) == 3 assert evidence["passes"][0]["stderr"] == "failure detail" + assert evidence["consistent_failures"] == [{ + "nodeid": "tests.test_example::test_failure", + "occurrences": 3, + }] + + +def test_repeat_gate_identifies_intermittent_failures(monkeypatch, tmp_path): + results = iter([ + SimpleNamespace(returncode=1, stdout="failed", stderr=""), + SimpleNamespace(returncode=0, stdout="passed", stderr=""), + SimpleNamespace(returncode=1, stdout="failed", stderr=""), + ]) + snapshots = iter([ + {"tests": 1, "failed_nodeids": ["tests.test_example::test_flake"], "skipped_nodeids": []}, + {"tests": 1, "failed_nodeids": [], "skipped_nodeids": []}, + {"tests": 1, "failed_nodeids": ["tests.test_example::test_flake"], "skipped_nodeids": []}, + ]) + monkeypatch.setattr(run_repeat_gate.subprocess, "run", lambda *_args, **_kwargs: next(results)) + monkeypatch.setattr(run_repeat_gate, "junit_snapshot", lambda _path: next(snapshots)) + + output = tmp_path / "repeat.json" + assert run_repeat_gate.run_repeat_gate(["test.py"], output=output) == 1 + evidence = json.loads(output.read_text()) + assert evidence["flake_candidates"] == [{ + "nodeid": "tests.test_example::test_flake", + "occurrences": 2, + }] diff --git a/tests/test_quality_policy.py b/tests/test_quality_policy.py index 7d7c2ec..3b2d170 100644 --- a/tests/test_quality_policy.py +++ b/tests/test_quality_policy.py @@ -3,6 +3,7 @@ from __future__ import annotations from copy import deepcopy +from datetime import date from scripts import check_quality_policy as quality @@ -10,6 +11,7 @@ from scripts import check_quality_policy as quality def _policy(): return { "minimums": dict(quality.BASELINE_FLOORS), + "critical_cpu_paths": ["obliteratus/pure.py"], "mature_cpu_scope": { "exclusions": [{ "path": "obliteratus/external.py", @@ -19,6 +21,13 @@ def _policy(): "conditional_gate": "network-services", }], }, + "test_evidence": { + "retention_days": 90, + "flake_window_days": 30, + "maximum_quarantine_days": 30, + "flake_history": [], + "quarantines": [], + }, "threshold_exceptions": [], } @@ -94,3 +103,71 @@ def test_mature_scope_rejects_regression_and_stale_exclusion(): assert failures == [ "coverage report is missing excluded source file obliteratus/external.py", ] + + +def test_second_flake_in_window_requires_active_quarantine(): + policy = _policy() + nodeid = "tests.test_example::test_unstable" + policy["test_evidence"]["flake_history"] = [ + { + "nodeid": nodeid, + "observed_on": "2026-08-01", + "head_sha": "a" * 40, + "gate": "repeat", + }, + { + "nodeid": nodeid, + "observed_on": "2026-08-14", + "head_sha": "b" * 40, + "gate": "mandatory-cpu", + }, + ] + + assert quality.validate_policy(policy, today=date(2026, 8, 14)) == [ + f"test {nodeid} flaked 2 times in 30 days without an active quarantine", + ] + + policy["test_evidence"]["quarantines"] = [{ + "nodeid": nodeid, + "owner": "@maintainers", + "reason": "Ordering-sensitive global state is being isolated.", + "issue": "https://github.com/elder-plinius/OBLITERATUS/issues/999", + "opened": "2026-08-14", + "expires": "2026-09-13", + }] + assert quality.validate_policy(policy, today=date(2026, 8, 14)) == [] + + +def test_quarantine_requires_bounded_owned_issue_linked_entry(): + policy = _policy() + policy["test_evidence"]["quarantines"] = [{ + "nodeid": "tests.test_example::test_unstable", + "owner": "maintainers", + "reason": "", + "issue": "https://example.com/issue/1", + "opened": "2026-08-01", + "expires": "2026-10-01", + }] + + failures = quality.validate_policy(policy, today=date(2026, 8, 14)) + assert "test quarantine 0 requires an @owner" in failures + assert "test quarantine 0 requires a non-empty reason" in failures + assert "test quarantine 0 requires an OBLITERATUS issue URL" in failures + assert "test quarantine 0 exceeds the 30-day maximum" in failures + + +def test_flake_history_rejects_malformed_duplicate_and_future_entries(): + policy = _policy() + entry = { + "nodeid": "tests.test_example::test_unstable", + "observed_on": "2026-08-15", + "head_sha": "short", + "gate": "", + } + policy["test_evidence"]["flake_history"] = [entry, deepcopy(entry)] + + failures = quality.validate_policy(policy, today=date(2026, 8, 14)) + assert "flake history 0 requires a 40-character head_sha" in failures + assert "flake history 0 requires a non-empty gate" in failures + assert "flake history 0 observed_on cannot be in the future" in failures + assert "duplicate flake history entry for tests.test_example::test_unstable" in failures diff --git a/tests/test_test_evidence.py b/tests/test_test_evidence.py new file mode 100644 index 0000000..b54cf86 --- /dev/null +++ b/tests/test_test_evidence.py @@ -0,0 +1,129 @@ +"""Tests for normalized retained test and quality evidence.""" + +from __future__ import annotations + +import pytest + +from scripts import write_test_evidence + + +def _coverage(line: float = 80.0, branch: float = 75.0): + return { + "totals": { + "percent_statements_covered": line, + "percent_branches_covered": branch, + }, + "files": { + "obliteratus/example.py": { + "summary": { + "covered_lines": 8, + "num_statements": 10, + "covered_branches": 3, + "num_branches": 4, + }, + }, + }, + } + + +def _risk_map(): + return {"modules": [{"path": "obliteratus/example.py"}]} + + +def test_coverage_snapshot_records_repository_and_risk_module_metrics(): + snapshot = write_test_evidence.coverage_snapshot(_coverage(), _risk_map()) + + assert snapshot == { + "line_percent": 80.0, + "branch_percent": 75.0, + "risk_modules": { + "obliteratus/example.py": { + "line_percent": 80.0, + "branch_percent": 75.0, + }, + }, + } + + +def test_junit_snapshot_records_failures_skips_and_slowest_tests(tmp_path): + junit = tmp_path / "junit.xml" + junit.write_text( + '' + '' + '' + '' + '' + '', + ) + + snapshot = write_test_evidence.junit_snapshot(junit) + assert snapshot["total"] == 3 + assert snapshot["passed"] == 1 + assert snapshot["failures"] == ["tests.test_a::test_fail"] + assert snapshot["skipped"] == ["tests.test_b::test_skip"] + assert snapshot["duration_seconds"] == 0.6 + assert snapshot["slowest"][0] == { + "nodeid": "tests.test_a::test_fail", + "seconds": 0.3, + } + + +def test_evidence_records_base_deltas_repeat_and_mutation(): + evidence = write_test_evidence.build_evidence( + head_sha="b" * 40, + base_sha="a" * 40, + python_version="3.12", + coverage=_coverage(80, 75), + base_coverage=_coverage(79, 74), + risk_map=_risk_map(), + repeat={"schema_version": 1, "status": "passed", "passes": []}, + mutation={"killed": 8, "total": 10}, + generated_at="2026-08-14T00:00:00+00:00", + ) + + assert evidence["coverage"]["delta"] == { + "line_percentage_points": 1.0, + "branch_percentage_points": 1.0, + } + assert evidence["repeat"]["status"] == "passed" + assert evidence["mutation"] == { + "killed": 8, + "total": 10, + "score_percent": 80.0, + } + + +def test_evidence_rejects_missing_inputs_and_malformed_measurements(): + with pytest.raises(ValueError, match="at least one"): + write_test_evidence.build_evidence( + head_sha="local", + base_sha="", + python_version="3.12", + ) + + malformed = _coverage() + malformed["files"]["obliteratus/example.py"]["summary"]["covered_lines"] = True + with pytest.raises(ValueError, match="invalid covered_lines"): + write_test_evidence.coverage_snapshot(malformed, _risk_map()) + + malformed = _coverage(line=float("nan")) + with pytest.raises(ValueError, match="numeric line"): + write_test_evidence.coverage_snapshot(malformed, _risk_map()) + + +def test_base_snapshot_allows_risk_module_added_by_head(): + snapshot = write_test_evidence.coverage_snapshot( + {"totals": _coverage()["totals"], "files": {}}, + _risk_map(), + allow_missing_risk_modules=True, + ) + + assert snapshot["risk_modules"]["obliteratus/example.py"] == { + "line_percent": None, + "branch_percent": None, + } + + +def test_mutation_snapshot_rejects_impossible_counts(): + with pytest.raises(ValueError, match="valid killed and total"): + write_test_evidence.mutation_snapshot({"killed": 2, "total": 1}) diff --git a/tests/test_test_risk_map.py b/tests/test_test_risk_map.py new file mode 100644 index 0000000..e508246 --- /dev/null +++ b/tests/test_test_risk_map.py @@ -0,0 +1,96 @@ +"""Executable contracts for the source-to-test risk map.""" + +from __future__ import annotations + +import json +from copy import deepcopy +from pathlib import Path + +from scripts import check_test_risk_map + + +ROOT = Path(__file__).parents[1] + + +def _write(path: Path, value: object) -> Path: + path.write_text(json.dumps(value), encoding="utf-8") + return path + + +def test_committed_test_risk_map_is_complete(): + assert check_test_risk_map.validate( + ROOT / "ci" / "test-risk-map.json", + ROOT / "ci" / "test-quality-policy.json", + ROOT / "ci" / "conditional-test-policy.json", + ) == [] + + +def test_risk_map_rejects_duplicate_missing_and_unowned_modules(tmp_path): + risk = json.loads((ROOT / "ci" / "test-risk-map.json").read_text()) + risk["owner"] = "" + duplicate = deepcopy(risk["modules"][0]) + duplicate["required_tests"] = ["tests/missing.py"] + duplicate["risk_class"] = "unknown" + risk["modules"].append(duplicate) + + errors = check_test_risk_map.validate( + _write(tmp_path / "risk.json", risk), + ROOT / "ci" / "test-quality-policy.json", + ROOT / "ci" / "conditional-test-policy.json", + ) + + assert "test risk map requires a non-empty owner" in errors + assert "duplicate risk module path: obliteratus/cli.py" in errors + assert "risk module obliteratus/cli.py has invalid risk_class" in errors + assert "risk module obliteratus/cli.py maps missing test path: tests/missing.py" in errors + + +def test_risk_map_rejects_missing_exclusion_and_gate_mapping(tmp_path): + risk = json.loads((ROOT / "ci" / "test-risk-map.json").read_text()) + risk["modules"] = [ + module for module in risk["modules"] if module["path"] != "obliteratus/remote.py" + ] + loader = next( + module for module in risk["modules"] if module["path"] == "obliteratus/models/loader.py" + ) + loader["conditional_gates"].remove("cuda-runtime") + + errors = check_test_risk_map.validate( + _write(tmp_path / "risk.json", risk), + ROOT / "ci" / "test-quality-policy.json", + ROOT / "ci" / "conditional-test-policy.json", + ) + + assert "CPU exclusion is missing from test risk map: obliteratus/remote.py" in errors + assert "conditional path is missing from test risk map: obliteratus/remote.py" in errors + assert "risk module obliteratus/models/loader.py is missing conditional gate cuda-runtime" in errors + + +def test_risk_map_rejects_gate_that_does_not_cover_module(tmp_path): + risk = json.loads((ROOT / "ci" / "test-risk-map.json").read_text()) + cli = next(module for module in risk["modules"] if module["path"] == "obliteratus/cli.py") + cli["conditional_gates"] = ["remote-execution", "unknown-gate"] + + errors = check_test_risk_map.validate( + _write(tmp_path / "risk.json", risk), + ROOT / "ci" / "test-quality-policy.json", + ROOT / "ci" / "conditional-test-policy.json", + ) + + assert "risk module obliteratus/cli.py is not covered by gate remote-execution" in errors + assert "risk module obliteratus/cli.py references unknown gate unknown-gate" in errors + + +def test_risk_map_cannot_drop_critical_cpu_path(tmp_path): + risk = json.loads((ROOT / "ci" / "test-risk-map.json").read_text()) + risk["modules"] = [ + module for module in risk["modules"] if module["path"] != "obliteratus/cli.py" + ] + + errors = check_test_risk_map.validate( + _write(tmp_path / "risk.json", risk), + ROOT / "ci" / "test-quality-policy.json", + ROOT / "ci" / "conditional-test-policy.json", + ) + + assert "critical CPU path is missing from test risk map: obliteratus/cli.py" in errors