Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@ numeric next development line is tracked in `DEVELOPMENT_VERSION`.

### Fixed

- Use the same readable blocked-trust explanations in workspace test preflight,
individual project command diagnostics, and `basectl trust status`.
- Make `basectl workspace test` preflight manifest-command trust across the
entire selected set before running any project test, and report explicit
blocked, not-run, start, completion, and dry-run states.
Expand Down
92 changes: 92 additions & 0 deletions cli/python/base_projects/tests/test_workspace_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,13 +11,16 @@
from pathlib import Path
from unittest import mock

from base_cli.testing import invoke
from base_projects import engine
from base_projects.test_preflight import project_test_preflight
from base_projects.workspace_manifest import WorkspaceManifestRepo
from base_projects.workspace_repo_inspection import WorkspaceRepoInspection
from base_projects.workspace_test import workspace_test_manifest_target
from base_setup.manifest import read_manifest
from base_setup.manifest_trust import ManifestCommandTrustStore
from base_setup.manifest_trust import compute_trust_identity
from base_trust import engine as trust_engine


def write_workspace_manifest(path: Path, repositories: tuple[str, ...]) -> None:
Expand Down Expand Up @@ -107,6 +110,95 @@ def invoke_workspace_test(


class WorkspaceTestCommandTests(unittest.TestCase):
def assert_trust_status_guidance(self, workspace: Path, home: Path, base_home: Path, explanation: str) -> str:
with mock.patch("base_cli.is_terminal", return_value=True):
result = invoke(
trust_engine.app,
["status", "demo", "--workspace", str(workspace)],
home=home,
env={"BASE_HOME": str(base_home)},
)
self.assertEqual(result.exit_code, 0, result.output)
self.assertEqual(result.stdout.splitlines()[0], f"{explanation}.")
self.assertIn("Current identity:", result.stdout)
self.assertIn("Review first:", result.stdout)
self.assertIn("Allow after review:", result.stdout)
return result.stdout

def test_blocked_trust_guidance_matches_individual_project_preflight(self) -> None:
for reason in ("not_allowed", "manifest_changed", "test_requirements_changed"):
with self.subTest(reason=reason), tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
home = root / "home"
home.mkdir()
base_home = root / "base"
write_fake_basectl(base_home)
workspace = root / "workspace"
workspace_manifest = root / "workspace.yaml"
write_workspace_manifest(workspace_manifest, ("demo",))
project_root = workspace / "demo"
write_test_manifest(project_root, "demo")
manifest_path = project_root / "base_manifest.yaml"
manifest_path.write_text(
manifest_path.read_text(encoding="utf-8").replace(
" command: ./run-tests.sh\n",
" command: ./run-tests.sh\n requirements: requirements-dev.txt\n",
),
encoding="utf-8",
)
requirements_path = project_root / "requirements-dev.txt"
requirements_path.write_text("pytest==8.0.0\n", encoding="utf-8")
store = ManifestCommandTrustStore(home=home)
if reason != "not_allowed":
store.allow(compute_trust_identity(read_manifest(manifest_path)), base_version="test")
if reason == "manifest_changed":
with manifest_path.open("a", encoding="utf-8") as stream:
stream.write("# Changed after approval.\n")
elif reason == "test_requirements_changed":
requirements_path.write_text("pytest==8.1.0\n", encoding="utf-8")

manifest = read_manifest(manifest_path)
self.assertEqual(store.status(compute_trust_identity(manifest)).reason, reason)
project_stderr = StringIO()
with mock.patch.dict(os.environ, {"HOME": str(home)}), redirect_stderr(project_stderr):
self.assertFalse(project_test_preflight(mock.Mock(), manifest))
diagnostic = project_stderr.getvalue()
self.assertIn("Review first:", diagnostic)
self.assertIn("Allow after review:", diagnostic)
self.assertIn("basectl trust allow demo", diagnostic)
explanation = next(
line.removeprefix("ERROR: ") for line in diagnostic.splitlines() if line.startswith("ERROR: ")
).removesuffix(".")

trust_output = self.assert_trust_status_guidance(workspace, home, base_home, explanation)
self.assertEqual("review the requirements file" in trust_output, reason == "test_requirements_changed")

for output_format in ("text", "json"):
with self.subTest(output_format=output_format), mock.patch(
"base_projects.workspace_test.run_workspace_subprocess"
) as run:
status, stdout, stderr = invoke_workspace_test(
[
"test", "--workspace", str(workspace), "--manifest", str(workspace_manifest),
"--format", output_format,
],
base_home,
home,
trusted_projects=(),
)
self.assertEqual(status, 1)
self.assertEqual(stderr, "")
run.assert_not_called()
if output_format == "json":
payload = json.loads(stdout)
self.assertFalse(payload["execution_started"])
self.assertEqual(payload["projects"][0]["status"], "blocked")
self.assertEqual(payload["projects"][0]["trust_status"], "blocked")
self.assertEqual(payload["projects"][0]["detail"], explanation)
else:
self.assertIn(f"reason: {explanation}\n", stdout)
self.assertIn(f"at '{project_root.resolve()}': {explanation}.\n", stdout)

def test_workspace_test_fails_closed_when_inspected_manifest_is_missing(self) -> None:
root = Path("/workspace/api")
inspection = WorkspaceRepoInspection(
Expand Down
6 changes: 2 additions & 4 deletions cli/python/base_projects/workspace_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
from base_setup.process import redact_command_output
from base_setup.manifest_model import BaseManifest
from base_setup.manifest_trust import ManifestCommandTrustStore
from base_setup.manifest_trust_guidance import blocked_command_summary


WorkspaceTestAction = Literal["test", "skip"]
Expand Down Expand Up @@ -422,10 +423,7 @@ def workspace_test_preflight(
"blocked",
command=command,
trust_status="blocked",
detail=(
"manifest command trust is blocked "
f"({trust_status.reason})"
),
detail=blocked_command_summary(trust_status),
)
return preflights

Expand Down
31 changes: 13 additions & 18 deletions cli/python/base_setup/manifest_trust_guidance.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,18 +40,13 @@ def print_status_text(trust_status: TrustStatus, surfaces: tuple[str, ...]) -> N
print_trust_scope_warning()
return

print(f"{blocked_command_summary(trust_status)}.")
if trust_status.reason in {"manifest_changed", "test_requirements_changed"}:
print(
f"Manifest command trust is blocked for project '{identity.project_name}': "
"manifest or declared test requirements changed."
)
if trust_status.reason == "test_requirements_changed":
print("Declared test requirements changed; review the requirements file before allowing commands.")
changed_project = (trust_status.changed_record or {}).get("project", {})
if isinstance(changed_project, dict) and changed_project.get("manifest_sha256"):
print(f"Recorded Manifest SHA-256: {changed_project['manifest_sha256']}")
else:
print(f"Manifest command trust is blocked for project '{identity.project_name}'.")
print_identity("Current identity", identity)
print()
print_trust_scope_warning()
Expand All @@ -62,6 +57,17 @@ def print_status_text(trust_status: TrustStatus, surfaces: tuple[str, ...]) -> N
print(f" {allow_command_text(identity)}")


def blocked_command_summary(trust_status: TrustStatus) -> str:
"""Return the shared blocked-trust explanation without terminal punctuation."""
project_name = trust_status.identity.project_name
if trust_status.reason in {"manifest_changed", "test_requirements_changed"}:
return (
f"Manifest command trust is blocked for project '{project_name}': "
"manifest command or declared test requirements contract changed"
)
return f"Manifest-declared commands are not allowed for project '{project_name}' on this machine"


def print_blocked_command_text(
trust_status: TrustStatus,
surfaces: tuple[str, ...],
Expand All @@ -70,18 +76,7 @@ def print_blocked_command_text(
) -> None:
print("EXECUTION BLOCKED: No project command was executed.", file=stream)
identity = trust_status.identity
if trust_status.reason in {"manifest_changed", "test_requirements_changed"}:
print(
f"ERROR: Manifest command trust is blocked for project '{identity.project_name}': "
"manifest command or declared test requirements contract changed.",
file=stream,
)
else:
print(
f"ERROR: Manifest-declared commands are not allowed for project "
f"'{identity.project_name}' on this machine.",
file=stream,
)
print(f"ERROR: {blocked_command_summary(trust_status)}.", file=stream)
print(f"Project root: {identity.project_root}", file=stream)
print(f"Manifest: {identity.manifest_path}", file=stream)
if trust_status.reason in {"manifest_changed", "test_requirements_changed"}:
Expand Down
5 changes: 3 additions & 2 deletions cli/python/base_trust/tests/test_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -415,7 +415,7 @@ def test_status_guidance_covers_demo_and_manifest_backed_activation(self) -> Non
)

self.assertEqual(result.exit_code, 0, result.output)
self.assertIn("Manifest command trust is blocked for project 'demo'.", result.stdout)
self.assertIn("Manifest-declared commands are not allowed for project 'demo' on this machine.", result.stdout)
self.assertIn(f"basectl run demo --list --workspace {workspace.resolve()}", result.stdout)
self.assertIn(f"basectl build demo --list --workspace {workspace.resolve()}", result.stdout)
self.assertIn(f"basectl test demo --dry-run --workspace {workspace.resolve()}", result.stdout)
Expand Down Expand Up @@ -451,7 +451,8 @@ def test_changed_manifest_status_shows_recorded_digest_and_reapproval_guidance(s

self.assertEqual(result.exit_code, 0, result.output)
self.assertIn(
"Manifest command trust is blocked for project 'demo': manifest or declared test requirements changed.",
"Manifest command trust is blocked for project 'demo': "
"manifest command or declared test requirements contract changed.",
result.stdout,
)
self.assertIn(f"Recorded Manifest SHA-256: {identity.manifest_sha256}", result.stdout)
Expand Down
Loading