ci: enforce targeted coverage floors (#246) (#256)

Fixes #246

Author: codeforester

.github/workflows/tests.yml

@@ -69,7 +69,9 @@ jobs:
           python scripts/benchmark_runtime.py --check
           python -m compileall -q examples
       - name: Run tests with coverage threshold
-        run: python -m pytest --cov=base_cli --cov-report=term-missing --cov-fail-under=80
+        run: python -m pytest --cov=base_cli --cov-report=term-missing --cov-report=json:coverage.json --cov-fail-under=80
+      - name: Enforce high-risk module coverage floors
+        run: python scripts/validate_coverage.py coverage.json
       - name: Run static security checks
         run: |
           bandit -q -r lib/python/base_cli scripts -lll -iii

.gitignore

@@ -6,3 +6,4 @@ __pycache__/
 dist/
 build/
 *.egg-info/
+coverage.json

CHANGELOG.md

@@ -13,6 +13,8 @@ and versions are tracked in the repo-root `VERSION` file.
 
 ### Changed
 
+- Add branch-aware coverage reporting with documented floors for lifecycle,
+  filesystem, compatibility, history, and contract modules.
 - Decompose the application implementation behind a compatibility facade into
   focused core, lifecycle-installation, attachment, and invocation-runner
   modules without changing the public import surface.

docs/coverage-policy.md

@@ -0,0 +1,22 @@
+# Coverage policy
+
+The aggregate test gate requires at least 80% branch-aware coverage. The
+following high-risk modules also have explicit floors because regressions in
+these paths affect filesystem safety, lifecycle attachment, or machine-facing
+contracts:
+
+| Module | Floor |
+| --- | ---: |
+| `base_cli._attach` | 75% |
+| `base_cli._click_compat` | 80% |
+| `base_cli._private_files` | 75% |
+| `base_cli._runtime` | 80% |
+| `base_cli.command_protocol` | 85% |
+| `base_cli.history` | 75% |
+| `base_cli.redaction` | 90% |
+
+CI writes a branch-aware `coverage.json` report and runs
+`scripts/validate_coverage.py` against these floors. A floor change belongs in
+the same pull request as the tests that justify it. Platform-specific tests
+must cover both the native path and the safe fallback where the operating
+system provides a different filesystem primitive.

mkdocs.yml

@@ -49,6 +49,7 @@ nav:
       - Typer adapter: typer-adapter.md
       - Optional integrations: integrations.md
   - Operations and security:
+      - Coverage policy: coverage-policy.md
       - Local configuration: local-config.md
       - Cache ownership and layout: cache-ownership-and-layout.md
       - Performance: performance.md

scripts/validate_coverage.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python3
+"""Enforce coverage floors for security- and contract-sensitive modules."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any, NoReturn
+
+# These modules sit on filesystem, lifecycle, and machine-contract boundaries.
+# Their floors are intentionally stricter than the aggregate project gate.
+MODULE_FLOORS: dict[str, float] = {
+    "lib/python/base_cli/_attach.py": 75.0,
+    "lib/python/base_cli/_click_compat.py": 80.0,
+    "lib/python/base_cli/_private_files.py": 75.0,
+    "lib/python/base_cli/_runtime.py": 80.0,
+    "lib/python/base_cli/command_protocol.py": 85.0,
+    "lib/python/base_cli/history.py": 75.0,
+    "lib/python/base_cli/redaction.py": 90.0,
+}
+
+
+def fail(message: str) -> NoReturn:
+    print(f"coverage policy failed: {message}", file=sys.stderr)
+    raise SystemExit(1)
+
+
+def _file_summary(files: dict[str, Any], expected_path: str) -> dict[str, Any]:
+    for path, payload in files.items():
+        if path == expected_path or path.endswith(expected_path):
+            summary = payload.get("summary")
+            if isinstance(summary, dict):
+                return summary
+    fail(f"coverage report is missing {expected_path}")
+
+
+def validate(report_path: Path) -> None:
+    try:
+        report = json.loads(report_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        fail(f"cannot read {report_path}: {exc}")
+    files = report.get("files") if isinstance(report, dict) else None
+    if not isinstance(files, dict):
+        fail(f"{report_path} does not contain a coverage 'files' mapping")
+
+    failures: list[str] = []
+    for path, floor in MODULE_FLOORS.items():
+        summary = _file_summary(files, path)
+        covered = summary.get("percent_covered")
+        if not isinstance(covered, (int, float)):
+            failures.append(f"{path}: missing percent_covered")
+        elif covered < floor:
+            failures.append(f"{path}: {covered:.2f}% < {floor:.2f}%")
+    if failures:
+        fail("; ".join(failures))
+
+    for path, floor in MODULE_FLOORS.items():
+        summary = _file_summary(files, path)
+        print(f"{path}: {summary['percent_covered']:.2f}% (floor {floor:.2f}%)")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("report", type=Path, help="coverage.py JSON report")
+    args = parser.parse_args()
+    validate(args.report)
+
+
+if __name__ == "__main__":
+    main()

scripts/validate_docs.py

@@ -18,6 +18,7 @@
         "api-stability.md",
         "cache-ownership-and-layout.md",
         "consumer-profiles.md",
+        "coverage-policy.md",
         "dependency-support.md",
         "extensions.md",
         "framework-choice.md",

tests/test_platform_edge_paths.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+import tempfile
+import types
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest import mock
+
+import base_cli._click_compat as click_compat
+import base_cli._private_files as private_files
+import base_cli._runtime as runtime
+from base_cli import history
+from base_cli._attach import (
+    _click_command_has_pending_children,
+    _normalize_attached_option_declaration,
+    _normalize_sensitive_parameters,
+    _restore_attached_click_command,
+    _restore_attached_click_main,
+    _selected_click_path,
+    _selected_click_paths,
+)
+
+
+class PrivateFileEdgeTests(unittest.TestCase):
+    def test_permission_helpers_skip_posix_mode_changes_on_windows(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "file"
+            path.write_text("payload", encoding="utf-8")
+            with mock.patch.object(private_files.os, "name", "nt"):
+                private_files.restrict_file(path)
+                private_files.restrict_directory(path.parent)
+
+    def test_sync_directory_tolerates_filesystems_without_directory_fsync(self) -> None:
+        with mock.patch.object(private_files.os, "fsync", side_effect=OSError("unsupported")):
+            private_files._sync_directory(123)  # pylint: disable=protected-access
+
+    def test_windows_replace_retries_transient_sharing_failure(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            source = Path(tmpdir) / "source"
+            destination = Path(tmpdir) / "destination"
+            source.write_text("payload", encoding="utf-8")
+            with (
+                mock.patch.object(private_files.os, "name", "nt"),
+                mock.patch.object(
+                    private_files.os,
+                    "replace",
+                    side_effect=[PermissionError("busy"), lambda src, dst: Path(dst).write_text(Path(src).read_text())],
+                ) as replace,
+                mock.patch.object(private_files.time, "sleep") as sleep,
+            ):
+                private_files._replace_with_retry(source, destination)  # pylint: disable=protected-access
+            self.assertEqual(replace.call_count, 2)
+            sleep.assert_called_once()
+
+    def test_parent_directory_open_is_disabled_on_windows(self) -> None:
+        path = Path("/tmp")
+        with mock.patch.object(private_files.os, "name", "nt"):
+            self.assertIsNone(private_files._open_parent_directory(path))  # pylint: disable=protected-access
+
+
+class ClickCompatibilityEdgeTests(unittest.TestCase):
+    def test_dialect_falls_back_to_public_click_without_typer(self) -> None:
+        with mock.patch.dict("sys.modules", {"typer": None}):
+            command = object()
+            self.assertIs(click_compat.dialect_for_command(command), __import__("click"))
+            self.assertFalse(click_compat.is_command(command))
+
+    def test_vendored_dialect_requires_a_click_module(self) -> None:
+        self.assertIsNone(click_compat._vendored_typer_dialect(types.SimpleNamespace()))  # pylint: disable=protected-access
+        self.assertIs(click_compat.dialect_for_typer(types.SimpleNamespace()), __import__("click"))
+
+    def test_marking_an_immutable_command_is_best_effort(self) -> None:
+        class Immutable:
+            __slots__ = ()
+
+        command = Immutable()
+        self.assertIs(click_compat.mark_command_dialect(command, object()), command)
+
+    def test_vendor_version_option_requires_a_version(self) -> None:
+        decorator = click_compat._vendor_version_option_factory(lambda **_: None, lambda *_args, **_kwargs: None)  # pylint: disable=protected-access
+        with self.assertRaisesRegex(RuntimeError, "version is required"):
+            decorator()
+
+
+class RuntimeEdgeTests(unittest.TestCase):
+    def test_owned_runtime_directory_collision_is_not_claimed(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "runs" / "run"
+            path.mkdir(parents=True)
+            with self.assertRaises(runtime.RuntimeDirectoryError):
+                runtime.create_owned_runtime_directory(path, Path(tmpdir))
+
+    def test_owned_runtime_directory_uses_portable_fallback(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "runs" / "portable"
+            with mock.patch.object(runtime, "_supports_secure_owned_directory_creation", return_value=False):
+                identity, descriptor = runtime.create_owned_runtime_directory(path, Path(tmpdir))
+            self.assertIsInstance(identity, tuple)
+            self.assertIsNone(descriptor)
+            self.assertTrue(path.is_dir())
+
+    def test_malformed_metadata_and_timestamps_are_ignored(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            bundle = Path(tmpdir) / "bundle"
+            bundle.mkdir()
+            (bundle / "run.json").write_text("[]", encoding="utf-8")
+            self.assertIsNone(runtime._read_bundle_metadata(bundle))  # pylint: disable=protected-access
+        self.assertIsNone(runtime._timestamp_to_epoch("not-a-timestamp"))  # pylint: disable=protected-access
+        self.assertIsNone(runtime._timestamp_to_epoch(123))  # pylint: disable=protected-access
+
+    def test_retention_without_policy_is_a_noop(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            logger = mock.Mock()
+            runtime.prune_run_bundles(Path(tmpdir), logger=logger)
+            logger.warning.assert_not_called()
+
+    def test_runtime_directory_failures_are_actionable(self) -> None:
+        path = Path("/tmp/base-cli-edge/runtime")
+        with mock.patch.object(Path, "mkdir", side_effect=OSError("read-only")):
+            with self.assertRaisesRegex(runtime.RuntimeDirectoryError, "Check permissions"):
+                runtime.create_runtime_directory(path, Path("/tmp/base-cli-edge"))
+
+    def test_safe_resolved_path_falls_back_when_resolution_fails(self) -> None:
+        path = Path("relative/path")
+        with mock.patch.object(Path, "resolve", side_effect=OSError("unavailable")):
+            self.assertEqual(runtime._safe_resolved_path(path), path.absolute())  # pylint: disable=protected-access
+
+
+class AttachmentHelperEdgeTests(unittest.TestCase):
+    def test_attachment_normalizers_accept_strings_and_reject_bad_values(self) -> None:
+        self.assertEqual(_normalize_sensitive_parameters("token"), frozenset({"token"}))
+        with self.assertRaises(TypeError):
+            _normalize_sensitive_parameters([""])
+        self.assertEqual(_normalize_attached_option_declaration("--NAME", str.lower), "--name")
+        self.assertEqual(_normalize_attached_option_declaration("NAME", str.lower), "NAME")
+
+    def test_selected_paths_follow_resolved_click_children(self) -> None:
+        root_command = object()
+        child_command = object()
+        root = types.SimpleNamespace(command=root_command, parent=None, info_name="root")
+        child = types.SimpleNamespace(command=child_command, parent=root, info_name="child")
+        resolutions = {id(root): [("child", child_command, child)]}
+        self.assertEqual(_selected_click_path(root, child, resolutions), (("child", child_command),))
+        self.assertEqual(_selected_click_paths(root, resolutions, {id(root): root}), ((("child", child_command),),))
+
+    def test_pending_children_and_restore_helpers_cover_fallbacks(self) -> None:
+        command = types.SimpleNamespace(resolve_command=lambda *_args: None)
+        context = types.SimpleNamespace(_protected_args=("child",), args=())
+        self.assertTrue(_click_command_has_pending_children(context, command))
+        self.assertFalse(_click_command_has_pending_children(types.SimpleNamespace(args=()), object()))
+
+        def original_invoke(_ctx: object) -> None:
+            return None
+
+        def original_resolve(_ctx: object, _args: object) -> None:
+            return None
+
+        command.invoke = lambda _ctx: "wrapped"
+        command.resolve_command = lambda _ctx, _args: "wrapped"
+        command.__base_cli_original_invoke__ = original_invoke
+        command.__base_cli_original_resolve__ = original_resolve
+        _restore_attached_click_command(command)
+        self.assertIs(command.invoke, original_invoke)
+        self.assertIs(command.resolve_command, original_resolve)
+
+        command.main = lambda: "wrapped"
+        command.__base_cli_original_main__ = original_invoke
+        _restore_attached_click_main(command)
+        self.assertIs(command.main, original_invoke)
+
+
+class HistoryEdgeTests(unittest.TestCase):
+    def test_primary_record_and_parser_cover_optional_and_invalid_fields(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            path = root / "history.jsonl"
+            history.write_primary_record(
+                path,
+                "demo --token=secret",
+                ["demo", "--token=secret"],
+                datetime(2026, 1, 1, tzinfo=timezone.utc),
+                1,
+                "run-1",
+                project="demo",
+                project_root=str(root),
+                manifest=str(root / "manifest.yaml"),
+                log_path=root / "primary.log",
+                bundle_path=root / "bundle",
+            )
+            line = path.read_text(encoding="utf-8").splitlines()[0]
+            self.assertIsNotNone(history.parse_finished_history_record_line(line))
+        self.assertIsNone(history.parse_finished_history_record_line("not json"))
+        self.assertIsNone(history.parse_finished_history_record_line("{}"))
+        self.assertEqual(history.optional_string("value"), "value")
+        self.assertIsNone(history.optional_string(1))
+        self.assertEqual(history.optional_int(2), 2)
+        self.assertIsNone(history.optional_int("2"))
+
+
+if __name__ == "__main__":
+    unittest.main()