diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 243c563..9ee87d4 100644 --- a/.github/workflows/tests.yml +++ b/.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 diff --git a/.gitignore b/.gitignore index d3974d3..462b3ae 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ __pycache__/ dist/ build/ *.egg-info/ +coverage.json diff --git a/CHANGELOG.md b/CHANGELOG.md index 90e0ee5..cb53260 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/docs/coverage-policy.md b/docs/coverage-policy.md new file mode 100644 index 0000000..f7bf506 --- /dev/null +++ b/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. diff --git a/mkdocs.yml b/mkdocs.yml index 9baffab..209d209 100644 --- a/mkdocs.yml +++ b/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 diff --git a/scripts/validate_coverage.py b/scripts/validate_coverage.py new file mode 100644 index 0000000..c1199a3 --- /dev/null +++ b/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() diff --git a/scripts/validate_docs.py b/scripts/validate_docs.py index fa81374..0585531 100644 --- a/scripts/validate_docs.py +++ b/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", diff --git a/tests/test_platform_edge_paths.py b/tests/test_platform_edge_paths.py new file mode 100644 index 0000000..0fc1347 --- /dev/null +++ b/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()