host
32620c4517
Server POST /tasks/{id}/merge no longer has record/enforce/off:
hard_mode_from_env and HARD_MODES deleted (not re-defaulted),
handler always evaluates the hard gate, env CLOUD_MERGE_GATE_MODE
and body merge_gate_mode are ignored server-side. The record
'no-evidence passes' and 'unprotected+confirm passes with
merge_gate_risk' server paths are gone with the dead risk audit;
no production caller justified keeping them (only production
caller of the non-hard evaluator was the merge handler itself;
audit_loop_wiring already used the hard gate). cloud/merge_gate
pure functions keep their #209 semantics but are documented as
non-server: they are no longer wired to any endpoint.
Server-level record/enforce/off tests rewritten to the new
behavior (renamed, not deleted silently); pure-function tests
unchanged.
(cherry picked from commit 08e75edbd6d04701965c5db951fa054cfbb75bb0)
Reconcile-port: handler_lifecycle.py hunks redirected into cloud/server.py _merge_task (target keeps 6-mixin split; mixin file does not exist here); handler_support.py kept at HEAD (its A1 hunk is a docstring of _validate_hard_gate_body, ported to server.py copy); test mock paths cloud.handler_lifecycle.* rewritten to cloud.server.*.
459 lines
20 KiB
Python
459 lines
20 KiB
Python
"""服务端合并硬门禁（cloud/merge_gate_hard，唯一服务端门禁）。
|
|
|
|
A1 之后服务端不再有 record/enforce/off 模式：环境变量
|
|
CLOUD_MERGE_GATE_MODE 与请求体 merge_gate_mode 在服务端一律被
|
|
忽略，旧值设了也等同 hard。本文件测硬模式：无证据即 409、
|
|
五项全查、缺一 409、状态不变、merge_gate_blocked 审计；unknown
|
|
保护不等于 protected；显式确认在硬模式下不覆盖任何一项；旧模式
|
|
环境变量不再产生旧行为（下面“旧模式值一律按 hard”用例钉住）。
|
|
#209 纯函数的 record 语义只见 test_merge_gate_protectskill.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_hard import (
|
|
blocked_audit_detail, evaluate_hard_merge_gate)
|
|
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)
|
|
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 ServerModesRemovedTestCase(unittest.TestCase):
|
|
def test_no_mode_parser_or_mode_table_left(self):
|
|
# A1 删模式的锁定：服务端模式解析函数与模式表已删除，
|
|
# 不是改默认值——旧值无处可沿用。
|
|
import cloud.merge_gate_hard as hard_mod
|
|
|
|
self.assertFalse(hasattr(hard_mod, "hard_mode_from_env"))
|
|
self.assertFalse(hasattr(hard_mod, "HARD_MODES"))
|
|
gate = evaluate_hard_merge_gate(**_hard_kwargs())
|
|
self.assertEqual(gate["mode"], "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），行为一律 hard。
|
|
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_no_longer_passes_no_fields(self):
|
|
# A1：record 环境变量在服务端被忽略，无证据调用按 hard 409，
|
|
# 不再是旧接口的“无证据放行”。
|
|
os.environ["CLOUD_MERGE_GATE_MODE"] = "record"
|
|
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.assertEqual(self._status_of(task["id"]), "completed")
|
|
|
|
def test_record_env_unprotected_confirm_no_longer_passes(self):
|
|
# A1：record 下“未保护 + 确认放行并记 merge_gate_risk”已消失：
|
|
# 确认在 hard 下不覆盖任何一项，且不再写 merge_gate_risk。
|
|
os.environ["CLOUD_MERGE_GATE_MODE"] = "record"
|
|
task = self._completed_task()
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/merge" % task["id"],
|
|
{"branch_protection": "unprotected",
|
|
"merge_gate_confirm": True},
|
|
self.owner_key)
|
|
self.assertEqual(status, 409)
|
|
self.assertIn("branch_protection_unprotected",
|
|
body["merge_gate"]["blocks"])
|
|
audit = self.auth.list_audit(self.project["id"])
|
|
self.assertFalse(any(a["action"] == "merge_gate_risk"
|
|
for a in audit))
|
|
|
|
def test_enforce_env_is_ignored_and_hard_blocks_unprotected(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)
|
|
# hard 门禁的码，不是旧 enforce 纯函数的 branch_protection_enforce。
|
|
self.assertIn("branch_protection_unprotected",
|
|
body["merge_gate"]["blocks"])
|
|
|
|
def test_off_env_is_ignored_and_hard_blocks_unprotected(self):
|
|
# A1：off 环境变量在服务端被忽略，不再是“只记风险不阻断”。
|
|
os.environ["CLOUD_MERGE_GATE_MODE"] = "off"
|
|
task = self._completed_task()
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/merge" % task["id"],
|
|
{"branch_protection": "unprotected"}, self.owner_key)
|
|
self.assertEqual(status, 409)
|
|
self.assertEqual(body["merge_gate"]["mode"], "hard")
|
|
self.assertIn("branch_protection_unprotected",
|
|
body["merge_gate"]["blocks"])
|
|
|
|
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()
|