WeiCHEN
ae001a85c1
Disposable Git fixtures only copied agentctl.py, so imports of the new W4 helper module crashed before any guard assertions could run.
229 lines
11 KiB
Python
229 lines
11 KiB
Python
"""Real guard integration in disposable Git repositories; Docker is never real."""
|
|
import os
|
|
import json
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
SOURCE = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class OpsGuardsTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name) / "repo"
|
|
self.root.mkdir()
|
|
(self.root / "scripts").mkdir()
|
|
for name in ("env-service.sh", "candidate-images-build.sh", "agentctl.py", "agentctl", "agentctl_w4.py"):
|
|
shutil.copy2(SOURCE / name, self.root / "scripts" / name)
|
|
(self.root / ".gitignore").write_text(".worktrees/\n.locks/\nconfigs/.env.*\n!configs/.env.*.example\ndata/\n")
|
|
(self.root / ".dockerignore").write_text(".doc\n.locks\n.worktrees\n")
|
|
(self.root / "configs").mkdir()
|
|
for target in ("shared", "dev", "test", "main"):
|
|
(self.root / f"docker-compose.{target}.yml").write_text("services: {}\n")
|
|
(self.root / "configs" / f".env.{target}.example").write_text("EXAMPLE=1\n")
|
|
self.git("init", "-b", "dev")
|
|
self.git("config", "user.email", "test@example.invalid")
|
|
self.git("config", "user.name", "Test")
|
|
self.git("add", ".")
|
|
self.git("commit", "-qm", "fixture")
|
|
self.base = self.git("rev-parse", "HEAD").stdout.strip()
|
|
self.bin = Path(self.tmp.name) / "bin"
|
|
self.bin.mkdir()
|
|
self.docker_log = Path(self.tmp.name) / "docker-calls"
|
|
docker = self.bin / "docker"
|
|
docker.write_text('#!/bin/sh\nprintf "%s\\n" "$*" >> "$DOCKER_LOG"\n'
|
|
'case "$1" in network) echo chat-agent-shared-net;; inspect) echo sha256:fixture;; esac\n')
|
|
docker.chmod(0o755)
|
|
self.env = {**os.environ, "PATH": f"{self.bin}:{os.environ['PATH']}",
|
|
"DOCKER_LOG": str(self.docker_log), "PYTHONDONTWRITEBYTECODE": "1"}
|
|
for name in ("AGENTCTL_RUN_ID", "AGENTCTL_TASK_ID"):
|
|
self.env.pop(name, None)
|
|
|
|
def run_cmd(self, args, cwd=None, env=None):
|
|
return subprocess.run(args, cwd=cwd or self.root, env=env or getattr(self, "env", None),
|
|
text=True, capture_output=True, timeout=15)
|
|
|
|
def git(self, *args, cwd=None):
|
|
result = self.run_cmd(["git", *args], cwd=cwd)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
return result
|
|
|
|
def service(self, target, action, root=None):
|
|
root = root or self.root
|
|
return self.run_cmd(["bash", str(root / "scripts/env-service.sh"), target, action], cwd=root)
|
|
|
|
def ctl(self, *args, cwd=None):
|
|
return self.run_cmd([sys.executable, str(self.root / "scripts/agentctl.py"), *args], cwd=cwd)
|
|
|
|
def init(self):
|
|
result = self.ctl("init")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def register_operator(self, *capabilities):
|
|
self.init()
|
|
linked = self.root / ".worktrees/operator"
|
|
self.git("worktree", "add", "-b", "feature/operator", str(linked), self.base)
|
|
args = ["task", "register", "--id", "operator", "--owner", "test", "--scope", "scripts"]
|
|
for capability in capabilities:
|
|
args += ["--capability", capability]
|
|
result = self.ctl(*args, cwd=linked)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
return linked
|
|
|
|
def supervised_service(self, resource, target, action, cwd=None, script_root=None):
|
|
return self.ctl("with", "--task", "operator", "--resource", resource,
|
|
"--cwd", str(cwd or self.root), "--", "bash",
|
|
str((script_root or self.root) / "scripts/env-service.sh"), target, action)
|
|
|
|
def assert_no_write_effects(self, root=None):
|
|
root = root or self.root
|
|
self.assertFalse((root / "data").exists())
|
|
for target in ("shared", "dev", "test", "main"):
|
|
self.assertFalse((root / "configs" / f".env.{target}").exists())
|
|
|
|
def test_queries_never_initialize_environment(self):
|
|
for action in ("status", "ps", "logs"):
|
|
with self.subTest(action=action):
|
|
result = self.service("dev", action)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assert_no_write_effects()
|
|
self.assertFalse((self.root / ".locks").exists())
|
|
self.assertIn("logs -f --tail=100", self.docker_log.read_text())
|
|
|
|
def test_service_writes_without_lease_refuse_before_any_side_effect(self):
|
|
for action in ("start", "stop", "restart", "rebuild", "down"):
|
|
with self.subTest(action=action):
|
|
result = self.service("dev", action)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assert_no_write_effects()
|
|
self.assertFalse(self.docker_log.exists())
|
|
|
|
def test_linked_worktree_writes_refuse_and_queries_remain_read_only(self):
|
|
linked = self.root / ".worktrees/task"
|
|
self.git("worktree", "add", "-b", "feature/task", str(linked), self.base)
|
|
for action in ("start", "stop", "restart", "rebuild", "down"):
|
|
result = self.service("dev", action, linked)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assert_no_write_effects(linked)
|
|
self.assertFalse(self.docker_log.exists())
|
|
result = self.service("dev", "logs", linked)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assert_no_write_effects(linked)
|
|
|
|
def test_invalid_action_does_not_initialize_environment(self):
|
|
self.assertNotEqual(self.service("dev", "bogus").returncode, 0)
|
|
self.assert_no_write_effects()
|
|
self.assertFalse(self.docker_log.exists())
|
|
|
|
def test_build_accepts_nightly_evidence_excluded_from_build_context(self):
|
|
self.init()
|
|
evidence = self.root / ".doc/test-reports/assets/promotion-receipts/evidence-fixture/go-test.txt"
|
|
evidence.parent.mkdir(parents=True)
|
|
evidence.write_text("--- PASS: TestFixture (1s)\n")
|
|
result = self.run_cmd(["bash","scripts/candidate-images-build.sh","--commit",self.base])
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertTrue(self.docker_log.exists())
|
|
|
|
def test_build_context_is_frozen_when_managed_source_changes(self):
|
|
self.init()
|
|
(self.root / "build_marker.txt").write_text("original")
|
|
self.git("add", "build_marker.txt")
|
|
self.git("commit", "-qm", "source marker")
|
|
source = self.root / ".worktrees/source"
|
|
self.git("worktree","add","-b","feature/source",str(source),"dev")
|
|
result = self.ctl("task","register","--id","source","--owner","test","--scope","build_marker.txt",cwd=source)
|
|
self.assertEqual(result.returncode,0,result.stderr)
|
|
observed = Path(self.tmp.name) / "observed-content"
|
|
docker = self.bin / "docker"
|
|
docker.write_text('#!/bin/sh\n'
|
|
'if [ "$1" = build ]; then\n'
|
|
' for context in "$@"; do :; done\n'
|
|
' python3 "$PWD/scripts/agentctl.py" with --task source --resource path:build_marker.txt -- python3 -c '
|
|
+ '\'from pathlib import Path; Path("build_marker.txt").write_text("changed")\'' + '\n'
|
|
' cat "$context/build_marker.txt" > "$OBSERVED_CONTENT"\n'
|
|
'elif [ "$1" = inspect ]; then echo sha256:fixture; fi\n')
|
|
docker.chmod(0o755)
|
|
result = self.run_cmd(["bash",str(source/"scripts/candidate-images-build.sh")],cwd=source,
|
|
env={**self.env,"OBSERVED_CONTENT":str(observed)})
|
|
self.assertEqual(result.returncode,0,result.stderr)
|
|
self.assertEqual((source/"build_marker.txt").read_text(),"changed")
|
|
self.assertEqual(observed.read_text(),"original")
|
|
|
|
def test_build_mismatched_commit_refuses_before_docker_or_receipts(self):
|
|
self.init()
|
|
self.git("commit", "--allow-empty", "-qm", "second")
|
|
result = self.run_cmd(["bash", "scripts/candidate-images-build.sh", "--commit", self.base])
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(self.docker_log.exists())
|
|
self.assertFalse((self.root / ".doc").exists())
|
|
|
|
def test_build_dirty_source_refuses_before_docker_or_receipts(self):
|
|
self.init()
|
|
(self.root / "docker-compose.dev.yml").write_text("dirty\n")
|
|
result = self.run_cmd(["bash", "scripts/candidate-images-build.sh"])
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(self.docker_log.exists())
|
|
self.assertFalse((self.root / ".doc").exists())
|
|
|
|
def test_build_untracked_source_refuses_before_docker_or_receipts(self):
|
|
self.init()
|
|
(self.root / "surprise.txt").write_text("untracked\n")
|
|
result = self.run_cmd(["bash", "scripts/candidate-images-build.sh"])
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(self.docker_log.exists())
|
|
self.assertFalse((self.root / ".doc").exists())
|
|
|
|
def test_clean_build_records_exact_source_commit(self):
|
|
self.init()
|
|
result = self.run_cmd(["bash", "scripts/candidate-images-build.sh", "--commit", self.base])
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
manifest = self.root / ".doc/test-reports/assets/promotion-receipts" / f"{self.base}.images.json"
|
|
receipt = json.loads(manifest.read_text())
|
|
self.assertEqual(receipt["commit"], self.base)
|
|
self.assertEqual(receipt["tags"]["app"], f"shuimu-agent-app:candidate-{self.base[:12]}")
|
|
self.assertEqual(self.docker_log.read_text().count("build -f "), 2)
|
|
|
|
def test_authorized_pinned_rebuild_preserves_no_build(self):
|
|
self.register_operator("env:test")
|
|
(self.root / "configs/.env.test.images").write_text("PINNED_IMAGE_MODE=1\n")
|
|
result = self.supervised_service("env:test", "test", "rebuild")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
calls = self.docker_log.read_text()
|
|
self.assertIn("up -d --no-build --force-recreate", calls)
|
|
self.assertNotIn("up -d --build", calls)
|
|
|
|
def test_one_environment_lease_does_not_authorize_all(self):
|
|
self.register_operator("env:dev")
|
|
result = self.supervised_service("env:dev", "all", "start")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assert_no_write_effects()
|
|
self.assertFalse(self.docker_log.exists())
|
|
|
|
def test_environment_lease_cannot_deploy_dirty_main_source(self):
|
|
self.register_operator("env:dev")
|
|
(self.root / "docker-compose.dev.yml").write_text("dirty source\n")
|
|
result = self.supervised_service("env:dev", "dev", "rebuild")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assert_no_write_effects()
|
|
self.assertFalse(self.docker_log.exists())
|
|
|
|
def test_valid_lease_cannot_hide_linked_caller_or_linked_script(self):
|
|
linked = self.register_operator("env:dev")
|
|
for cwd, script_root in ((linked, self.root), (self.root, linked)):
|
|
with self.subTest(cwd=cwd, script_root=script_root):
|
|
result = self.supervised_service("env:dev", "dev", "start", cwd=cwd, script_root=script_root)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assert_no_write_effects()
|
|
self.assert_no_write_effects(linked)
|
|
self.assertFalse(self.docker_log.exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|