"""主分支保护 + 目录隔离门禁（cloud/merge_gate）与技能工程双向断言。 门禁口径： - 目录隔离复用 v0.3 技能文本的 git diff --name-only 检查，不另造； 服务端 POST /tasks/{id}/merge 在此之前只改状态、不查目录、不查保护。 - 保护默认 record：未保护 + 无确认阻断，未保护 + 确认放行并记风险； enforce 下未保护一律阻断；unknown 记风险、不阻断（代码无法核实 GitHub 保护，unknown 不等于已保护，也不等于未保护）。 - 显式确认只覆盖保护风险，绝不覆盖目录违规。 技能工程： - 正文与附录双向断言：每个新关键词在正文、附录都要出现，失败 信息指明缺失的一侧（正文缺失 / 附录缺失），不是笼统的 assertIn。 - 播种仍只复制 SKILL.md：包内多余文件不进租户 home，租户已改 的 SKILL.md 绝不被覆盖；整目录复制明确不做（见 skill_seed.py）。 """ import http.client import json import os import sqlite3 import sys import tempfile import threading import unittest from pathlib import Path try: from cloud.auth import AuthStore from cloud.merge_gate import ( check_directory_isolation, directory_diff_command, evaluate_merge_gate, gate_from_request, gate_mode_from_env, protection_from_env) from cloud.server import create_server from cloud.skill_seed import ensure_codex_home_skills except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.auth import AuthStore from cloud.merge_gate import ( check_directory_isolation, directory_diff_command, evaluate_merge_gate, gate_from_request, gate_mode_from_env, protection_from_env) from cloud.server import create_server from cloud.skill_seed import ensure_codex_home_skills SKILL = Path(__file__).resolve().parents[1] / "skills" \ / "software-engineering" / "SKILL.md" BOTH_KEYWORDS = ("合并门禁", "CLOUD_MERGE_GATE_MODE", "显式确认", "git diff --name-only", "不做整目录复制") class MergeGateUnitTestCase(unittest.TestCase): def test_record_unprotected_needs_confirm(self): blocked = evaluate_merge_gate(protection="unprotected") self.assertFalse(blocked["allowed"]) self.assertIn("branch_protection_risk_unconfirmed", blocked["blocks"]) ok = evaluate_merge_gate(protection="unprotected", explicit_confirm=True) self.assertTrue(ok["allowed"]) self.assertIn("branch protection is not enabled", " ".join(ok["risks"])) def test_enforce_blocks_even_with_confirm(self): gate = evaluate_merge_gate(protection="unprotected", mode="enforce", explicit_confirm=True) self.assertFalse(gate["allowed"]) self.assertIn("branch_protection_enforce", gate["blocks"]) def test_off_allows_but_records_risk(self): gate = evaluate_merge_gate(protection="unprotected", mode="off") self.assertTrue(gate["allowed"]) self.assertTrue(gate["risks"]) def test_unknown_is_risk_not_block(self): gate = evaluate_merge_gate(protection="unknown") self.assertTrue(gate["allowed"]) self.assertTrue(gate["risks"]) self.assertEqual(gate["directory"], None) def test_protected_allows(self): self.assertTrue(evaluate_merge_gate( protection="protected")["allowed"]) def test_invalid_mode_falls_back_to_record_not_off(self): self.assertEqual(gate_mode_from_env({"CLOUD_MERGE_GATE_MODE": "x"}), "record") self.assertEqual(gate_mode_from_env({}), "record") self.assertEqual(protection_from_env({}), "unknown") gate = gate_from_request({}, env={"CLOUD_MERGE_GATE_MODE": "bogus", "CLOUD_BRANCH_PROTECTION": "unprotected"}) self.assertFalse(gate["allowed"]) def test_directory_reuses_v03_rule(self): self.assertEqual( directory_diff_command(), "git diff --name-only $(git merge-base origin/main HEAD)...HEAD") ok = check_directory_isolation( ["apps/web/a.py", "apps/web/sub/b.py"], "apps/web") self.assertTrue(ok["passed"]) self.assertEqual(ok["violations"], []) bad = check_directory_isolation( ["apps/web/a.py", "apps/ios/b.py", "server/c.py"], "apps/web") self.assertFalse(bad["passed"]) self.assertEqual(sorted(bad["violations"]), ["apps/ios/b.py", "server/c.py"]) # 目录前缀边界：apps/web2 不是 apps/web 的子目录。 edge = check_directory_isolation(["apps/web2/x.py"], "apps/web") self.assertEqual(edge["violations"], ["apps/web2/x.py"]) # 越界路径与契约变更批准：前者违规，后者放行但要记入 used。 esc = check_directory_isolation(["../secret.py", "/abs.py"], "apps/web") self.assertEqual(sorted(esc["violations"]), ["../secret.py", "/abs.py"]) exc = check_directory_isolation( ["apps/web/a.py", "contracts/api.md"], "apps/web", approved_exceptions=["contracts/api.md"]) self.assertTrue(exc["passed"]) self.assertEqual(exc["exceptions_used"], ["contracts/api.md"]) def test_confirm_never_overrides_directory_violation(self): gate = evaluate_merge_gate( files_changed=["apps/ios/b.py"], owned_dir="apps/web", protection="unprotected", explicit_confirm=True) self.assertFalse(gate["allowed"]) self.assertIn("directory_isolation_violation", gate["blocks"]) def test_files_without_owned_dir_is_inconclusive_and_blocks(self): gate = evaluate_merge_gate(files_changed=["apps/web/a.py"], owned_dir=None) self.assertFalse(gate["allowed"]) self.assertIn("directory_owned_dir_unknown", gate["blocks"]) no_evidence = evaluate_merge_gate() self.assertTrue(no_evidence["allowed"]) self.assertIsNone(no_evidence["directory"]) class SkillBidirectionalTestCase(unittest.TestCase): def test_protectskill_keywords_in_body_and_appendix(self): text = SKILL.read_text(encoding="utf-8") main, sep, appendix = text.partition("## 附录：角色模板") self.assertTrue(sep, "appendix marker missing") for kw in BOTH_KEYWORDS: self.assertIn(kw, main, "正文缺失：" + kw) self.assertIn(kw, appendix, "附录缺失：" + kw) def test_gate_in_coordinator_and_client_appendix_sections(self): text = SKILL.read_text(encoding="utf-8") coord = text.split("## 附录角色：00-coordinator", 1)[1].split( "## 附录角色：01-", 1)[0] client = text.split("## 附录角色：04-client-developer", 1)[1].split( "## 附录角色：05-", 1)[0] self.assertIn("合并门禁", coord, "附录协调员章节缺失：合并门禁") self.assertIn("合并门禁", client, "附录客户端章节缺失：合并门禁") class SkillSeedNoWholeDirTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() def tearDown(self): self.tmp.cleanup() def _pack(self): src = os.path.join(self.tmp.name, "pack", "demo-skill") os.makedirs(src, exist_ok=True) open(os.path.join(src, "SKILL.md"), "w").write("pack version\n") # 整目录复制才会带走的多余文件：现行行为必须忽略它们。 open(os.path.join(src, "extra.txt"), "w").write("extra\n") os.makedirs(os.path.join(src, "templates"), exist_ok=True) open(os.path.join(src, "templates", "x.md"), "w").write("x\n") return os.path.join(self.tmp.name, "pack") def test_only_skill_md_seeded_and_tenant_edit_never_overwritten(self): home = os.path.join(self.tmp.name, "home") out = ensure_codex_home_skills(home, self._pack()) self.assertEqual(out["skills_copied"], ["demo-skill"]) dest_dir = os.path.join(home, "skills", "demo-skill") self.assertTrue(os.path.isfile( os.path.join(dest_dir, "SKILL.md"))) self.assertFalse(os.path.exists( os.path.join(dest_dir, "extra.txt"))) self.assertFalse(os.path.exists( os.path.join(dest_dir, "templates"))) dest = os.path.join(dest_dir, "SKILL.md") open(dest, "w").write("tenant edited version\n") out2 = ensure_codex_home_skills(home, self._pack()) self.assertEqual(out2["skills_skipped_differ"], ["demo-skill"]) self.assertEqual(open(dest).read(), "tenant edited version\n") class MergeEndpointGateTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") self.auth = AuthStore(self.db_path) self.owner_id, self.owner_key = self.auth.create_user("owner") self.server = create_server(self.db_path, "127.0.0.1", 0) self.port = self.server.server_address[1] self.thread = threading.Thread( target=self.server.serve_forever, daemon=True) self.thread.start() status, self.project = self._req( "POST", "/projects", {"name": "demo"}, self.owner_key) self.assertEqual(status, 201) self._saved_env = {k: os.environ.get(k) for k in ("CLOUD_MERGE_GATE_MODE", "CLOUD_BRANCH_PROTECTION")} def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) for key, value in self._saved_env.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value try: self.auth.conn.close() except Exception: pass self.tmp.cleanup() def _req(self, method, path, body=None, key=None): conn = http.client.HTTPConnection( "127.0.0.1", self.port, timeout=10) hdrs = {} payload = None if body is not None: payload = json.dumps(body).encode("utf-8") hdrs["Content-Type"] = "application/json" if key is not None: hdrs["X-API-Key"] = key try: conn.request(method, path, body=payload, headers=hdrs) resp = conn.getresponse() data = resp.read() finally: conn.close() try: parsed = json.loads(data.decode("utf-8")) if data else None except ValueError: parsed = None return resp.status, parsed def _completed_task(self): status, task = self._req( "POST", "/tasks", {"project_id": self.project["id"], "prompt": "x"}, self.owner_key) self.assertEqual(status, 201) conn = sqlite3.connect(self.db_path) try: conn.execute( "UPDATE tasks SET status = 'completed' WHERE id = ?", (task["id"],)) conn.commit() finally: conn.close() return task def test_no_gate_fields_behaves_like_old_endpoint(self): # 本用例测的是 record 模式的过渡语义（#209 原意）：默认已是 # hard，见 test_merge_gate_hard.py 的默认 409 用例；这里显式 # 切 record 才沿用“无证据放行、只附带门禁信息”。 os.environ["CLOUD_MERGE_GATE_MODE"] = "record" task = self._completed_task() status, merged = self._req( "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key) self.assertEqual(status, 200) self.assertEqual(merged["status"], "merged") self.assertIn("merge_gate", merged) def test_directory_violation_blocks_without_changing_status(self): task = self._completed_task() status, body = self._req( "POST", "/tasks/%s/merge" % task["id"], {"files_changed": ["apps/web/a.py", "apps/ios/b.py"], "owned_dir": "apps/web"}, self.owner_key) self.assertEqual(status, 409) self.assertIn("directory_isolation_violation", body["merge_gate"]["blocks"]) _s, got = self._req("GET", "/tasks/%s" % task["id"], key=self.owner_key) self.assertEqual(got["status"], "completed") def test_unprotected_confirm_allows_and_audits_risk(self): # record 模式语义：硬模式下未保护 + 确认仍阻断，见硬门禁测试。 os.environ["CLOUD_MERGE_GATE_MODE"] = "record" task = self._completed_task() status, body = self._req( "POST", "/tasks/%s/merge" % task["id"], {"branch_protection": "unprotected"}, self.owner_key) self.assertEqual(status, 409) task2 = self._completed_task() status, merged = self._req( "POST", "/tasks/%s/merge" % task2["id"], {"branch_protection": "unprotected", "merge_gate_confirm": True}, self.owner_key) self.assertEqual(status, 200) self.assertEqual(merged["merge_gate"]["protection"], "unprotected") audit = self.auth.list_audit(self.project["id"]) self.assertTrue(any(a["action"] == "merge_gate_risk" for a in audit)) def test_enforce_env_blocks_even_with_confirm(self): os.environ["CLOUD_MERGE_GATE_MODE"] = "enforce" os.environ["CLOUD_BRANCH_PROTECTION"] = "unprotected" task = self._completed_task() status, body = self._req( "POST", "/tasks/%s/merge" % task["id"], {"merge_gate_confirm": True}, self.owner_key) self.assertEqual(status, 409) self.assertIn("branch_protection_enforce", body["merge_gate"]["blocks"]) if __name__ == "__main__": unittest.main()