"""Regression tests for callback_lifecycle_classifier. 규칙: - 순수 함수 호출만 (실 cron 0, 실 발사 0, subprocess 0, live workspace 의존 0) - frozen fixture 만 입력 (tests/fixtures/callback_lifecycle/) - pytest 로 실행 가능 dev6 담당: 벨레스 (fixture 구성 + regression 테스트) classifier core: 스바로그 (utils/callback_lifecycle_classifier.py) """ import json import sys from pathlib import Path # worktree root 를 sys.path 에 추가 (안전망) _WORKTREE_ROOT = Path(__file__).resolve().parents[2] if str(_WORKTREE_ROOT) not in sys.path: sys.path.insert(0, str(_WORKTREE_ROOT)) import pytest from utils.callback_lifecycle_classifier import classify_callback_lifecycle # frozen fixture root FIXTURE_ROOT = Path(__file__).resolve().parents[1] / "fixtures" / "callback_lifecycle" # ───────────────────────────────────────────────────────────────────────────── # 헬퍼 # ───────────────────────────────────────────────────────────────────────────── def _load_fixture(fixture_dir_name: str): """evidence.json 과 expected.json 을 로드해 반환.""" base = FIXTURE_ROOT / fixture_dir_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 # ───────────────────────────────────────────────────────────────────────────── # 테스트 1: 3 fixture 회장 매핑 정확 단일값 assert (parametrize) # ───────────────────────────────────────────────────────────────────────────── @pytest.mark.parametrize("fixture_dir", [ "task-2625", "task-2628", "task-2628_plus_1", ]) def test_chairman_mapping_exact(fixture_dir): """각 fixture 의 evidence → classify → expected 정확 일치 검증. - delivery_outcome, normal_callback_miss_cause, evidence_completeness, classification: == 정확 일치 - root_cause_tags: 순서 무관 집합 동치 (set 동치 + len 동치 → 중복 없음 보장) """ evidence, expected = _load_fixture(fixture_dir) result = classify_callback_lifecycle(evidence) # 단일값 정확 일치 assert result["delivery_outcome"] == expected["delivery_outcome"], ( f"[{fixture_dir}] delivery_outcome mismatch: " f"got={result['delivery_outcome']!r}, want={expected['delivery_outcome']!r}" ) assert result["normal_callback_miss_cause"] == expected["normal_callback_miss_cause"], ( f"[{fixture_dir}] normal_callback_miss_cause mismatch: " f"got={result['normal_callback_miss_cause']!r}, want={expected['normal_callback_miss_cause']!r}" ) assert result["evidence_completeness"] == expected["evidence_completeness"], ( f"[{fixture_dir}] evidence_completeness mismatch: " f"got={result['evidence_completeness']!r}, want={expected['evidence_completeness']!r}" ) assert result["classification"] == expected["classification"], ( f"[{fixture_dir}] classification mismatch: " f"got={result['classification']!r}, want={expected['classification']!r}" ) # root_cause_tags: 순서 무관 집합 동치 (brittleness 회피) result_tags = result["root_cause_tags"] expected_tags = expected["root_cause_tags"] assert set(result_tags) == set(expected_tags), ( f"[{fixture_dir}] root_cause_tags set mismatch: " f"got={sorted(result_tags)}, want={sorted(expected_tags)}" ) assert len(result_tags) == len(expected_tags), ( f"[{fixture_dir}] root_cause_tags length mismatch (중복 의심): " f"got={result_tags}, want={expected_tags}" ) # ───────────────────────────────────────────────────────────────────────────── # 테스트 2: UNKNOWN/INSUFFICIENT_EVIDENCE fixture (unknown_insufficient) # ───────────────────────────────────────────────────────────────────────────── def test_unknown_insufficient_evidence(): """core evidence 부재 시 추정 없이 UNKNOWN + MISSING 반환 검증.""" evidence, _ = _load_fixture("unknown_insufficient") result = classify_callback_lifecycle(evidence) assert result["delivery_outcome"] == "UNKNOWN_INSUFFICIENT_EVIDENCE", ( f"delivery_outcome: got={result['delivery_outcome']!r}" ) assert result["normal_callback_miss_cause"] == "UNKNOWN_INSUFFICIENT_EVIDENCE", ( f"normal_callback_miss_cause: got={result['normal_callback_miss_cause']!r}" ) assert result["evidence_completeness"] == "MISSING", ( f"evidence_completeness: got={result['evidence_completeness']!r}" ) assert result["root_cause_tags"] == [], ( f"root_cause_tags should be empty, got={result['root_cause_tags']!r}" ) # missing_evidence_sources 에 3대 core source 모두 포함 missing_srcs = result.get("missing_evidence_sources", []) assert "result_json" in missing_srcs, ( f"'result_json' not in missing_evidence_sources: {missing_srcs}" ) assert "collectors" in missing_srcs, ( f"'collectors' not in missing_evidence_sources: {missing_srcs}" ) assert "schedule_history" in missing_srcs, ( f"'schedule_history' not in missing_evidence_sources: {missing_srcs}" ) # ───────────────────────────────────────────────────────────────────────────── # 테스트 3: CALLBACK_DELIVERY_GAP residual 규칙 (inline evidence, 2종) # ───────────────────────────────────────────────────────────────────────────── def test_callback_delivery_gap_positive(): """ANU key cron FIRED + collector 부재 + BEFORE_FIRE 원인 전무 → CALLBACK_DELIVERY_GAP.""" evidence = { "task_id": "inline-delivery-gap-positive", "result_json": { "callback_registration_status": "FIRED", "callback_cron_id": "AAAA1111", "owner_key": "c119085addb0f8b7", "envelope_utf8_bytes": 300 }, "schedule_history": { "normal_callback": { "cron_id": "AAAA1111", "owner_key": "c119085addb0f8b7", "fired": True, "argv_present": True }, "fallback": {"registered": False, "cron_id": None, "owner_key": None, "fired": False} }, "escalate": {"reason": None, "source": None}, "collectors": { "normal_collection_ack": False, "self_collection": {"present": False, "owner_key": None, "authoritative": False}, "fallback_collector": {"present": False, "normal_callback_received": False, "no_op": False, "authoritative": False}, "manual_anu_reverify": {"present": False} }, "applied_count": 0, "git": {"foreign_dirty": [], "git_gate_blocked": False}, "reflection_merged": True, "bot_app_token_present": True, "self_key_fail_closed": False, "envelope_only": True } result = classify_callback_lifecycle(evidence) assert result["normal_callback_miss_cause"] == "CALLBACK_DELIVERY_GAP", ( f"expected CALLBACK_DELIVERY_GAP, got={result['normal_callback_miss_cause']!r}" ) def test_callback_delivery_gap_negative_self_key_fail_closed(): """BEFORE_FIRE 원인(self-key fail-closed) 존재 → DELIVERY_GAP 아님, SELF_KEY_FAIL_CLOSED_BEFORE_FIRE.""" evidence = { "task_id": "inline-delivery-gap-negative", "result_json": { "callback_registration_status": "ENVELOPE_PREPARED_NOT_FIRED", "callback_cron_id": None, "owner_key": "c119085addb0f8b7", "envelope_utf8_bytes": 300 }, "schedule_history": { "normal_callback": { "cron_id": None, "owner_key": "1e41a2324a3ccdd0", "fired": False, "argv_present": False }, "fallback": {"registered": False, "cron_id": None, "owner_key": None, "fired": False} }, "escalate": {"reason": None, "source": None}, "collectors": { "normal_collection_ack": False, "self_collection": {"present": False, "owner_key": None, "authoritative": False}, "fallback_collector": {"present": False, "normal_callback_received": False, "no_op": False, "authoritative": False}, "manual_anu_reverify": {"present": False} }, "applied_count": 0, "git": {"foreign_dirty": [], "git_gate_blocked": False}, "reflection_merged": True, "bot_app_token_present": True, "self_key_fail_closed": True, "envelope_only": True } result = classify_callback_lifecycle(evidence) assert result["normal_callback_miss_cause"] != "CALLBACK_DELIVERY_GAP", ( f"should not be CALLBACK_DELIVERY_GAP when BEFORE_FIRE cause exists, " f"got={result['normal_callback_miss_cause']!r}" ) assert result["normal_callback_miss_cause"] == "SELF_KEY_FAIL_CLOSED_BEFORE_FIRE", ( f"expected SELF_KEY_FAIL_CLOSED_BEFORE_FIRE, got={result['normal_callback_miss_cause']!r}" ) # ───────────────────────────────────────────────────────────────────────────── # 테스트 4: 결정성(determinism) — 동일 evidence 2회 classify → 동일 결과 # ───────────────────────────────────────────────────────────────────────────── def test_determinism_task_2625(): """task-2625 frozen fixture 로 2회 분류 → 동일 결과 (live 비의존 결정적 증명).""" evidence, _ = _load_fixture("task-2625") result_a = classify_callback_lifecycle(evidence) result_b = classify_callback_lifecycle(evidence) assert result_a["delivery_outcome"] == result_b["delivery_outcome"] assert result_a["normal_callback_miss_cause"] == result_b["normal_callback_miss_cause"] assert result_a["evidence_completeness"] == result_b["evidence_completeness"] assert result_a["classification"] == result_b["classification"] assert result_a["root_cause_tags"] == result_b["root_cause_tags"], ( "root_cause_tags 순서까지 동일해야 결정적 (deterministic order)" ) # ───────────────────────────────────────────────────────────────────────────── # 테스트 5: anu_v3 import 0 정적 검증 # ───────────────────────────────────────────────────────────────────────────── def test_no_anu_v3_import_in_classifier_sources(): """classifier 및 states 소스 파일에 'import anu_v3' 구문이 없음을 정적 검증. 주의: docstring/주석에서 'anu_v3' 문자열을 언급하는 것은 허용. 금지되는 것은 실제 import 구문 (import anu_v3 / from anu_v3 ...). """ import re utils_dir = _WORKTREE_ROOT / "utils" target_files = [ utils_dir / "callback_lifecycle_classifier.py", utils_dir / "callback_lifecycle_states.py", ] # 실제 import 구문 패턴: 'import anu_v3' 또는 'from anu_v3' _IMPORT_ANU_V3_PATTERN = re.compile(r"^\s*(import\s+anu_v3|from\s+anu_v3\b)", re.MULTILINE) for fpath in target_files: if not fpath.exists(): # 파일이 아직 없으면 스킵 (스바로그 미완성 상태) pytest.skip(f"{fpath.name} 아직 미생성 — classifier 완성 후 재실행 필요") source_text = fpath.read_text(encoding="utf-8") match = _IMPORT_ANU_V3_PATTERN.search(source_text) assert match is None, ( f"{fpath.name} 에 실제 anu_v3 import 구문 발견 (line contains: {match.group().strip()!r}) — " f"금지된 의존성. classifier 는 evidence-only decoupled 이어야 함." )