"""anu_v2.tests.test_owner_trigger_only_2554 — OWNER_TRIGGER_ONLY_CAPABILITY core 회귀 (task-2554). 회장 §명시 14장 §10 1:1 박제 (2026-05-11 KST). 본 파일은 system spec §10의 14건 중 1~9, 13~14 (10건)을 다룬다. 나머지는 다음 파일들로 분리: - test_owner_trigger_decision_schema_2554.py - test_owner_trigger_dedupe_2554.py - test_owner_trigger_security_boundaries_2554.py - test_owner_trigger_merge_path_separation_2554.py 본 회귀는 anu_v2/* 모듈만 import 한다 (one-way isolation). """ from __future__ import annotations import json import sys from pathlib import Path import pytest WORKSPACE_ROOT = Path(__file__).resolve().parents[2] if str(WORKSPACE_ROOT) not in sys.path: sys.path.insert(0, str(WORKSPACE_ROOT)) from anu_v2.owner_trigger_only import ( # noqa: E402 ALLOWED_ACTION, CommentBodyViolation, ForbiddenEndpointError, OwnerTriggerOnly, TokenBoundaryViolation, TriggerResult, assert_endpoint_allowed, ) from anu_v2.owner_trigger_audit import OwnerTriggerAudit # noqa: E402 from anu_v2.owner_trigger_decision import DecisionInvalidError # noqa: E402 _HEAD_A = "a" * 40 _HEAD_B = "b" * 40 def _write_decision( tmp_path: Path, *, pr: int = 103, head: str = _HEAD_A, queue_head: bool = True, current_head_confirmed: bool = True, gemini_evidence_fresh: bool = False, nudge_count_for_pr_head: int = 0, allowed_action: str = "POST_GEMINI_REVIEW_TRIGGER_COMMENT", comment_body: str = "/gemini review", allowed: bool = True, task_id: str = "task-2554", ) -> Path: decision = { "schema": "anu_v2.owner_trigger_decision.v1", "task_id": task_id, "pr": pr, "current_head": head, "queue_head": queue_head, "current_head_confirmed": current_head_confirmed, "gemini_evidence_fresh": gemini_evidence_fresh, "nudge_count_for_pr_head": nudge_count_for_pr_head, "allowed_action": allowed_action, "comment_body": comment_body, "allowed": allowed, } path = tmp_path / "decision.json" path.write_text(json.dumps(decision), encoding="utf-8") return path def _build_module( tmp_path: Path, *, posts: list[dict] | None = None, token: str = "dummy-token-VALUE-MUST-NOT-LEAK-1234", ): """OwnerTriggerOnly 인스턴스 + 호출 캡처 stubs. token raw value 인자가 절대 audit/로그에 노출되지 않음을 검증하기 위해 sentinel 포함 (``MUST-NOT-LEAK``). """ if posts is None: posts = [] def http_post(method: str, path: str, body: dict, headers: dict) -> dict: posts.append({"method": method, "path": path, "body": body, "headers": headers}) return {"status": 201, "id": 1234567} def token_provider() -> str: return token audit = OwnerTriggerAudit(workspace_root=tmp_path) mod = OwnerTriggerOnly( workspace_root=tmp_path, http_post=http_post, token_provider=token_provider, audit=audit, ) return mod, posts, audit # ─── test 1: allowed decision → POST /gemini review 만 발생 (§10-1) ────────── def test_allowed_decision_posts_gemini_review_comment_only(tmp_path): decision_path = _write_decision(tmp_path) mod, posts, audit = _build_module(tmp_path) result: TriggerResult = mod.trigger_gemini_review( decision_path=decision_path, owner="Jeon-Jonghyuk", repo="dev_workspace", current_head_actual=_HEAD_A, ) assert result.status == "POSTED" assert result.action == ALLOWED_ACTION assert result.comment_body == "/gemini review" assert result.endpoint == "/repos/Jeon-Jonghyuk/dev_workspace/issues/103/comments" assert len(posts) == 1 call = posts[0] assert call["method"] == "POST" assert call["path"].endswith("/issues/103/comments") assert call["body"] == {"body": "/gemini review"} # token raw value is in headers (구현 시점), but audit는 redact. assert audit.path.exists() rows = [json.loads(line) for line in audit.path.read_text(encoding="utf-8").splitlines() if line] # task-2554+2 §1: http_post 직전 PENDING + 성공 후 POSTED — 2 행. assert len(rows) == 2 assert rows[0]["action"] == ALLOWED_ACTION assert rows[0]["result"] == "PENDING" assert rows[1]["action"] == ALLOWED_ACTION assert rows[1]["result"] == "POSTED" assert rows[1]["token_value_logged"] is False assert "token_hash_prefix" in rows[1] and len(rows[1]["token_hash_prefix"]) == 8 # ─── test 2: comment body가 /gemini review 가 아니면 fail (§10-2) ──────────── def test_comment_body_mismatch_fails_closed(tmp_path): decision_path = _write_decision(tmp_path, comment_body="/gemini approve") mod, posts, _ = _build_module(tmp_path) with pytest.raises(DecisionInvalidError) as exc: mod.trigger_gemini_review( decision_path=decision_path, owner="Jeon-Jonghyuk", repo="dev_workspace", current_head_actual=_HEAD_A, ) assert exc.value.code == "E_COMMENT_BODY" assert posts == [] def test_caller_provided_comment_body_must_match_constant(tmp_path): decision_path = _write_decision(tmp_path) mod, posts, _ = _build_module(tmp_path) with pytest.raises(CommentBodyViolation): mod.trigger_gemini_review( decision_path=decision_path, owner="Jeon-Jonghyuk", repo="dev_workspace", current_head_actual=_HEAD_A, comment_body="hello world", ) assert posts == [] # ─── test 3: queue_head == false fail (§10-3) ──────────────────────────────── def test_non_queue_head_fails_closed(tmp_path): decision_path = _write_decision(tmp_path, queue_head=False) mod, posts, _ = _build_module(tmp_path) with pytest.raises(DecisionInvalidError) as exc: mod.trigger_gemini_review( decision_path=decision_path, owner="Jeon-Jonghyuk", repo="dev_workspace", current_head_actual=_HEAD_A, ) assert exc.value.code == "E_QUEUE_HEAD" assert posts == [] # ─── test 4: current_head mismatch fail (§10-4) ───────────────────────────── def test_current_head_mismatch_fails_closed(tmp_path): decision_path = _write_decision(tmp_path, head=_HEAD_A) mod, posts, _ = _build_module(tmp_path) with pytest.raises(DecisionInvalidError) as exc: mod.trigger_gemini_review( decision_path=decision_path, owner="Jeon-Jonghyuk", repo="dev_workspace", current_head_actual=_HEAD_B, # 다른 head ) assert exc.value.code == "E_HEAD_MISMATCH" assert posts == [] # ─── test 5: gemini_evidence_fresh==true fail (§10-5) ─────────────────────── def test_gemini_evidence_fresh_fails_closed(tmp_path): decision_path = _write_decision(tmp_path, gemini_evidence_fresh=True) mod, posts, _ = _build_module(tmp_path) with pytest.raises(DecisionInvalidError) as exc: mod.trigger_gemini_review( decision_path=decision_path, owner="Jeon-Jonghyuk", repo="dev_workspace", current_head_actual=_HEAD_A, ) assert exc.value.code == "E_GEMINI_FRESH" assert posts == [] # ─── test 7: update-branch 전 trigger 시도 fail (§10-7) ───────────────────── # = current_head_confirmed == false 케이스 (update-branch 직후에만 confirmed=True) def test_update_branch_before_trigger_fails_closed(tmp_path): decision_path = _write_decision(tmp_path, current_head_confirmed=False) mod, posts, _ = _build_module(tmp_path) with pytest.raises(DecisionInvalidError) as exc: mod.trigger_gemini_review( decision_path=decision_path, owner="Jeon-Jonghyuk", repo="dev_workspace", current_head_actual=_HEAD_A, ) assert exc.value.code == "E_HEAD_NOT_CONFIRMED" assert posts == [] # ─── test 8: non-queue-head PR trigger 시도 fail (§10-8) ──────────────────── # §10-3과 비슷하지만 비-queue-head를 명시적으로 표현 (다른 트리거 시도). def test_attempting_to_trigger_non_queue_head_pr_fails_closed(tmp_path): # nudge_count_for_pr_head > 0 으로 다른 시도를 표현 (이미 trigger 한 시도) decision_path = _write_decision(tmp_path, nudge_count_for_pr_head=1, queue_head=True) mod, posts, _ = _build_module(tmp_path) with pytest.raises(DecisionInvalidError) as exc: mod.trigger_gemini_review( decision_path=decision_path, owner="Jeon-Jonghyuk", repo="dev_workspace", current_head_actual=_HEAD_A, ) assert exc.value.code == "E_NUDGE_COUNT" assert posts == [] # ─── test 9: merge/approve/close/reopen endpoint 호출 시 fail (§10-9) ─────── def test_forbidden_endpoints_raise_permission_error(): forbidden = [ ("POST", "/repos/o/r/pulls/103/merge"), ("POST", "/repos/o/r/pulls/103/reviews"), ("POST", "/repos/o/r/git/refs/heads/x"), ("PUT", "/repos/o/r/contents/x.py"), ("PATCH", "/repos/o/r/issues/103"), ("POST", "/repos/o/r/issues/103/labels"), ("DELETE", "/repos/o/r/branches/main"), ("POST", "/repos/o/r/actions/runs/1/rerun"), ("POST", "/repos/o/r/check-runs/1/rerequest"), ("PATCH", "/repos/o/r/pulls/103"), ] for method, path in forbidden: with pytest.raises(ForbiddenEndpointError): assert_endpoint_allowed(method, path) def test_allowed_endpoint_only_post_issues_comments_passes(): # 정확히 1개만 통과 — 본 함수는 raise 안 하면 OK assert_endpoint_allowed("POST", "/repos/o/r/issues/103/comments") def test_module_class_does_not_expose_merge_or_approve(tmp_path): mod, _, _ = _build_module(tmp_path) # 1) 실제 호출 시 항상 PermissionError with pytest.raises(PermissionError): mod.merge(pr=103) with pytest.raises(PermissionError): mod.approve(pr=103) with pytest.raises(PermissionError): mod.close(pr=103) with pytest.raises(PermissionError): mod.reopen(pr=103) # ─── test 14: head 변경 후 stale reset 가능 (§10-14) ──────────────────────── def test_head_change_allows_new_trigger_after_stale_reset(tmp_path): # 첫 trigger decision_path_a = tmp_path / "decision_a.json" decision_path_a.write_text( json.dumps( { "schema": "anu_v2.owner_trigger_decision.v1", "task_id": "task-2554", "pr": 103, "current_head": _HEAD_A, "queue_head": True, "current_head_confirmed": True, "gemini_evidence_fresh": False, "nudge_count_for_pr_head": 0, "allowed_action": "POST_GEMINI_REVIEW_TRIGGER_COMMENT", "comment_body": "/gemini review", "allowed": True, } ), encoding="utf-8", ) mod, posts, _ = _build_module(tmp_path) r1 = mod.trigger_gemini_review( decision_path=decision_path_a, owner="o", repo="r", current_head_actual=_HEAD_A, ) assert r1.status == "POSTED" # head 변경 시 동일 PR 이라도 새 decision 으로 trigger 가능 (stale reset) decision_path_b = tmp_path / "decision_b.json" decision_path_b.write_text( json.dumps( { "schema": "anu_v2.owner_trigger_decision.v1", "task_id": "task-2554", "pr": 103, "current_head": _HEAD_B, "queue_head": True, "current_head_confirmed": True, "gemini_evidence_fresh": False, "nudge_count_for_pr_head": 0, "allowed_action": "POST_GEMINI_REVIEW_TRIGGER_COMMENT", "comment_body": "/gemini review", "allowed": True, } ), encoding="utf-8", ) r2 = mod.trigger_gemini_review( decision_path=decision_path_b, owner="o", repo="r", current_head_actual=_HEAD_B, ) assert r2.status == "POSTED" assert len(posts) == 2 # ─── token boundary smoke check — 자세한 cases는 security_boundaries 파일 ─── def test_token_provider_returning_empty_fails_closed(tmp_path): decision_path = _write_decision(tmp_path) mod, posts, _ = _build_module(tmp_path, token="") with pytest.raises(TokenBoundaryViolation): mod.trigger_gemini_review( decision_path=decision_path, owner="o", repo="r", current_head_actual=_HEAD_A, ) assert posts == []