"""服务端合并硬门禁（cloud/merge_gate_hard，默认 hard 模式）。 与 #209 的分工（不要把两套语义混成一套断言）： - #209 的纯函数与 record/enforce/off 语义原样保留，见 test_merge_gate_protectskill.py；本文件不重测 record 的 “无证据放行”，只测：record 仍是显式 env 才能选到的过渡模式。 - 本文件测硬模式：默认（env 未设或非法值）即 hard，五项全查、 缺一 409、状态不变、merge_gate_blocked 审计；unknown 保护 不等于 protected；显式确认在硬模式下不覆盖任何一项； 调用方请求体里的 merge_gate_mode 不能把硬门禁降级。 """ 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_hard import ( blocked_audit_detail, evaluate_hard_merge_gate, hard_mode_from_env) from cloud.server import create_server except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.auth import AuthStore from cloud.merge_gate_hard import ( blocked_audit_detail, evaluate_hard_merge_gate, hard_mode_from_env) from cloud.server import create_server def _hard_kwargs(**overrides): kwargs = { "files_changed": ["apps/web/a.py"], "owned_dir": "apps/web", "protection": "protected", "ci_status": "success", "ci_checks": [{"name": "ci", "status": "success"}], "required_checks": ["ci"], "contract_version": "v2", "contract_frozen": True, "coordinator_approved": True, "unresolved_review_comments": 0, } kwargs.update(overrides) return kwargs class HardModeEnvTestCase(unittest.TestCase): def test_unset_and_invalid_mode_fall_to_hard(self): self.assertEqual(hard_mode_from_env({}), "hard") self.assertEqual(hard_mode_from_env( {"CLOUD_MERGE_GATE_MODE": ""}), "hard") self.assertEqual(hard_mode_from_env( {"CLOUD_MERGE_GATE_MODE": "bogus"}), "hard") self.assertEqual(hard_mode_from_env( {"CLOUD_MERGE_GATE_MODE": "record"}), "record") self.assertEqual(hard_mode_from_env( {"CLOUD_MERGE_GATE_MODE": "HARD"}), "hard") class HardGateUnitTestCase(unittest.TestCase): def test_full_evidence_passes(self): gate = evaluate_hard_merge_gate(**_hard_kwargs()) self.assertTrue(gate["allowed"]) self.assertEqual(gate["blocks"], []) self.assertEqual(gate["mode"], "hard") self.assertTrue(gate["ci"]["all_green"]) def test_no_evidence_blocks_every_missing_item(self): gate = evaluate_hard_merge_gate() self.assertFalse(gate["allowed"]) for code in ("ci_evidence_missing", "directory_evidence_missing", "contract_version_missing", "contract_not_frozen", "coordinator_approval_missing", "review_evidence_missing", "branch_protection_unknown"): self.assertIn(code, gate["blocks"]) def test_ci_variants_block(self): pending = evaluate_hard_merge_gate(**_hard_kwargs(ci_status="pending")) self.assertIn("ci_not_green", pending["blocks"]) failed_check = evaluate_hard_merge_gate(**_hard_kwargs( ci_checks=[{"name": "ci", "status": "failure"}])) self.assertIn("ci_check_failed", failed_check["blocks"]) self.assertFalse(failed_check["ci"]["all_green"]) empty_required = evaluate_hard_merge_gate(**_hard_kwargs( required_checks=[])) self.assertIn("ci_required_checks_unconfigured", empty_required["blocks"]) missing_required = evaluate_hard_merge_gate(**_hard_kwargs( required_checks=["ci", "lint"])) self.assertIn("ci_required_check_missing", missing_required["blocks"]) bare_success = evaluate_hard_merge_gate(**_hard_kwargs( ci_checks=None, required_checks=None)) self.assertIn("ci_evidence_missing", bare_success["blocks"]) def test_directory_reuses_v03_rule(self): edge = evaluate_hard_merge_gate(**_hard_kwargs( files_changed=["apps/web2/x.py"])) self.assertIn("directory_isolation_violation", edge["blocks"]) escape = evaluate_hard_merge_gate(**_hard_kwargs( files_changed=["../secret.py", "/abs.py"])) self.assertEqual(escape["directory"]["violations"], ["../secret.py", "/abs.py"]) exc = evaluate_hard_merge_gate(**_hard_kwargs( files_changed=["apps/web/a.py", "contracts/api.md"], approved_exceptions=["contracts/api.md"])) self.assertTrue(exc["allowed"]) self.assertEqual(exc["directory"]["exceptions_used"], ["contracts/api.md"]) no_dir = evaluate_hard_merge_gate(**_hard_kwargs( files_changed=None)) self.assertIn("directory_evidence_missing", no_dir["blocks"]) no_owned = evaluate_hard_merge_gate(**_hard_kwargs(owned_dir=None)) self.assertIn("directory_owned_dir_unknown", no_owned["blocks"]) def test_contract_variants_block(self): missing = evaluate_hard_merge_gate(**_hard_kwargs( contract_version=None)) self.assertIn("contract_version_missing", missing["blocks"]) unfrozen = evaluate_hard_merge_gate(**_hard_kwargs( contract_frozen=False)) self.assertIn("contract_not_frozen", unfrozen["blocks"]) mismatch = evaluate_hard_merge_gate(**_hard_kwargs( contract_expected_version="v3")) self.assertIn("contract_version_mismatch", mismatch["blocks"]) matched = evaluate_hard_merge_gate(**_hard_kwargs( contract_expected_version="v2")) self.assertTrue(matched["allowed"]) def test_coordinator_and_reviews_block(self): missing = evaluate_hard_merge_gate(**_hard_kwargs( coordinator_approved=None)) self.assertIn("coordinator_approval_missing", missing["blocks"]) denied = evaluate_hard_merge_gate(**_hard_kwargs( coordinator_approved=False)) self.assertIn("coordinator_not_approved", denied["blocks"]) unresolved = evaluate_hard_merge_gate(**_hard_kwargs( unresolved_review_comments=["c1"])) self.assertIn("unresolved_review_comments", unresolved["blocks"]) count = evaluate_hard_merge_gate(**_hard_kwargs( unresolved_review_comments=2)) self.assertIn("unresolved_review_comments", count["blocks"]) missing_reviews = evaluate_hard_merge_gate(**_hard_kwargs( unresolved_review_comments=None)) self.assertIn("review_evidence_missing", missing_reviews["blocks"]) def test_protection_unknown_and_unprotected_block(self): unknown = evaluate_hard_merge_gate(**_hard_kwargs( protection="unknown")) self.assertIn("branch_protection_unknown", unknown["blocks"]) illegal = evaluate_hard_merge_gate(**_hard_kwargs( protection="bogus")) self.assertEqual(illegal["protection"], "unknown") self.assertIn("branch_protection_unknown", illegal["blocks"]) def test_unprotected_no_confirm_blocks(self): gate = evaluate_hard_merge_gate(**_hard_kwargs( protection="unprotected")) self.assertFalse(gate["allowed"]) self.assertIn("branch_protection_unprotected", gate["blocks"]) def test_confirm_never_overrides_anything_in_hard_mode(self): # 硬模式没有“确认即放行”：确认不是本函数的入参，调用方 # 即使在请求体里带 merge_gate_confirm，服务端硬路径也不读它； # 这里钉住纯函数侧：未保护 + 目录违规同时存在时两项都拦。 gate = evaluate_hard_merge_gate(**_hard_kwargs( protection="unprotected", files_changed=["apps/web/a.py", "apps/ios/b.py"])) self.assertIn("branch_protection_unprotected", gate["blocks"]) self.assertIn("directory_isolation_violation", gate["blocks"]) def test_blocked_audit_detail_format(self): gate = evaluate_hard_merge_gate() detail = blocked_audit_detail("task-1", gate) self.assertTrue(detail.startswith("task-1 mode=hard blocks=")) self.assertIn("ci_evidence_missing", detail) class HardGateEndpointTestCase(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) # 默认即 hard：每个用例开始时清掉模式/保护 env，个别用例 # 再显式设置 record/enforce/off/bogus。 self._saved_env = {k: os.environ.get(k) for k in ("CLOUD_MERGE_GATE_MODE", "CLOUD_BRANCH_PROTECTION")} os.environ.pop("CLOUD_MERGE_GATE_MODE", None) os.environ.pop("CLOUD_BRANCH_PROTECTION", None) 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 _hard_body(self, **overrides): body = { "files_changed": ["apps/web/a.py"], "owned_dir": "apps/web", "branch_protection": "protected", "ci_status": "success", "ci_checks": [{"name": "ci", "status": "success"}], "required_checks": ["ci"], "contract_version": "v2", "contract_frozen": True, "coordinator_approved": True, "unresolved_review_comments": 0, } body.update(overrides) return body def _status_of(self, task_id): _s, got = self._req("GET", "/tasks/%s" % task_id, key=self.owner_key) return got["status"] def test_default_no_fields_blocks_status_unchanged_and_audits(self): task = self._completed_task() status, body = self._req( "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key) self.assertEqual(status, 409) self.assertEqual(body["merge_gate"]["mode"], "hard") self.assertIn("ci_evidence_missing", body["merge_gate"]["blocks"]) self.assertEqual(self._status_of(task["id"]), "completed") audit = self.auth.list_audit(self.project["id"]) blocked = [a for a in audit if a["action"] == "merge_gate_blocked"] self.assertTrue(blocked) self.assertIn(task["id"], blocked[0]["detail"]) self.assertIn("ci_evidence_missing", blocked[0]["detail"]) def test_default_full_evidence_merges(self): task = self._completed_task() status, merged = self._req( "POST", "/tasks/%s/merge" % task["id"], self._hard_body(), self.owner_key) self.assertEqual(status, 200) self.assertEqual(merged["status"], "merged") self.assertEqual(merged["merge_gate"]["mode"], "hard") self.assertEqual(merged["merge_gate"]["blocks"], []) def test_default_missing_one_item_blocks(self): task = self._completed_task() status, body = self._req( "POST", "/tasks/%s/merge" % task["id"], self._hard_body(unresolved_review_comments=2), self.owner_key) self.assertEqual(status, 409) self.assertIn("unresolved_review_comments", body["merge_gate"]["blocks"]) self.assertEqual(self._status_of(task["id"]), "completed") def test_default_unknown_protection_blocks(self): task = self._completed_task() body = self._hard_body() del body["branch_protection"] status, resp = self._req( "POST", "/tasks/%s/merge" % task["id"], body, self.owner_key) self.assertEqual(status, 409) self.assertIn("branch_protection_unknown", resp["merge_gate"]["blocks"]) def test_default_unprotected_blocks_even_with_confirm(self): task = self._completed_task() status, body = self._req( "POST", "/tasks/%s/merge" % task["id"], self._hard_body(branch_protection="unprotected", merge_gate_confirm=True), self.owner_key) self.assertEqual(status, 409) self.assertIn("branch_protection_unprotected", body["merge_gate"]["blocks"]) self.assertEqual(self._status_of(task["id"]), "completed") def test_default_confirm_never_overrides_directory_violation(self): task = self._completed_task() status, body = self._req( "POST", "/tasks/%s/merge" % task["id"], self._hard_body( files_changed=["apps/web/a.py", "apps/ios/b.py"], merge_gate_confirm=True), self.owner_key) self.assertEqual(status, 409) self.assertIn("directory_isolation_violation", body["merge_gate"]["blocks"]) def test_body_mode_cannot_downgrade_hard(self): task = self._completed_task() status, body = self._req( "POST", "/tasks/%s/merge" % task["id"], {"merge_gate_mode": "record"}, self.owner_key) self.assertEqual(status, 409) self.assertEqual(body["merge_gate"]["mode"], "hard") def test_invalid_env_mode_falls_to_hard(self): os.environ["CLOUD_MERGE_GATE_MODE"] = "bogus" task = self._completed_task() status, body = self._req( "POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key) self.assertEqual(status, 409) self.assertEqual(body["merge_gate"]["mode"], "hard") def test_record_env_preserves_old_no_fields_behavior(self): 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") def test_record_env_unprotected_confirm_still_passes_with_risk(self): 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) 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_unprotected_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"]) def test_off_env_is_explicit_opt_down_only(self): os.environ["CLOUD_MERGE_GATE_MODE"] = "off" task = self._completed_task() status, merged = self._req( "POST", "/tasks/%s/merge" % task["id"], {"branch_protection": "unprotected"}, self.owner_key) self.assertEqual(status, 200) self.assertEqual(merged["merge_gate"]["mode"], "off") def test_malformed_hard_body_returns_400(self): task = self._completed_task() status, _ = self._req( "POST", "/tasks/%s/merge" % task["id"], self._hard_body(ci_checks="ci"), self.owner_key) self.assertEqual(status, 400) status, _ = self._req( "POST", "/tasks/%s/merge" % task["id"], self._hard_body(unresolved_review_comments=True), self.owner_key) self.assertEqual(status, 400) self.assertEqual(self._status_of(task["id"]), "completed") class HardGateSkillTextTestCase(unittest.TestCase): def test_hard_keywords_in_body_and_appendix(self): # 正文换行会把“默认 hard”断成两行，断言前先把空白归一化。 import re raw = (Path(__file__).resolve().parents[1] / "skills" / "software-engineering" / "SKILL.md").read_text( encoding="utf-8") text = re.sub(r"\s+", " ", raw) main, sep, appendix = text.partition("## 附录：角色模板") self.assertTrue(sep, "appendix marker missing") coord = text.split("## 附录角色：00-coordinator", 1)[1].split( "## 附录角色：01-", 1)[0] client = text.split("## 附录角色：04-client-developer", 1)[1].split( "## 附录角色：05-", 1)[0] for kw in ("默认 hard", "merge_gate_blocked"): self.assertIn(kw, main, "正文缺失：" + kw) self.assertIn(kw, coord, "附录协调员章节缺失：" + kw) self.assertIn(kw, client, "附录客户端章节缺失：" + kw) self.assertIn("五项", main, "正文缺失：五项") if __name__ == "__main__": unittest.main()