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:
1 parent
593197a4a2
commit
e6a1346420
8 files changed
+200
-183
No files matched your search
@@ -601,10 +601,13 @@ class EndpointChainTestCase(unittest.TestCase):
|
||||
self._hard_body(), self.owner_key)
|
||||
self.assertEqual(status, 200, merged)
|
||||
|
||||
def test_pushback_blocks_even_in_record_mode(self):
|
||||
def test_pushback_blocks_with_full_hard_evidence(self):
|
||||
# A1 改写:旧用例 test_pushback_blocks_even_in_record_mode
|
||||
# 已改名。本用例的性质不变——链上推回阻塞时,即使五项
|
||||
# 证据全齐也 409;只是不再经 record 模式证明“与模式无关”,
|
||||
# 因为服务端已无其它模式可比。
|
||||
task = self._completed_task()
|
||||
self._block_task(task)
|
||||
os.environ["CLOUD_MERGE_GATE_MODE"] = "record"
|
||||
status, blocked = self._req(
|
||||
"POST", "/tasks/%s/merge" % task["id"],
|
||||
self._hard_body(), self.owner_key)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -1,12 +1,16 @@
|
||||
"""主分支保护 + 目录隔离门禁(cloud/merge_gate)与技能工程双向断言。
|
||||
|
||||
门禁口径:
|
||||
- 目录隔离复用 v0.3 技能文本的 git diff --name-only 检查,不另造;
|
||||
服务端 POST /tasks/{id}/merge 在此之前只改状态、不查目录、不查保护。
|
||||
- 保护默认 record:未保护 + 无确认阻断,未保护 + 确认放行并记风险;
|
||||
enforce 下未保护一律阻断;unknown 记风险、不阻断(代码无法核实
|
||||
GitHub 保护,unknown 不等于已保护,也不等于未保护)。
|
||||
- 显式确认只覆盖保护风险,绝不覆盖目录违规。
|
||||
门禁口径(纯函数部分,见 A1 声明):
|
||||
- 本文件的纯函数用例测 cloud/merge_gate 本身:目录隔离复用 v0.3
|
||||
技能文本的 git diff --name-only 检查,不另造;纯函数内 record
|
||||
默认:未保护 + 无确认阻断,未保护 + 确认放行并记风险;
|
||||
enforce 下未保护一律阻断;unknown 记风险、不阻断。
|
||||
- 这些纯函数语义不是服务端行为:自 A1 起服务端 POST
|
||||
/tasks/{id}/merge 一律按 hard 求值,旧模式环境变量在服务端被
|
||||
忽略;本文件末尾的服务端用例已按新行为改写(旧 record/enforce
|
||||
服务端用例不再断言放行)。
|
||||
- 显式确认只覆盖保护风险,绝不覆盖目录违规(纯函数内;服务端
|
||||
hard 下确认不覆盖任何一项)。
|
||||
技能工程:
|
||||
- 正文与附录双向断言:每个新关键词在正文、附录都要出现,失败
|
||||
信息指明缺失的一侧(正文缺失 / 附录缺失),不是笼统的 assertIn。
|
||||
@@ -260,17 +264,19 @@ class MergeEndpointGateTestCase(unittest.TestCase):
|
||||
conn.close()
|
||||
return task
|
||||
|
||||
def test_no_gate_fields_behaves_like_old_endpoint(self):
|
||||
# 本用例测的是 record 模式的过渡语义(#209 原意):默认已是
|
||||
# hard,见 test_merge_gate_hard.py 的默认 409 用例;这里显式
|
||||
# 切 record 才沿用“无证据放行、只附带门禁信息”。
|
||||
def test_record_env_no_longer_passes_no_fields(self):
|
||||
# A1:服务端不再有 record 模式。旧用例名
|
||||
# test_no_gate_fields_behaves_like_old_endpoint 已改名改语义:
|
||||
# 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.assertIn("merge_gate", merged)
|
||||
self.assertEqual(status, 409)
|
||||
self.assertEqual(body["merge_gate"]["mode"], "hard")
|
||||
_s, got = self._req("GET", "/tasks/%s" % task["id"],
|
||||
key=self.owner_key)
|
||||
self.assertEqual(got["status"], "completed")
|
||||
|
||||
def test_directory_violation_blocks_without_changing_status(self):
|
||||
task = self._completed_task()
|
||||
@@ -286,28 +292,30 @@ class MergeEndpointGateTestCase(unittest.TestCase):
|
||||
key=self.owner_key)
|
||||
self.assertEqual(got["status"], "completed")
|
||||
|
||||
def test_unprotected_confirm_allows_and_audits_risk(self):
|
||||
# record 模式语义:硬模式下未保护 + 确认仍阻断,见硬门禁测试。
|
||||
def test_record_env_unprotected_confirm_no_longer_passes(self):
|
||||
# A1:旧用例 test_unprotected_confirm_allows_and_audits_risk
|
||||
# 已改名改语义。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(merged["merge_gate"]["protection"], "unprotected")
|
||||
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_even_with_confirm(self):
|
||||
def test_enforce_env_is_ignored_and_hard_blocks(self):
|
||||
# A1:旧用例 test_enforce_env_blocks_even_with_confirm 改写:
|
||||
# enforce 环境变量被忽略,阻断码是 hard 门禁的
|
||||
# branch_protection_unprotected,不是纯函数的
|
||||
# branch_protection_enforce。
|
||||
os.environ["CLOUD_MERGE_GATE_MODE"] = "enforce"
|
||||
os.environ["CLOUD_BRANCH_PROTECTION"] = "unprotected"
|
||||
task = self._completed_task()
|
||||
@@ -316,7 +324,7 @@ class MergeEndpointGateTestCase(unittest.TestCase):
|
||||
{"merge_gate_confirm": True},
|
||||
self.owner_key)
|
||||
self.assertEqual(status, 409)
|
||||
self.assertIn("branch_protection_enforce",
|
||||
self.assertIn("branch_protection_unprotected",
|
||||
body["merge_gate"]["blocks"])
|
||||
|
||||
|
||||
|
||||
@@ -4,10 +4,10 @@
|
||||
(本地或 Docker 容器内)执行,成功后再调本接口把云端
|
||||
任务从 completed 改为 merged。规则:404/401/403/409
|
||||
语义、maintainer 以上才能合并、审计里有 merge_task。
|
||||
门禁注意:默认模式已是 hard(见 test_merge_gate_hard.py),
|
||||
门禁注意:服务端只有 hard(见 test_merge_gate_hard.py),
|
||||
所以本文件里期望 200 的合并调用必须带完整硬门禁证据;
|
||||
无证据调用在默认下 409(本文件末尾两项新用例钉住它),
|
||||
record 下的旧行为由硬门禁测试文件覆盖。
|
||||
旧 record 下的“无证据放行”已随 A1 删模式消失,不再覆盖。
|
||||
"""
|
||||
|
||||
import http.client
|
||||
@@ -176,13 +176,14 @@ class MergeTestCase(unittest.TestCase):
|
||||
"GET", "/tasks/%s" % task["id"], key=self.owner_key)
|
||||
self.assertEqual(got["status"], "completed")
|
||||
|
||||
def test_merge_record_env_preserves_old_no_evidence_call(self):
|
||||
def test_merge_record_env_no_longer_passes_no_evidence_call(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")
|
||||
|
||||
def test_merge_missing_and_no_key(self):
|
||||
status, _ = self._req(
|
||||
|
||||
@@ -546,13 +546,25 @@ class MergeLoopTestCase(_Base):
|
||||
self.assertIn("files_changed must be a list", payload["error"])
|
||||
|
||||
def test_merge_pushback_unavailable_409(self):
|
||||
# A1 改写:旧用例以 CLOUD_MERGE_GATE_MODE=off + 空 body 走到
|
||||
# 推回检查;服务端删模式后空 body 在硬门禁体校验前已 400,
|
||||
# 故改用完整硬证据 body,仍钉“推回状态不可用 → 409”这一性质。
|
||||
self._project()
|
||||
self._task(status="completed")
|
||||
h = self._h(body={})
|
||||
with mock.patch.dict(os.environ,
|
||||
{"CLOUD_MERGE_GATE_MODE": "off"}), \
|
||||
mock.patch("cloud.server._pushback_block",
|
||||
side_effect=RuntimeError("chain down")):
|
||||
h = self._h(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,
|
||||
})
|
||||
with mock.patch("cloud.server._pushback_block",
|
||||
side_effect=RuntimeError("chain down")):
|
||||
h._merge_task("u1", "t1")
|
||||
code, payload = self._last(h)
|
||||
self.assertEqual(code, 409)
|
||||
@@ -561,11 +573,24 @@ class MergeLoopTestCase(_Base):
|
||||
self.assertTrue(payload["audit_pushback"]["merge_blocked"])
|
||||
|
||||
def test_merge_pushback_unavailable_audit_fails_too(self):
|
||||
# A1 改写:同上,off + 空 body 改为完整硬证据 body,仍钉
|
||||
# “审计本身也失败时 handler 仍回同一 409”。
|
||||
# The audit best-effort except-pass: even when the audit
|
||||
# write itself raises, the handler still answers the same 409.
|
||||
self._project()
|
||||
self._task(status="completed")
|
||||
h = self._h(body={})
|
||||
h = self._h(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,
|
||||
})
|
||||
real_auth = server_mod._auth_store
|
||||
calls = {"n": 0}
|
||||
|
||||
@@ -574,10 +599,8 @@ class MergeLoopTestCase(_Base):
|
||||
if calls["n"] == 1:
|
||||
return real_auth(db_path)
|
||||
raise RuntimeError("audit down")
|
||||
with mock.patch.dict(os.environ,
|
||||
{"CLOUD_MERGE_GATE_MODE": "off"}), \
|
||||
mock.patch("cloud.server._pushback_block",
|
||||
side_effect=RuntimeError("chain down")), \
|
||||
with mock.patch("cloud.server._pushback_block",
|
||||
side_effect=RuntimeError("chain down")), \
|
||||
mock.patch("cloud.server._auth_store",
|
||||
side_effect=_auth_store):
|
||||
h._merge_task("u1", "t1")
|
||||
|
||||
Reference in new issue
Block a user