diff --git a/cli/python/base_clean/tests/test_error_paths.py b/cli/python/base_clean/tests/test_error_paths.py new file mode 100644 index 00000000..70c13e8e --- /dev/null +++ b/cli/python/base_clean/tests/test_error_paths.py @@ -0,0 +1,172 @@ +from __future__ import annotations + +import os +from pathlib import Path +from unittest import mock + +from base_clean import engine + + +def rendered_warning(logger: mock.Mock) -> str: + call = logger.warning.call_args + return call.args[0] % call.args[1:] + + +def test_safe_directory_entries_reports_unreadable_directory(tmp_path: Path) -> None: + logger = mock.Mock() + directory = tmp_path / "unreadable" + + with mock.patch.object(Path, "iterdir", side_effect=PermissionError("listing denied")): + assert engine.safe_directory_entries(directory, "category root", logger) == [] + + assert "could not list directory: listing denied" in rendered_warning(logger) + + +def test_clean_path_rejects_non_directory_parent_component(tmp_path: Path) -> None: + cache_root = tmp_path / "cache" + cache_root.mkdir() + parent_file = cache_root / "base" + parent_file.write_text("not a directory", encoding="utf-8") + logger = mock.Mock() + + assert not engine.clean_path_is_safe(cache_root, parent_file / "runs", "candidate", logger) + + assert "parent component" in rendered_warning(logger) + assert "is not a directory" in rendered_warning(logger) + + +def test_clean_path_reports_cache_root_resolution_failure(tmp_path: Path) -> None: + cache_root = tmp_path / "cache" + cache_root.mkdir() + candidate = cache_root / "candidate" + candidate.mkdir() + logger = mock.Mock() + original_resolve = Path.resolve + + def fail_root_resolution(path: Path, *, strict: bool = False) -> Path: + if path == cache_root: + raise PermissionError("root denied") + return original_resolve(path, strict=strict) + + with mock.patch.object(Path, "resolve", new=fail_root_resolution): + assert not engine.clean_path_is_safe(cache_root, candidate, "candidate", logger) + + assert "could not resolve the Base cache root: root denied" in rendered_warning(logger) + + +def test_retention_skips_candidate_with_unreadable_metadata(tmp_path: Path) -> None: + cache_root = tmp_path / "cache" + run_root = cache_root / "base" / "runs" / "run-1" + run_root.mkdir(parents=True) + logger = mock.Mock() + + with mock.patch.object(engine, "run_metadata_mtime", side_effect=PermissionError("metadata denied")): + candidates = engine.find_log_retention_candidates(cache_root, 1, logger) + + assert not candidates + assert "could not read metadata: metadata denied" in rendered_warning(logger) + + +def test_category_scan_skips_candidate_with_unreadable_metadata(tmp_path: Path) -> None: + cache_root = tmp_path / "cache" + category_root = cache_root / "base" / "cache" / "components" + candidate = category_root / "entry" + candidate.mkdir(parents=True) + logger = mock.Mock() + original_stat = Path.stat + + def fail_candidate_stat(path: Path, *args, **kwargs): + if path == candidate: + raise PermissionError("stat denied") + return original_stat(path, *args, **kwargs) + + with mock.patch.object(Path, "stat", new=fail_candidate_stat), mock.patch.object( + engine, + "clean_path_is_safe", + return_value=True, + ): + candidates = engine.find_category_candidates( + cache_root, + category_root, + "cache", + cutoff=float("inf"), + logger=logger, + ) + + assert not candidates + assert "could not read metadata: stat denied" in rendered_warning(logger) + + +def test_remove_path_fails_closed_when_descriptor_relative_removal_fails(tmp_path: Path) -> None: + cache_root = tmp_path / "cache" + candidate = cache_root / "base" / "runs" / "run-1" + candidate.mkdir(parents=True) + proof = candidate / "proof.log" + proof.write_text("keep", encoding="utf-8") + logger = mock.Mock() + + with mock.patch.object(engine, "descriptor_safe_removal_supported", return_value=True), mock.patch.object( + engine, + "secure_remove_entry", + side_effect=PermissionError("removal denied"), + ): + assert not engine.remove_path(cache_root, candidate, logger) + + assert proof.read_text(encoding="utf-8") == "keep" + assert "could not open cache path without following symlinks: removal denied" in rendered_warning(logger) + + +def test_secure_remove_entry_rejects_directory_replacement() -> None: + initial = mock.Mock(st_mode=0o040755, st_dev=1, st_ino=2) + opened = mock.Mock(st_dev=1, st_ino=3) + + with mock.patch.object(os, "stat", return_value=initial), mock.patch.object( + os, "open", return_value=99 + ), mock.patch.object(os, "fstat", return_value=opened), mock.patch.object(os, "close") as close_mock: + try: + engine.secure_remove_entry(10, "candidate", 0) + except OSError as exc: + assert "changed while it was being opened" in str(exc) + else: + raise AssertionError("replacement must fail closed") + + close_mock.assert_called_once_with(99) + + +def test_run_that_becomes_active_is_retained_without_removal( + tmp_path: Path, + monkeypatch, + capsys, +) -> None: + cache_root = tmp_path / "cache" + run_root = cache_root / "base" / "runs" / "run-1" + run_root.mkdir(parents=True) + candidate = engine.CleanCandidate(run_root, "run", 100) + monkeypatch.setenv("BASE_CACHE_DIR", str(cache_root)) + monkeypatch.setattr(engine, "find_clean_candidates", lambda *args, **kwargs: [candidate]) + monkeypatch.setattr(engine, "run_is_running", lambda path: True) + remove = mock.Mock() + monkeypatch.setattr(engine, "remove_path", remove) + + assert engine.main(["--older-than", "1s", "--yes"]) == 0 + + assert f"Retaining\tactive run\t{run_root}" in capsys.readouterr().out + remove.assert_not_called() + + +def test_preview_with_only_unsafe_matches_reports_no_safe_result( + tmp_path: Path, + monkeypatch, + capsys, +) -> None: + cache_root = tmp_path / "cache" + candidate_path = cache_root / "candidate" + candidate = engine.CleanCandidate(candidate_path, "cache", 100) + monkeypatch.setenv("BASE_CACHE_DIR", str(cache_root)) + monkeypatch.setattr(engine, "find_clean_candidates", lambda *args, **kwargs: [candidate]) + monkeypatch.setattr(engine, "clean_path_is_safe", lambda *args, **kwargs: False) + + assert engine.main(["--older-than", "1s", "--dry-run"]) == 0 + + assert "No safe Base runtime artifacts matched" in capsys.readouterr().err + assert not candidate_path.exists() diff --git a/cli/python/base_devcontainer/tests/test_export.py b/cli/python/base_devcontainer/tests/test_export.py index 9096fa3f..de27bb9e 100644 --- a/cli/python/base_devcontainer/tests/test_export.py +++ b/cli/python/base_devcontainer/tests/test_export.py @@ -1,11 +1,17 @@ from __future__ import annotations import json +import io import tempfile import unittest +from contextlib import redirect_stdout from pathlib import Path +from base_devcontainer.export import DevcontainerExportError from base_devcontainer.export import build_devcontainer_export +from base_devcontainer.export import devcontainer_export_to_json +from base_devcontainer.export import dumps_export_json +from base_devcontainer.export import print_devcontainer_export_text from base_devcontainer.export import write_devcontainer_export from base_setup.manifest import read_manifest @@ -61,6 +67,143 @@ def test_write_devcontainer_export_writes_stable_json(self) -> None: self.assertEqual(payload, {"name": "demo"}) + def test_build_export_classifies_supported_unsupported_and_ambiguous_fields(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + manifest_path = root / "base_manifest.yaml" + (root / "Brewfile").write_text("brew 'jq'\n", encoding="utf-8") + write_manifest( + manifest_path, + "\n".join( + [ + "project:", + " name: demo", + "brewfile: Brewfile", + "mise: .mise.toml", + "python:", + " manager: uv", + " requires_python: '>=3.12'", + "ide:", + " vscode:", + " settings:", + " python.defaultInterpreterPath: .venv/bin/python", + " cursor:", + " extensions: [github.copilot]", + "artifacts:", + " - type: tool", + " name: terraform", + " version: latest", + "test:", + " command: pytest", + "health:", + " required_env: [API_TOKEN]", + " required_ports:", + " - port: 8080", + " state: listening", + "commands:", + " lint: ruff check .", + "activate:", + " source: [.base/activate.sh]", + "build:", + " targets:", + " app:", + " command: make build", + "", + ] + ), + ) + + export = build_devcontainer_export(read_manifest(manifest_path)) + + self.assertEqual( + export.devcontainer["customizations"], + {"vscode": {"settings": {"python.defaultInterpreterPath": ".venv/bin/python"}}}, + ) + self.assertEqual(export.supported, ("project.name", "ide.vscode.settings")) + self.assertEqual( + {finding.field for finding in export.unsupported}, + { + "ide.cursor", + "brewfile", + "mise", + "artifacts[1]", + "test", + "health.required_env", + "health.required_ports", + "commands", + "activate.source", + "build", + }, + ) + self.assertEqual( + {finding.field for finding in export.ambiguous}, + {"python.manager", "python.requires_python"}, + ) + + def test_build_export_classifies_external_python_environment_as_ambiguous(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + manifest_path = Path(tmpdir) / "base_manifest.yaml" + write_manifest( + manifest_path, + "project:\n name: demo\npython:\n venv_location: external\nartifacts: []\n", + ) + + export = build_devcontainer_export(read_manifest(manifest_path)) + + self.assertEqual([finding.field for finding in export.ambiguous], ["python.venv_location"]) + + def test_write_export_refuses_to_replace_existing_target(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + manifest_path = root / "base_manifest.yaml" + write_manifest(manifest_path, "project:\n name: demo\nartifacts: []\n") + target_path = root / ".devcontainer" / "devcontainer.json" + target_path.parent.mkdir() + target_path.write_text('{"name":"owned"}\n', encoding="utf-8") + export = build_devcontainer_export(read_manifest(manifest_path), write=True) + + with self.assertRaisesRegex(DevcontainerExportError, "refusing to replace"): + write_devcontainer_export(export) + + self.assertEqual(target_path.read_text(encoding="utf-8"), '{"name":"owned"}\n') + + def test_json_and_text_rendering_are_stable(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + manifest_path = root / "base_manifest.yaml" + write_manifest( + manifest_path, + "project:\n name: demo\ntest:\n command: pytest\nartifacts: []\n", + ) + export = build_devcontainer_export(read_manifest(manifest_path)) + output = io.StringIO() + + with redirect_stdout(output): + print_devcontainer_export_text(export) + + payload = devcontainer_export_to_json(export) + self.assertEqual(json.loads(dumps_export_json(export)), payload) + self.assertEqual(payload["schema_version"], 1) + self.assertEqual(payload["unsupported"][0]["field"], "test") + self.assertIn("Mode: dry-run", output.getvalue()) + self.assertIn("Dry run: no files were written.", output.getvalue()) + self.assertIn("Unsupported fields:\n- test:", output.getvalue()) + self.assertNotIn("Ambiguous fields:", output.getvalue()) + + def test_text_rendering_reports_write_mode(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + manifest_path = root / "base_manifest.yaml" + write_manifest(manifest_path, "project:\n name: demo\nartifacts: []\n") + export = build_devcontainer_export(read_manifest(manifest_path), write=True) + output = io.StringIO() + + with redirect_stdout(output): + print_devcontainer_export_text(export) + + self.assertIn("Mode: write", output.getvalue()) + self.assertIn("Wrote devcontainer JSON.", output.getvalue()) + if __name__ == "__main__": unittest.main() diff --git a/cli/python/base_trust/tests/test_command_error_paths.py b/cli/python/base_trust/tests/test_command_error_paths.py new file mode 100644 index 00000000..8c77e32d --- /dev/null +++ b/cli/python/base_trust/tests/test_command_error_paths.py @@ -0,0 +1,196 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +import yaml + +from base_cli.testing import invoke +from base_trust import engine + + +@pytest.mark.parametrize("output_format", ["yaml", "csv", "tsv"]) +def test_project_status_supports_documented_output_formats( + tmp_path: Path, + manifest_factory, + output_format: str, +) -> None: + home = tmp_path / "home" + workspace = tmp_path / "work" + manifest_factory.write(workspace / "demo") + + result = invoke( + engine.app, + ["status", "demo", "--workspace", str(workspace), "--format", output_format], + home=home, + env={"BASE_HOME": str(workspace / "base")}, + ) + + assert result.exit_code == 0, result.output + if output_format == "yaml": + payload = yaml.safe_load(result.stdout) + assert payload["project"]["name"] == "demo" + assert payload["status"] == "blocked" + else: + delimiter = "," if output_format == "csv" else "\t" + assert result.stdout.splitlines() == [delimiter.join(("demo", "blocked", "not_allowed"))] + + +def test_workspace_status_supports_yaml_and_empty_terminal_output( + tmp_path: Path, + manifest_factory, + monkeypatch: pytest.MonkeyPatch, +) -> None: + home = tmp_path / "home" + workspace = tmp_path / "work" + manifest_factory.write(workspace / "demo") + yaml_result = invoke( + engine.app, + ["status", "--workspace", str(workspace), "--format", "yaml"], + home=home, + env={"BASE_HOME": str(workspace / "base")}, + ) + + empty_workspace = tmp_path / "empty" + empty_workspace.mkdir() + monkeypatch.setattr(engine.base_cli, "is_terminal", lambda: True) + text_result = invoke( + engine.app, + ["status", "--workspace", str(empty_workspace)], + home=home, + env={"BASE_HOME": str(tmp_path / "base")}, + ) + + assert yaml.safe_load(yaml_result.stdout)["projects"][0]["project"]["name"] == "demo" + assert text_result.exit_code == 0, text_result.output + assert text_result.stdout == "No discovered projects require manifest command trust.\n" + + +def test_invalid_status_format_is_a_usage_error(tmp_path: Path) -> None: + result = invoke( + engine.app, + ["status", "--format", "toml"], + home=tmp_path / "home", + env={"BASE_HOME": str(tmp_path / "base")}, + ) + + assert result.exit_code == 2 + assert "Unsupported output format 'toml'" in result.stderr + + +@pytest.mark.parametrize("command", ["status", "require", "allow", "revoke"]) +def test_project_commands_report_resolution_errors( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + command: str, +) -> None: + monkeypatch.setattr( + engine, + "resolve_trust_identity" if command != "require" else "resolve_trust_identity_for_require", + lambda *args, **kwargs: (_ for _ in ()).throw(engine.TrustError("identity failed")), + ) + + result = invoke( + engine.app, + [command, "demo"], + home=tmp_path / "home", + env={"BASE_HOME": str(tmp_path / "base")}, + ) + + assert result.exit_code == 1 + assert "identity failed" in result.stderr + + +def test_workspace_status_reports_discovery_error( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + engine, + "workspace_status_projects", + lambda *args: (_ for _ in ()).throw(engine.TrustError("workspace failed")), + ) + + result = invoke( + engine.app, + ["status"], + home=tmp_path / "home", + env={"BASE_HOME": str(tmp_path / "base")}, + ) + + assert result.exit_code == 1 + assert "workspace failed" in result.stderr + + +def test_require_rejects_manifest_declaring_a_different_project( + tmp_path: Path, + manifest_factory, +) -> None: + manifest_path = manifest_factory.write(tmp_path / "work" / "actual", name="actual") + + result = invoke( + engine.app, + ["require", "expected", "--manifest", str(manifest_path)], + home=tmp_path / "home", + env={"BASE_HOME": str(tmp_path / "base")}, + ) + + assert result.exit_code == 1 + assert "declares project 'actual', not 'expected'" in result.stderr + + +def test_active_project_name_mismatch_is_reported( + tmp_path: Path, + manifest_factory, +) -> None: + manifest_path = manifest_factory.write(tmp_path / "active", name="actual") + + result = invoke( + engine.app, + ["status"], + home=tmp_path / "home", + env={ + "BASE_HOME": str(tmp_path / "base"), + "BASE_TRUST_ACTIVE_PROJECT": "expected", + "BASE_TRUST_ACTIVE_PROJECT_MANIFEST": str(manifest_path), + }, + ) + + assert result.exit_code == 1 + assert "Active project is 'expected' but its manifest declares project 'actual'" in result.stderr + + +def test_revoke_without_record_reports_noop( + tmp_path: Path, + manifest_factory, +) -> None: + workspace = tmp_path / "work" + manifest_factory.write(workspace / "demo") + + result = invoke( + engine.app, + ["revoke", "demo", "--workspace", str(workspace)], + home=tmp_path / "home", + env={"BASE_HOME": str(tmp_path / "base")}, + ) + + assert result.exit_code == 0, result.output + assert result.stdout == "No manifest command trust record found for project 'demo'.\n" + + +def test_status_payload_ignores_malformed_changed_record(tmp_path: Path, manifest_factory) -> None: + manifest_path = manifest_factory.write(tmp_path / "demo") + identity = engine.compute_trust_identity_for_manifest(manifest_path) + status = engine.TrustStatus( + status="blocked", + reason="manifest_changed", + identity=identity, + record=None, + changed_record={"project": {"manifest_sha256": 42}}, + ) + + payload = engine.status_payload(status) + + assert "recorded_manifest_sha256" not in payload + assert json.loads(json.dumps(payload))["reason"] == "manifest_changed"