"""全自动执行器（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.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()