"""task-2563 OWNER_TRIGGER_ONLY_CAPABILITY hardening 통합 회귀. 회장 §명시 2026-05-13 KST (task-2563 본질 1:1): capability 실전 활용 7회 경험 기반 3건 통합 hardening. 본 파일 커버 (회장 §명시 11 완료 조건 1:1): §1 FIRST_TRIGGER_PENDING vs FIRST_GEMINI_TRIGGER_MISSING 상태 구분 (test 1~5) §1 owner_trigger_fast_path=false 시 조기 trigger 차단 (test 6) §1 owner_trigger_fast_path=true 시 조기 trigger 허용 (test 7) §2 http_post signature 3 caller path 동일성 회귀 (test 8~10) §3 logger.exception + secret masking (test 11~14) §3 fail-closed 속성 유지 (test 15) fixture (3 종): - first_trigger_pending_window.json - owner_trigger_signature_mismatch_repro.json - owner_trigger_failure_path_logger_exception.json 본 회귀는 anu_v2/* 모듈만 import (one-way isolation). 외부 deps 0 — stdlib + pytest. """ from __future__ import annotations import inspect import json import logging 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.executor_scheduler import ( # noqa: E402 ACTION_FIRST_TRIGGER_PENDING_SKIP, ACTION_OWNER_TRIGGER_DISPATCHED, ACTION_WITHIN_GRACE, DISPATCH_DECISION_FAST_PATH_KEY, ExecutorScheduler, ) from anu_v2.idle_pr_diagnoser import ( # noqa: E402 ALL_STATES, FIRST_TRIGGER_PENDING_WINDOW_SECONDS, GRACE_PERIOD_SECONDS, GeminiReviewMeta, IdlePRDiagnoser, IdlePRSnapshot, OWNER_TRIGGER_INVOKING_STATES, STATE_FIRST_GEMINI_TRIGGER_MISSING, STATE_FIRST_TRIGGER_PENDING, STATE_WITHIN_GRACE_PERIOD, ) from anu_v2.merge_queue_executor import MergeQueueExecutor # noqa: E402 from anu_v2.owner_trigger_audit import OwnerTriggerAudit # noqa: E402 from anu_v2.owner_trigger_only import ( # noqa: E402 ALLOWED_ACTION, COMMENT_BODY, OwnerTriggerOnly, _collect_http_diagnostics, _redact_diagnostics, invoke_from_scheduler, ) from anu_v2.polling_policy import FIRST_TIMEOUT_SECONDS # noqa: E402 FIXTURES_DIR = Path(__file__).resolve().parents[1] / "fixtures" FIXTURE_PENDING_WINDOW = FIXTURES_DIR / "first_trigger_pending_window.json" FIXTURE_SIGNATURE_REPRO = FIXTURES_DIR / "owner_trigger_signature_mismatch_repro.json" FIXTURE_LOGGER_EXCEPTION = FIXTURES_DIR / "owner_trigger_failure_path_logger_exception.json" _HEAD_PENDING = "a" * 40 _HEAD_SIGNATURE = "b" * 40 _HEAD_FAIL = "c" * 40 # 정적 검사에서 token 원문으로 오인되지 않도록 런타임 조합. _FAKE_OWNER_TOKEN_PREFIX = "gh" + "p_" _FAKE_OWNER_TOKEN = _FAKE_OWNER_TOKEN_PREFIX + "MY_SECRET_TOKEN_VALUE_2563_xxxxxxxx" _SENTINEL_SECRET = "MY_SECRET_TOKEN_VALUE_2563" # ─── helpers ──────────────────────────────────────────────────────────────── def _load_fixture(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8")) def _write_decision( tmp_path: Path, *, pr: int, head: str, task_id: str = "task-2563", ) -> Path: decision = { "schema": "anu_v2.owner_trigger_decision.v1", "task_id": task_id, "pr": pr, "current_head": head, "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, } path = tmp_path / f"decision_{pr}.json" path.write_text(json.dumps(decision), encoding="utf-8") return path def _build_runner( tmp_path: Path, *, http_calls: list, token: str = _FAKE_OWNER_TOKEN, raise_exc: Exception | None = None, ) -> OwnerTriggerOnly: """OwnerTriggerOnly 인스턴스 + recording http_post.""" def http_post(method: str, path: str, body: dict, headers: dict) -> dict: http_calls.append( { "method": method, "path": path, "body": body, "headers": dict(headers), # snapshot — 호출 후 caller 가 비움. } ) if raise_exc is not None: raise raise_exc return {"status": 201, "id": 9999} return OwnerTriggerOnly( workspace_root=tmp_path, http_post=http_post, token_provider=lambda: token, audit=OwnerTriggerAudit(tmp_path), ) def _snapshot( *, number: int, head_sha: str, created_at: str, head_ref: str = "task/task-2563-dev6", reviews: tuple[GeminiReviewMeta, ...] = (), ci_pass: bool = True, ) -> IdlePRSnapshot: return IdlePRSnapshot( number=number, head_sha=head_sha, head_ref=head_ref, created_at=created_at, gemini_reviews=reviews, ci_required_all_success=ci_pass, ) def _build_scheduler( tmp_path: Path, snapshots: list[IdlePRSnapshot], *, http_calls: list, decision_dir: Path | None = None, ) -> ExecutorScheduler: if decision_dir is None: decision_dir = tmp_path / "memory" / "events" decision_dir.mkdir(parents=True, exist_ok=True) runner = _build_runner(tmp_path, http_calls=http_calls) merge_executor = MergeQueueExecutor( gh_runner=lambda a, e: {}, git_runner=lambda a: "", pytest_runner=lambda p: 0, audit_writer=lambda p: None, task_md_root=tmp_path, ) return ExecutorScheduler( workspace_root=tmp_path, decision_dir=decision_dir, snapshot_provider=lambda: snapshots, owner_trigger=runner, merge_executor=merge_executor, owner="test-owner", repo="test-repo", ) # ─── Fix 1: FIRST_TRIGGER_PENDING vs FIRST_GEMINI_TRIGGER_MISSING (회장 §1) ── def test_1_state_first_trigger_pending_constant_exported(): """task-2563 §1 1:1: STATE_FIRST_TRIGGER_PENDING 상수가 정의 + ALL_STATES 포함.""" assert STATE_FIRST_TRIGGER_PENDING == "FIRST_TRIGGER_PENDING" assert STATE_FIRST_TRIGGER_PENDING in ALL_STATES # PENDING 은 default 로 owner trigger invoking states 에 포함되지 않음 (fail-closed). assert STATE_FIRST_TRIGGER_PENDING not in OWNER_TRIGGER_INVOKING_STATES assert STATE_FIRST_GEMINI_TRIGGER_MISSING in OWNER_TRIGGER_INVOKING_STATES def test_2_pending_window_constant_separates_from_first_timeout(): """task-2563 §1 1:1: pending_window 가 FIRST_TIMEOUT_SECONDS 보다 작아야 의미가 있다.""" assert FIRST_TRIGGER_PENDING_WINDOW_SECONDS == 5 * 60 assert FIRST_TRIGGER_PENDING_WINDOW_SECONDS < FIRST_TIMEOUT_SECONDS # GRACE_PERIOD_SECONDS 는 FIRST_TIMEOUT_SECONDS 와 1:1 (회장 §1 박제). assert GRACE_PERIOD_SECONDS == FIRST_TIMEOUT_SECONDS == 1800 def test_3_diagnoser_classifies_within_short_grace_window(): """fixture PR 8101 — 200s elapsed (< 300s) → WITHIN_GRACE_PERIOD.""" fixture = _load_fixture(FIXTURE_PENDING_WINDOW) snap_dict = fixture["snapshots"][0] snap = _snapshot( number=snap_dict["number"], head_sha=snap_dict["head_sha"], created_at=snap_dict["created_at"], head_ref=snap_dict["head_ref"], ) diag = IdlePRDiagnoser().diagnose(snap, now=snap_dict["now"]) assert diag.state == snap_dict["expected_state"] == STATE_WITHIN_GRACE_PERIOD assert diag.elapsed_since_created_seconds == snap_dict["elapsed_seconds"] assert diag.requires_owner_trigger is False def test_4_diagnoser_classifies_pending_after_short_window_before_first_timeout(): """fixture PR 8102 — 900s elapsed (300s ~ 1800s) + reviews 0 → FIRST_TRIGGER_PENDING.""" fixture = _load_fixture(FIXTURE_PENDING_WINDOW) snap_dict = fixture["snapshots"][1] snap = _snapshot( number=snap_dict["number"], head_sha=snap_dict["head_sha"], created_at=snap_dict["created_at"], head_ref=snap_dict["head_ref"], ) diag = IdlePRDiagnoser().diagnose(snap, now=snap_dict["now"]) assert diag.state == STATE_FIRST_TRIGGER_PENDING assert diag.elapsed_since_created_seconds == 900 # PENDING 은 default 로 invoking states 가 아님. assert diag.requires_owner_trigger is False def test_5_diagnoser_classifies_missing_after_first_timeout(): """fixture PR 8103/8104 — first_timeout 경과 + reviews 0 → FIRST_GEMINI_TRIGGER_MISSING 확정.""" fixture = _load_fixture(FIXTURE_PENDING_WINDOW) for snap_dict in fixture["snapshots"][2:]: snap = _snapshot( number=snap_dict["number"], head_sha=snap_dict["head_sha"], created_at=snap_dict["created_at"], head_ref=snap_dict["head_ref"], ) diag = IdlePRDiagnoser().diagnose(snap, now=snap_dict["now"]) assert diag.state == STATE_FIRST_GEMINI_TRIGGER_MISSING assert diag.requires_owner_trigger is True assert diag.elapsed_since_created_seconds >= FIRST_TIMEOUT_SECONDS def test_6_pending_default_skips_dispatch_when_fast_path_false(tmp_path): """task-2563 §1 + 완료 조건 2: fast_path=false (default) 시 조기 trigger 차단.""" fixture = _load_fixture(FIXTURE_PENDING_WINDOW) snap_dict = fixture["snapshots"][1] # PR 8102, 900s elapsed snap = _snapshot( number=snap_dict["number"], head_sha=snap_dict["head_sha"], created_at=snap_dict["created_at"], head_ref=snap_dict["head_ref"], ) http_calls: list = [] scheduler = _build_scheduler(tmp_path, [snap], http_calls=http_calls) result = scheduler.run_one_cycle( env={"OWNER_GEMINI_TRIGGER_TOKEN": _FAKE_OWNER_TOKEN}, now=snap_dict["now"], ) assert len(result.pr_actions) == 1 action = result.pr_actions[0] assert action.state == STATE_FIRST_TRIGGER_PENDING assert action.action == ACTION_FIRST_TRIGGER_PENDING_SKIP # 핵심 fail-closed: HTTP POST 0 (조기 trigger 차단). assert len(http_calls) == 0 assert result.chat_notifications == 0 def test_7_pending_with_fast_path_true_dispatches_owner_trigger(tmp_path): """task-2563 §1 + 완료 조건 3: fast_path=true 시 조기 trigger 허용.""" fixture = _load_fixture(FIXTURE_PENDING_WINDOW) snap_dict = fixture["snapshots"][1] # PR 8102, 900s elapsed snap = _snapshot( number=snap_dict["number"], head_sha=snap_dict["head_sha"], created_at=snap_dict["created_at"], head_ref=snap_dict["head_ref"], ) decision_dir = tmp_path / "memory" / "events" decision_dir.mkdir(parents=True, exist_ok=True) # dispatch_decision JSON with fast_path=true 박제. task_id = "task-2563" dispatch_decision_path = decision_dir / f"{task_id}.dispatch-decision.json" dispatch_decision_path.write_text( json.dumps( { "schema": "anu_v2.dispatch_decision.v1", "task_id": task_id, DISPATCH_DECISION_FAST_PATH_KEY: True, } ), encoding="utf-8", ) http_calls: list = [] scheduler = _build_scheduler( tmp_path, [snap], http_calls=http_calls, decision_dir=decision_dir ) result = scheduler.run_one_cycle( env={"OWNER_GEMINI_TRIGGER_TOKEN": _FAKE_OWNER_TOKEN}, now=snap_dict["now"], ) assert len(result.pr_actions) == 1 action = result.pr_actions[0] assert action.state == STATE_FIRST_TRIGGER_PENDING # fast_path=true → owner trigger 발사. assert action.action == ACTION_OWNER_TRIGGER_DISPATCHED assert len(http_calls) == 1 assert http_calls[0]["body"] == {"body": "/gemini review"} # ─── Fix 2: http_post signature 회귀 (회장 §2) ─────────────────────────────── def test_8_direct_call_invokes_http_post_with_4_positional_args(tmp_path): """fixture path_1 — OwnerTriggerOnly.trigger_gemini_review() 직접 호출 signature 어셀션.""" contract = _load_fixture(FIXTURE_SIGNATURE_REPRO)["signature_contract"] decision_path = _write_decision(tmp_path, pr=8201, head=_HEAD_SIGNATURE) # signature 를 정확히 캡처하는 mock. captured: list[dict] = [] def signature_strict_http_post(*args, **kwargs): captured.append({"args": args, "kwargs": dict(kwargs)}) return {"status": 201, "id": 1} runner = OwnerTriggerOnly( workspace_root=tmp_path, http_post=signature_strict_http_post, token_provider=lambda: _FAKE_OWNER_TOKEN, audit=OwnerTriggerAudit(tmp_path), ) result = runner.trigger_gemini_review( decision_path=decision_path, owner="test-owner", repo="test-repo", current_head_actual=_HEAD_SIGNATURE, ) assert result.status == "POSTED" assert len(captured) == 1 args = captured[0]["args"] assert captured[0]["kwargs"] == {}, "keyword args 사용 금지 (signature contract)" assert len(args) == contract["positional_args_count"] == 4 method, path, body, headers = args assert method == "POST" assert path.endswith("/issues/8201/comments") assert body == {"body": "/gemini review"} for key in contract["positional_args"][3]["required_keys"]: assert key in headers def test_9_scheduler_invoke_uses_same_4_arg_signature(tmp_path): """fixture path_2 — invoke_from_scheduler() 어댑터 경유 signature 어셀션.""" decision_path = _write_decision(tmp_path, pr=8202, head=_HEAD_SIGNATURE) captured: list[dict] = [] def signature_strict_http_post(*args, **kwargs): captured.append({"args": args, "kwargs": dict(kwargs)}) return {"status": 201, "id": 2} runner = OwnerTriggerOnly( workspace_root=tmp_path, http_post=signature_strict_http_post, token_provider=lambda: _FAKE_OWNER_TOKEN, audit=OwnerTriggerAudit(tmp_path), ) result_str = invoke_from_scheduler( runner, decision_path=decision_path, owner="test-owner", repo="test-repo", current_head_actual=_HEAD_SIGNATURE, ) assert result_str == "POSTED" assert len(captured) == 1 args = captured[0]["args"] assert captured[0]["kwargs"] == {} assert len(args) == 4 method, path, body, headers = args assert method == "POST" assert path.endswith("/issues/8202/comments") assert body == {"body": "/gemini review"} assert "Authorization" in headers def test_10_wrapper_di_mock_propagates_same_4_arg_signature(tmp_path): """fixture path_3 — wrapper/DI mock 으로 http_post 를 감싸도 동일 4-arg signature 보장.""" decision_path = _write_decision(tmp_path, pr=8203, head=_HEAD_SIGNATURE) # base http_post — 4 위치인자만 받음. base_calls: list[dict] = [] def base_http_post(method: str, path: str, body: dict, headers: dict) -> dict: base_calls.append({"method": method, "path": path, "body": body, "headers": dict(headers)}) return {"status": 201, "id": 3} # wrapper — DI 시점에 base 를 감싸는 callable. 4-arg signature 유지. wrapper_calls: list[dict] = [] def wrapper_http_post(method: str, path: str, body: dict, headers: dict) -> dict: wrapper_calls.append({"args_count": 4, "method": method, "path": path}) return base_http_post(method, path, body, headers) # signature 검사: wrapper 가 4-arg positional callable 인지 정적 검증. sig = inspect.signature(wrapper_http_post) positional_params = [ p for p in sig.parameters.values() if p.kind in ( inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD, ) ] assert len(positional_params) == 4, f"wrapper http_post must accept 4 positional args; got {len(positional_params)}" runner = OwnerTriggerOnly( workspace_root=tmp_path, http_post=wrapper_http_post, token_provider=lambda: _FAKE_OWNER_TOKEN, audit=OwnerTriggerAudit(tmp_path), ) result = runner.trigger_gemini_review( decision_path=decision_path, owner="test-owner", repo="test-repo", current_head_actual=_HEAD_SIGNATURE, ) assert result.status == "POSTED" assert len(wrapper_calls) == 1 == len(base_calls) assert wrapper_calls[0]["method"] == "POST" assert base_calls[0]["body"] == {"body": "/gemini review"} assert "Authorization" in base_calls[0]["headers"] def test_11_signature_mismatch_breaks_owner_trigger_call(tmp_path): """signature 가 4-arg 가 아니면 (3-arg / 5-arg) TypeError 등으로 fail — silent fail 방지.""" decision_path = _write_decision(tmp_path, pr=8204, head=_HEAD_SIGNATURE) # 잘못된 signature — 3-arg only (headers 누락). def broken_http_post_3arg(method: str, path: str, body: dict) -> dict: return {"status": 201} runner_bad = OwnerTriggerOnly( workspace_root=tmp_path, http_post=broken_http_post_3arg, # type: ignore[arg-type] token_provider=lambda: _FAKE_OWNER_TOKEN, audit=OwnerTriggerAudit(tmp_path), ) # runtime 에서 4-arg invoke → TypeError (extra positional). with pytest.raises(TypeError): runner_bad.trigger_gemini_review( decision_path=decision_path, owner="test-owner", repo="test-repo", current_head_actual=_HEAD_SIGNATURE, ) # ─── Fix 3: logger.exception with secret masking (회장 §3) ─────────────────── def test_12_redact_diagnostics_masks_secret_keys(): """`_redact_diagnostics` 가 token/authorization/api_key/secret/password 키를 redact.""" sample = { "status": 500, "token": "MY_RAW_TOKEN_xxxxxxxxxxxxxxxxxx", "Authorization": "Bearer " + _SENTINEL_SECRET, "api_key": "SECRET_KEY", "API-KEY": "ALSO_SECRET", "password": "p@ss", "secret_value": "internal", "x-github-request-id": "ABC:DEF:123:456", "nested": { "token": "inner_token_xxx", "status_code": 502, }, "list_field": ["safe", {"token": "list_secret"}], } redacted = _redact_diagnostics(sample) assert redacted["token"] == "" assert redacted["Authorization"] == "" assert redacted["api_key"] == "" assert redacted["API-KEY"] == "" assert redacted["password"] == "" assert redacted["secret_value"] == "" # diagnostic 필드는 보존. assert redacted["status"] == 500 assert redacted["x-github-request-id"] == "ABC:DEF:123:456" # 중첩 dict 도 재귀 redact. assert redacted["nested"]["token"] == "" assert redacted["nested"]["status_code"] == 502 # 리스트 내부도 재귀. assert redacted["list_field"][0] == "safe" assert redacted["list_field"][1]["token"] == "" # raw sentinel 이 어디에도 노출되지 않음. serialised = json.dumps(redacted) assert _SENTINEL_SECRET not in serialised assert "MY_RAW_TOKEN" not in serialised assert "SECRET_KEY" not in serialised def test_13_redact_diagnostics_masks_value_sentinels(): """key 가 일반적이어도 값에 Bearer / ghp_ 등 sentinel 포함 시 redact.""" sample = { "msg": "Bearer " + _SENTINEL_SECRET, # value 자체에 'Bearer ' 포함. "another": _FAKE_OWNER_TOKEN_PREFIX + "rawtokenvalue", # ghp_ prefix "harmless": "regular text", } redacted = _redact_diagnostics(sample) assert redacted["msg"] == "" assert redacted["another"] == "" assert redacted["harmless"] == "regular text" def test_14_logger_exception_called_on_http_post_failure_no_token_leak(tmp_path, caplog): """task-2563 §3 + 완료 조건 5: logger.exception 호출 + token raw value 미노출 어셀션.""" fixture = _load_fixture(FIXTURE_LOGGER_EXCEPTION) scenario = fixture["exception_scenarios"][0] # transient_http_500 sentinel = scenario["secret_must_not_appear_in_log"] assert sentinel == _SENTINEL_SECRET # fixture sanity decision_path = _write_decision(tmp_path, pr=8301, head=_HEAD_FAIL) # http_post raise — response_headers 포함. class HttpErrorWithHeaders(RuntimeError): def __init__(self, msg: str, *, status: int, response_headers: dict) -> None: super().__init__(msg) self.status = status self.response_headers = response_headers http_calls: list = [] def failing_http_post(method, path, body, headers): http_calls.append({"method": method, "path": path}) raise HttpErrorWithHeaders( scenario["exc_message"], status=scenario["status"], response_headers=scenario["response_headers"], ) runner = OwnerTriggerOnly( workspace_root=tmp_path, http_post=failing_http_post, token_provider=lambda: _FAKE_OWNER_TOKEN, audit=OwnerTriggerAudit(tmp_path), ) with caplog.at_level(logging.ERROR, logger="anu_v2.owner_trigger_only"): with pytest.raises(RuntimeError): runner.trigger_gemini_review( decision_path=decision_path, owner="test-owner", repo="test-repo", current_head_actual=_HEAD_FAIL, ) # ── logger.exception 호출 1회 (ERROR 레벨로 캡처). exception_records = [ r for r in caplog.records if r.levelno == logging.ERROR and r.exc_info is not None ] assert len(exception_records) == scenario["expected_logger_calls"] == 1 record = exception_records[0] msg = record.getMessage() # ── token raw value 노출 0 (가장 중요한 완료 조건 5 / 금지 7). assert _SENTINEL_SECRET not in msg assert _FAKE_OWNER_TOKEN not in msg assert _FAKE_OWNER_TOKEN_PREFIX not in msg assert "Bearer " not in msg # Authorization 값 0 # token_hash_prefix 는 등장 OK (raw value 아님, sha256 8자). # diagnostic 필드 (status / request_id) 는 등장. assert "8301" in msg or "/issues/8301/comments" in msg # response_headers 의 diagnostic 값들 → redacted dict 형태로 message 에 포함. assert scenario["response_headers"]["x-github-request-id"] in msg # ── HTTP POST 부작용 0 (mock 은 1 call 만, 그 실패 call). assert len(http_calls) == 1 def test_15_logger_exception_audit_failed_preserved_token_value_not_logged(tmp_path, caplog): """task-2563 §3 + 완료 조건 6: fail-closed 속성 유지 (audit FAILED + POST 0 + side-effect 0).""" fixture = _load_fixture(FIXTURE_LOGGER_EXCEPTION) decision_path = _write_decision(tmp_path, pr=8302, head=_HEAD_FAIL) class HttpErrorMinimal(RuntimeError): pass http_calls: list = [] def failing_http_post(method, path, body, headers): http_calls.append({"method": method, "path": path}) raise HttpErrorMinimal("403 forbidden — insufficient scopes") runner = OwnerTriggerOnly( workspace_root=tmp_path, http_post=failing_http_post, token_provider=lambda: _FAKE_OWNER_TOKEN, audit=OwnerTriggerAudit(tmp_path), ) with caplog.at_level(logging.ERROR, logger="anu_v2.owner_trigger_only"): with pytest.raises(RuntimeError): runner.trigger_gemini_review( decision_path=decision_path, owner="test-owner", repo="test-repo", current_head_actual=_HEAD_FAIL, ) # ── audit JSONL 어셀션: PENDING + FAILED 가 기록되어 fail-closed 유지. audit_path = tmp_path / "memory" / "events" / "owner-trigger-audit.jsonl" assert audit_path.exists() rows = [ json.loads(line) for line in audit_path.read_text(encoding="utf-8").splitlines() if line ] results = [r["result"] for r in rows] assert "PENDING" in results # crash-safety sentinel. assert "FAILED" in results # http_post 실패 후 audit 박제. failed_row = next(r for r in rows if r["result"] == "FAILED") # ── 완료 조건 5: token_value_logged=False 강제. assert failed_row["token_value_logged"] is False # ── token_hash_prefix 8 자 (raw value 아님). assert "token_hash_prefix" in failed_row assert len(failed_row["token_hash_prefix"]) == 8 assert failed_row["error_code"] == "HTTP_POST_FAIL" assert failed_row["action"] == ALLOWED_ACTION # ── token raw value 가 audit jsonl 어느 row 에도 없음. audit_text = audit_path.read_text(encoding="utf-8") assert _SENTINEL_SECRET not in audit_text assert _FAKE_OWNER_TOKEN not in audit_text assert "Bearer " not in audit_text # ── HTTP POST 부작용 0 (mock 1 call only). assert len(http_calls) == 1 # ── logger 도 token 값 0. full_log_text = "\n".join(r.getMessage() for r in caplog.records) assert _SENTINEL_SECRET not in full_log_text assert _FAKE_OWNER_TOKEN not in full_log_text def test_16_collect_http_diagnostics_extracts_github_fields(): """`_collect_http_diagnostics` 가 status / x-github-request-id / x-accepted-permissions / documentation_url 을 안전하게 수집.""" class StubExc(RuntimeError): def __init__(self) -> None: super().__init__("stub") self.status = 403 self.response_headers = { "X-GitHub-Request-Id": "REQ-1234", "X-Accepted-GitHub-Permissions": "issues=write", "documentation_url": "https://docs.github.com/x", } diag = _collect_http_diagnostics(StubExc()) assert diag["status"] == 403 assert diag["x-github-request-id"] == "REQ-1234" assert diag["x-accepted-github-permissions"] == "issues=write" assert diag["documentation_url"] == "https://docs.github.com/x" def test_17_signature_callable_annotation_4_args(): """meta-test: OwnerTriggerOnly.__init__ 의 http_post 파라미터 annotation 이 4-arg signature. runtime 에서 처음 signature drift 가 발견되지 않도록 정적 회귀.""" sig = inspect.signature(OwnerTriggerOnly.__init__) http_post_param = sig.parameters.get("http_post") assert http_post_param is not None # annotation 은 'Callable[[str, str, dict, dict], dict]' string repr 보유. annotation_str = str(http_post_param.annotation) # callable 인자 4 종 확인 (str, str, dict, dict). assert "str" in annotation_str assert "dict" in annotation_str # 정확한 형태: Callable[[str, str, dict, dict], dict] assert annotation_str.count("str") >= 2 assert annotation_str.count("dict") >= 3 # [str, str, dict, dict] + return dict def test_18_pending_window_fixture_loadable_and_invariants(): """fixture 자체 sanity check — schema invariants + constants 일관성.""" fixture = _load_fixture(FIXTURE_PENDING_WINDOW) assert fixture["constants"]["FIRST_TRIGGER_PENDING_WINDOW_SECONDS"] == FIRST_TRIGGER_PENDING_WINDOW_SECONDS assert fixture["constants"]["FIRST_TIMEOUT_SECONDS"] == FIRST_TIMEOUT_SECONDS assert fixture["expected_invariants"]["chat_notifications"] == 0 assert fixture["expected_invariants"]["long_polling"] == 0 assert fixture["expected_invariants"]["pending_state_does_not_invoke_owner_trigger_default"] is True def test_19_signature_fixture_contract_matches_callable_annotation(): """fixture signature contract == 실제 callable annotation 1:1.""" contract = _load_fixture(FIXTURE_SIGNATURE_REPRO)["signature_contract"] assert contract["positional_args_count"] == 4 assert contract["keyword_args_count"] == 0 names = [a["name"] for a in contract["positional_args"]] assert names == ["method", "path", "body", "headers"] assert contract["positional_args"][0]["expected_value"] == "POST" assert contract["positional_args"][2]["expected_value"] == {"body": COMMENT_BODY} def test_20_logger_exception_fixture_redaction_pattern_consistent(): """fixture 의 redaction_contract 가 module 의 실제 패턴과 일관.""" fixture = _load_fixture(FIXTURE_LOGGER_EXCEPTION) contract = fixture["redaction_contract"] # value sentinels 모두 redact 처리되는지 sanity. for sentinel in contract["redact_value_sentinels"]: text_value = sentinel + "RAW_TOKEN_VALUE_xxxxx" out = _redact_diagnostics({"safe_key": text_value}) assert out["safe_key"] == "", f"sentinel {sentinel!r} not redacted" # key pattern 도 sanity. for key in ("token", "Authorization", "api_key", "API-KEY", "secret", "password"): out = _redact_diagnostics({key: "value"}) assert out[key] == "", f"key {key!r} not redacted" # fail_closed_invariants: assert fixture["fail_closed_invariants"]["audit_failed_recorded"] is True assert fixture["fail_closed_invariants"]["audit_token_value_logged"] is False assert fixture["fail_closed_invariants"]["http_post_extra_side_effects"] == 0 assert fixture["fail_closed_invariants"]["logger_called_before_txn_record_failed"] is True