"""PR Open Watcher Auto-Wrapper dry-run regression — task-2643 Track D. 6 fixture parametrized + live call 단언 + owner/collector_role 위장 차단 검증. """ import json import sys from pathlib import Path import pytest _WT = Path(__file__).resolve().parents[2] if str(_WT) not in sys.path: sys.path.insert(0, str(_WT)) from utils.pr_open_watcher_wrapper import ( CollectorRoleViolationError, LiveCallViolationError, OwnerImpersonationError, PrOpenWatcherWrapperError, WrapperResult, open_pr_and_dispatch_watcher, ) FIXTURE_ROOT = _WT / "tests" / "fixtures" / "pr_open_watcher_wrapper" SUCCESS_FIXTURES = [ "pr_open_dev_bot_watcher_dispatch_success", "pr_open_cron_watcher_dispatch_success", ] EXCEPTION_FIXTURES = [ ("pr_open_anu_owner_in_contract_fail", OwnerImpersonationError), ("pr_open_collector_role_anu_watcher_fail", CollectorRoleViolationError), ("pr_open_dry_run_only_no_live_call", LiveCallViolationError), ("pr_145_bzaona6au_violation_replay", OwnerImpersonationError), ] def _load(name: str): base = FIXTURE_ROOT / name evidence = json.loads((base / "evidence.json").read_text(encoding="utf-8")) expected = json.loads((base / "expected.json").read_text(encoding="utf-8")) return evidence, expected def _make_gh_pr_create_spy(returns: dict, called: list): def _spy(**kwargs): called.append(kwargs) return returns return _spy def _make_watcher_dispatch_spy(returns: dict, called: list): def _spy(**kwargs): called.append(kwargs) return returns return _spy @pytest.mark.parametrize("fixture_name", SUCCESS_FIXTURES) def test_success_fixtures(fixture_name): evidence, expected = _load(fixture_name) gh_calls: list = [] disp_calls: list = [] gh_spy = _make_gh_pr_create_spy(evidence["gh_pr_create_mock_return"], gh_calls) disp_spy = _make_watcher_dispatch_spy(evidence["watcher_dispatch_mock_return"], disp_calls) inp = evidence["input"] result: WrapperResult = open_pr_and_dispatch_watcher( task_id=inp["task_id"], branch=inp["branch"], title=inp["title"], body=inp["body"], owner=inp["owner"], watcher_owner_kind=inp.get("watcher_owner_kind", "dev_bot"), collector_role=inp.get("collector_role", "ANU"), ttl_seconds=inp.get("ttl_seconds", 7200), gh_pr_create=gh_spy, watcher_dispatch=disp_spy, dry_run=True, ) assert result.pr_number == expected["pr_number"] assert result.head_sha == expected["head_sha"] assert result.watcher_schedule_id == expected["watcher_schedule_id"] assert result.watcher_owner == expected["watcher_owner"] assert result.dry_run is True assert result.live_calls_emitted == 0 c = result.contract ca = expected["contract_assertions"] assert c["schema"] == ca["schema"] assert c["owner"] == ca["owner"] assert c["collector_role"] == ca["collector_role"] assert c["watcher_owner_kind"] == ca["watcher_owner_kind"] if "callback_only_reporting" in ca: assert c["callback_only_reporting"] is ca["callback_only_reporting"] if "callback_target.owner_key" in ca: assert c["callback_target"]["owner_key"] == ca["callback_target.owner_key"] if "duplicate_policy" in ca: assert c["duplicate_policy"] == ca["duplicate_policy"] if "terminal_states_subset" in ca: assert set(ca["terminal_states_subset"]).issubset(set(c["terminal_states"])) if "ttl_seconds" in ca: assert c["ttl_seconds"] == ca["ttl_seconds"] assert len(gh_calls) == 1 assert len(disp_calls) == 1 @pytest.mark.parametrize("fixture_name,exc_cls", EXCEPTION_FIXTURES) def test_exception_fixtures(fixture_name, exc_cls): evidence, expected = _load(fixture_name) assert expected["result_type"] == "exception" assert expected["exception_class"] == exc_cls.__name__ gh_calls: list = [] disp_calls: list = [] gh_spy = _make_gh_pr_create_spy(evidence["gh_pr_create_mock_return"], gh_calls) disp_spy = _make_watcher_dispatch_spy( evidence["watcher_dispatch_mock_return"], disp_calls ) inp = evidence["input"] dry_run = inp.get("dry_run", True) with pytest.raises(exc_cls) as exc_info: open_pr_and_dispatch_watcher( task_id=inp["task_id"], branch=inp["branch"], title=inp["title"], body=inp["body"], owner=inp["owner"], watcher_owner_kind=inp.get("watcher_owner_kind", "dev_bot"), collector_role=inp.get("collector_role", "ANU"), gh_pr_create=gh_spy, watcher_dispatch=disp_spy, dry_run=dry_run, ) assert expected["exception_substring"] in str(exc_info.value) if "gh_pr_create" in expected["wrapper_must_not_call"]: assert len(gh_calls) == 0, f"gh_pr_create was called but shouldn't be: {gh_calls}" if "watcher_dispatch" in expected["wrapper_must_not_call"]: assert len(disp_calls) == 0, ( f"watcher_dispatch was called but shouldn't be: {disp_calls}" ) def test_wrapper_rejects_live_call_violation(): """dry_run=False 는 task-2643 범위 위반.""" with pytest.raises(LiveCallViolationError): open_pr_and_dispatch_watcher( task_id="t", branch="b", title="T", body="B", owner="dev6-perun", gh_pr_create=lambda **kw: {"pr_number": 1, "head_sha": "0" * 40}, watcher_dispatch=lambda **kw: {"schedule_id": "X"}, dry_run=False, ) def test_wrapper_rejects_anu_owner_self_identifiers(): for ident in ["anu", "ANU", "anu_main", "ANU_MAIN", "anu_session", "ANU_SESSION"]: with pytest.raises(OwnerImpersonationError): open_pr_and_dispatch_watcher( task_id="t", branch="b", title="T", body="B", owner=ident, gh_pr_create=lambda **kw: {"pr_number": 1, "head_sha": "0" * 40}, watcher_dispatch=lambda **kw: {"schedule_id": "X"}, ) def test_wrapper_rejects_collector_role_impersonation(): for role in ["anu_watcher", "anu_polling", "ANU_WATCHER", "dev6_collector"]: with pytest.raises(CollectorRoleViolationError): open_pr_and_dispatch_watcher( task_id="t", branch="b", title="T", body="B", owner="dev6-perun", collector_role=role, gh_pr_create=lambda **kw: {"pr_number": 1, "head_sha": "0" * 40}, watcher_dispatch=lambda **kw: {"schedule_id": "X"}, ) def test_head_sha_length_validation(): with pytest.raises(PrOpenWatcherWrapperError): open_pr_and_dispatch_watcher( task_id="t", branch="b", title="T", body="B", owner="dev6-perun", gh_pr_create=lambda **kw: {"pr_number": 1, "head_sha": "short"}, watcher_dispatch=lambda **kw: {"schedule_id": "X"}, )