host
a605ff2f92
Host auto executor (cloud/full_auto.py, stdlib only): default off (FULL_AUTO_ENABLED + verified installation), forbidden actions fail-closed (delete repo, force-push, visibility change, secret read/expose, branch-protection delete/bypass, GitHub account auto-registration), token only in request header, audit log redacted, branch protection requires PR + strict checks + no force-push + no deletions + enforce_admins, merge gate (CI green + directory check + contract version + coordinator approval), CI fix loop max 3 then escalate, rollback only via revert PR or previous tag. Skill: v0.4 full-auto mode, coordinator reviews, host executor executes, task agent never executes merge/deploy; repo-creation honesty branch (org via App token, user repo needs_user_once); no web auto-registration of GitHub accounts. Tests: 19 new, cloud full 1451 OK (skipped=4).
221 lines
8.8 KiB
Python
221 lines
8.8 KiB
Python
"""全自动执行器（cloud/full_auto.py）的轻量测试。
|
|
|
|
覆盖护栏（默认关闭、禁止/未知动作）、合并门、自动配置
|
|
计划顺序与分支保护 payload、计划执行（假客户端，不发
|
|
真实网络）、CI 修复循环限次、发版/回滚不 force-push、
|
|
审计记录不含 token。另断言技能正文与附录都含有全自动
|
|
模式的关键概念（与 test_software_engineering_github.py
|
|
同款正文+附录双向断言，附录没同步就会失败）。
|
|
"""
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from cloud import full_auto
|
|
from cloud.full_auto import (
|
|
FullAutoDenied,
|
|
FullAutoError,
|
|
append_audit_log,
|
|
audit_record,
|
|
auto_config_plan,
|
|
branch_protection_payload,
|
|
ci_fix_decision,
|
|
is_full_auto_enabled,
|
|
merge_gate,
|
|
release_plan,
|
|
rollback_plan,
|
|
run_plan,
|
|
validate_action,
|
|
)
|
|
|
|
SKILL = Path(__file__).resolve().parents[1] / "skills" / "software-engineering" / "SKILL.md"
|
|
|
|
FULL_AUTO_KEYWORDS = ("全自动模式", "FULL_AUTO_ENABLED",
|
|
"一次授权", "CI 自动修复", "回滚")
|
|
|
|
|
|
class FullAutoGateTestCase(unittest.TestCase):
|
|
def test_default_off_without_flag(self):
|
|
self.assertFalse(is_full_auto_enabled({}, True))
|
|
self.assertFalse(is_full_auto_enabled(
|
|
{"FULL_AUTO_ENABLED": "1"}, False))
|
|
self.assertFalse(is_full_auto_enabled(
|
|
{"FULL_AUTO_ENABLED": "0"}, True))
|
|
|
|
def test_enabled_only_with_flag_and_verified_installation(self):
|
|
self.assertTrue(is_full_auto_enabled(
|
|
{"FULL_AUTO_ENABLED": "1"}, True))
|
|
self.assertTrue(is_full_auto_enabled(
|
|
{"FULL_AUTO_ENABLED": "true"}, True))
|
|
|
|
def test_forbidden_and_unknown_actions_denied(self):
|
|
for action in ("delete_repo", "force_push", "force_push_main",
|
|
"change_visibility", "read_secret",
|
|
"expose_secret", "delete_branch_protection",
|
|
"bypass_branch_protection",
|
|
"register_github_account", "nonsense_action"):
|
|
with self.assertRaises(FullAutoDenied, msg=action):
|
|
validate_action(action)
|
|
|
|
def test_allowed_actions_pass(self):
|
|
for action in ("create_pr", "merge_pr", "create_release",
|
|
"rollback_to_tag", "set_branch_protection"):
|
|
self.assertTrue(validate_action(action))
|
|
|
|
|
|
class MergeGateTestCase(unittest.TestCase):
|
|
def test_all_green_passes(self):
|
|
passed, reasons = merge_gate(
|
|
{"verify": "success"}, ("verify",), True, True, True)
|
|
self.assertTrue(passed)
|
|
self.assertEqual(reasons, [])
|
|
|
|
def test_any_missing_gate_fails(self):
|
|
cases = [
|
|
({"verify": "failure"}, ("verify",), True, True, True),
|
|
({"verify": "success"}, ("verify",), False, True, True),
|
|
({"verify": "success"}, ("verify",), True, False, True),
|
|
({"verify": "success"}, ("verify",), True, True, False),
|
|
({}, ("verify",), True, True, True),
|
|
]
|
|
for checks, required, d, c, a in cases:
|
|
passed, reasons = merge_gate(checks, required, d, c, a)
|
|
self.assertFalse(passed)
|
|
self.assertTrue(reasons)
|
|
|
|
|
|
class AutoConfigPlanTestCase(unittest.TestCase):
|
|
def test_plan_order_and_branch_protection(self):
|
|
plan = auto_config_plan("owner/repo", "main", org="myorg",
|
|
repo_exists=False,
|
|
ci_workflow_present=False)
|
|
steps = [s["step"] for s in plan]
|
|
self.assertEqual(steps, ["ensure_repo", "set_default_branch",
|
|
"set_branch_protection",
|
|
"add_ci_workflow", "verify_webhook"])
|
|
protection = plan[2]["payload_summary"]
|
|
self.assertTrue(protection["enforce_admins"])
|
|
self.assertTrue(
|
|
protection["required_status_checks"]["strict"])
|
|
self.assertEqual(
|
|
protection["required_status_checks"]["contexts"],
|
|
["verify"])
|
|
self.assertFalse(protection["allow_force_pushes"])
|
|
self.assertFalse(protection["allow_deletions"])
|
|
self.assertIsNotNone(
|
|
protection["required_pull_request_reviews"])
|
|
|
|
def test_user_repo_without_org_is_user_once_not_fake_api(self):
|
|
plan = auto_config_plan("owner/repo", "main", org=None,
|
|
repo_exists=False)
|
|
first = plan[0]
|
|
self.assertEqual(first["manual"], "user_once")
|
|
self.assertIsNone(first["api"])
|
|
|
|
def test_existing_repo_skips_ensure_step(self):
|
|
plan = auto_config_plan("owner/repo", "main", org=None,
|
|
repo_exists=True)
|
|
self.assertNotIn("ensure_repo",
|
|
[s["step"] for s in plan])
|
|
|
|
def test_empty_required_checks_is_incomplete(self):
|
|
with self.assertRaises(FullAutoError):
|
|
branch_protection_payload([])
|
|
|
|
|
|
class RunPlanTestCase(unittest.TestCase):
|
|
def _fake_client(self, calls, fail_on=None):
|
|
def client(method, url, headers, body):
|
|
calls.append((method, url, headers))
|
|
if fail_on and fail_on in url:
|
|
return 422, {"message": "validation failed"}
|
|
return 201, {"ok": True}
|
|
return client
|
|
|
|
def test_executes_in_order_and_token_only_in_header(self):
|
|
calls = []
|
|
plan = auto_config_plan("owner/repo", "main", org="myorg",
|
|
repo_exists=False)
|
|
results = run_plan(self._fake_client(calls), "tok-123", plan)
|
|
self.assertTrue(all(r["ok"] for r in results))
|
|
self.assertEqual(calls[0][0], "POST")
|
|
self.assertIn("/orgs/myorg/repos", calls[0][1])
|
|
self.assertEqual(calls[0][2]["Authorization"],
|
|
"Bearer tok-123")
|
|
dumped = json.dumps(results)
|
|
self.assertNotIn("tok-123", dumped)
|
|
|
|
def test_stops_on_first_failure(self):
|
|
calls = []
|
|
plan = auto_config_plan("owner/repo", "main", org="myorg",
|
|
repo_exists=False)
|
|
with self.assertRaises(FullAutoError):
|
|
run_plan(self._fake_client(calls, fail_on="protection"),
|
|
"tok-123", plan)
|
|
self.assertEqual(len(calls), 3) # ensure + branch + protection
|
|
|
|
def test_missing_token_denied(self):
|
|
plan = auto_config_plan("owner/repo")
|
|
with self.assertRaises(FullAutoDenied):
|
|
run_plan(self._fake_client([]), "", plan)
|
|
|
|
|
|
class CiFixAndReleaseTestCase(unittest.TestCase):
|
|
def test_fix_loop_max_three_then_escalate(self):
|
|
for attempt in (1, 2, 3):
|
|
decision = ci_fix_decision(attempt, ["verify"])
|
|
self.assertEqual(decision["action"], "fix_and_rerun")
|
|
decision = ci_fix_decision(4, ["verify"])
|
|
self.assertEqual(decision["action"], "escalate")
|
|
self.assertEqual(decision["reason"],
|
|
"max_fix_attempts_exceeded")
|
|
|
|
def test_release_notes_from_merged_prs_only(self):
|
|
plan = release_plan("v0.1.0", [{"number": 12, "title": "feat: x"}])
|
|
self.assertIn("#12 feat: x", plan["release_notes"])
|
|
self.assertIn("never force-push", plan["rollback"])
|
|
|
|
def test_rollback_requires_previous_tag(self):
|
|
self.assertEqual(rollback_plan("v0.0.9")["previous_tag"],
|
|
"v0.0.9")
|
|
with self.assertRaises(FullAutoError):
|
|
rollback_plan("")
|
|
|
|
|
|
class AuditTestCase(unittest.TestCase):
|
|
def test_audit_record_redacts_token(self):
|
|
record = audit_record("merge_pr", "owner/repo",
|
|
"full-auto-executor:123", "ok",
|
|
{"token": "tok-123",
|
|
"note": "Bearer tok-123 here"})
|
|
dumped = json.dumps(record)
|
|
self.assertNotIn("tok-123", dumped)
|
|
self.assertIn("Bearer [redacted]", dumped)
|
|
|
|
def test_append_audit_log_jsonl(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = str(Path(tmp) / "audit.jsonl")
|
|
record = audit_record("create_pr", "owner/repo",
|
|
"full-auto-executor:1", "ok")
|
|
append_audit_log(path, record)
|
|
lines = Path(path).read_text(encoding="utf-8").strip().split("\n")
|
|
self.assertEqual(len(lines), 1)
|
|
self.assertEqual(json.loads(lines[0])["action"],
|
|
"create_pr")
|
|
|
|
|
|
class SkillFullAutoTestCase(unittest.TestCase):
|
|
def test_full_auto_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 FULL_AUTO_KEYWORDS:
|
|
self.assertIn(kw, main, kw)
|
|
self.assertIn(kw, appendix, "appendix: " + kw)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|