"""CodeGraph governance + read-only audit commands.""" import json import os import shutil import subprocess import sys import tempfile import unittest from pathlib import Path from unittest import mock SOURCE = Path(__file__).resolve().parents[1] class CodegraphGovernanceTest(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 ('agent-worktree.sh', 'agent_worktree.py', 'agentctl.py', 'agentctl', 'agentctl_w4.py', 'agentctl_catalog.py'): shutil.copy2(SOURCE / name, self.root / 'scripts' / name) (self.root / 'configs').mkdir(parents=True, exist_ok=True) shutil.copy2(SOURCE.parent / 'configs' / 'agent-governance.json', self.root / 'configs' / 'agent-governance.json') (self.root / '.gitignore').write_text('.worktrees/\n.locks/\n.codegraph/\n') (self.root / 'source.txt').write_text('base\n') doc = self.root / '.doc/status' doc.mkdir(parents=True) ops = self.root / '.doc/ops' ops.mkdir(parents=True) (ops / 'CURRENT.md').write_text( '# 当前焦点\n\n## 焦点\n\n0. x\n\n## 并行认领\n\n' '| owner | task | branch | worktree | mutex | status |\n' '|---|---|---|---|---|---|\n' '| Cursor | seed | feature/seed | .worktrees/seed | m | done |\n\n' '## 清理接管与遗留现场\n\n| 路径/事项 | 分类与保留原因 | cleanup_owner | 复查触发 |\n' '|---|---|---|---|\n| — | none | — | — |\n\n' '## 集成槽位\n| holder | phase | since | scope | queue |\n' '|--------|-------|-------|-------|-------|\n' '| - | idle | 2026-09-21 | - | - |\n\n' '## 受控约束\n\n- none\n\n## 活跃入口\n\n- x\n' ) (doc / 'FEATURE-MATRIX.md').write_text( '# 矩阵\n\n| record_id | 模块 | 声明范围 | 状态 | 发布关系 | 证据 | 剩余工作 / 关闭条件 | mutex |\n' '|------|----------|------|----------|------|---------------------|-------|-------|\n' '| alpha-mod | Alpha | scope | done | non_blocker | e | closed | a |\n' '| beta-mod | Beta | scope | partial | non_blocker | e | open | b |\n' ) self.env = {**os.environ, 'PYTHONDONTWRITEBYTECODE': '1'} for name in ('AGENTCTL_RUN_ID', 'AGENTCTL_TASK_ID'): self.env.pop(name, None) 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') def run_cmd(self, args, cwd=None, env=None, timeout=30): return subprocess.run(args, cwd=cwd or self.root, env=env or self.env, text=True, capture_output=True, timeout=timeout) def git(self, *args, cwd=None): result = self.run_cmd(['git', *args], cwd=cwd) self.assertEqual(result.returncode, 0, result.stderr) return result def ctl(self, *args, cwd=None, check=True): result = self.run_cmd( [sys.executable, str(self.root / 'scripts/agentctl.py'), *args], cwd=cwd) if check: self.assertEqual(result.returncode, 0, result.stdout + result.stderr) return result def test_codegraph_sync_fail_open_without_binary(self): self.ctl('init') env = {**self.env, 'PATH': '/usr/bin:/bin'} result = self.run_cmd( [sys.executable, str(self.root / 'scripts/agentctl.py'), 'hooks', 'codegraph-sync'], env=env) self.assertEqual(result.returncode, 0, result.stderr) payload = json.loads(result.stdout) self.assertEqual(payload['status'], 'ok') self.assertEqual(payload['hook'], 'codegraph-sync') self.assertEqual(payload['codegraph']['status'], 'not_installed') self.assertEqual(payload['codegraph'].get('exit_code', 0), 0) def test_hooks_install_includes_codegraph_sync_hooks(self): self.ctl('init') self.ctl('hooks', 'install') for name in ('post-merge', 'post-checkout'): path = self.root / '.git/hooks' / name self.assertTrue(path.exists(), name) text = path.read_text() self.assertIn('agentctl-managed-v1', text) self.assertIn('hooks codegraph-sync', text) def test_docs_drift_detects_done_regression(self): self.ctl('init') first = self.ctl('docs', 'drift') payload = json.loads(first.stdout) self.assertEqual(payload['status'], 'ok') self.assertEqual(payload['baseline'], 'first_snapshot') self.assertEqual(payload['regressions'], []) # Regress alpha-mod done → planned on disk + commit so tip reflects it. matrix = self.root / '.doc/status/FEATURE-MATRIX.md' text = matrix.read_text().replace( '| alpha-mod | Alpha | scope | done |', '| alpha-mod | Alpha | scope | planned |') matrix.write_text(text) self.git('add', '.doc/status/FEATURE-MATRIX.md') self.git('commit', '-qm', 'regress alpha') second = self.ctl('docs', 'drift') payload = json.loads(second.stdout) self.assertTrue(any( r['record_id'] == 'alpha-mod' and r['severity'] == 'hard' for r in payload['regressions']), payload) open_ids = {row['record_id'] for row in payload['open_capabilities']} self.assertIn('alpha-mod', open_ids) self.assertIn('beta-mod', open_ids) def test_cleanup_report_never_deletes(self): self.ctl('init') helper = self.root / 'scripts/agent-worktree.sh' created = self.run_cmd(['bash', str(helper), 'create', 'feature/task-a']) self.assertEqual(created.returncode, 0, created.stderr) wt = Path(json.loads(created.stdout)['worktree']) self.assertTrue(wt.is_dir()) report = self.ctl('cleanup', 'report') payload = json.loads(report.stdout) self.assertEqual(payload['status'], 'ok') self.assertGreaterEqual(payload['summary']['total'], 1) self.assertTrue(wt.is_dir(), 'cleanup report must not remove worktrees') def test_create_skips_codegraph_when_missing(self): self.ctl('init') helper = self.root / 'scripts/agent-worktree.sh' env = {**self.env, 'PATH': '/nonexistent'} # Keep bash/git on a minimal path: prepend system essentials. env['PATH'] = '/usr/bin:/bin' # Hide only codegraph by putting a stub dir without it first... PATH without # codegraph is enough if which looks at PATH. Ensure no codegraph on PATH. result = self.run_cmd( ['bash', str(helper), 'create', 'feature/task-b'], env=env, timeout=60) self.assertEqual(result.returncode, 0, result.stderr) out = json.loads(result.stdout) self.assertEqual(out.get('codegraph'), 'not_installed') if __name__ == '__main__': unittest.main()