code-to10 v6 A1: delete server merge-gate modes, hard only

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.*.
This commit is contained in:
host committed 2026-11-03 02:49:26 +00:00
1 parent b0d51b139c
commit 32620c4517
8 files changed
+200 -183

No files matched your search

+47 -42
View File
@@ -1,13 +1,13 @@
"""服务端合并硬门禁(cloud/merge_gate_hard,默认 hard 模式)。
"""服务端合并硬门禁(cloud/merge_gate_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 不能把硬门禁降级。
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
@@ -23,13 +23,13 @@ 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)
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, hard_mode_from_env)
blocked_audit_detail, evaluate_hard_merge_gate)
from cloud.server import create_server
@@ -50,17 +50,16 @@ def _hard_kwargs(**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 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):
@@ -200,7 +199,7 @@ class HardGateEndpointTestCase(unittest.TestCase):
"POST", "/projects", {"name": "demo"}, self.owner_key)
self.assertEqual(status, 201)
# 默认即 hard:每个用例开始时清掉模式/保护 env,个别用例
# 再显式设置 record/enforce/off/bogus。
# 再显式设置旧模式值(record/enforce/off/bogus),行为一律 hard。
self._saved_env = {k: os.environ.get(k) for k in
("CLOUD_MERGE_GATE_MODE",
"CLOUD_BRANCH_PROTECTION")}
@@ -368,33 +367,35 @@ class HardGateEndpointTestCase(unittest.TestCase):
self.assertEqual(status, 409)
self.assertEqual(body["merge_gate"]["mode"], "hard")
def test_record_env_preserves_old_no_fields_behavior(self):
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, merged = self._req(
status, body = self._req(
"POST", "/tasks/%s/merge" % task["id"], {}, self.owner_key)
self.assertEqual(status, 200)
self.assertEqual(merged["status"], "merged")
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_still_passes_with_risk(self):
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"}, 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(status, 409)
self.assertIn("branch_protection_unprotected",
body["merge_gate"]["blocks"])
audit = self.auth.list_audit(self.project["id"])
self.assertTrue(any(a["action"] == "merge_gate_risk"
for a in audit))
self.assertFalse(any(a["action"] == "merge_gate_risk"
for a in audit))
def test_enforce_env_blocks_unprotected_even_with_confirm(self):
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()
@@ -402,17 +403,21 @@ class HardGateEndpointTestCase(unittest.TestCase):
"POST", "/tasks/%s/merge" % task["id"],
{"merge_gate_confirm": True}, self.owner_key)
self.assertEqual(status, 409)
self.assertIn("branch_protection_enforce",
# hard 门禁的码,不是旧 enforce 纯函数的 branch_protection_enforce。
self.assertIn("branch_protection_unprotected",
body["merge_gate"]["blocks"])
def test_off_env_is_explicit_opt_down_only(self):
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, merged = self._req(
status, body = 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")
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()