"""★참조 획득 스텝 — **누구를 대상으로 사는가**. 두 자리를 **모두** 만족해야 한다: ① `grounding_plan` 이 `generation_difficulty` 로 연 것 (hard/uncertain) ② `entity_filter` 를 **살아남은** 행 ①만 보면 필터가 지운 것까지 사고, ②만 보면 흔한 것까지 산다. ★★`route` 로 고르면 안 된다 — 실측(§2-3c)에서 아홉 축이 **전부** `research` 였다. route 를 대리로 쓰면 **전부** 사게 된다. """ import pytest from app.core.steps.reference_acquisition_step import ReferenceAcquisitionStep def _step(plan_decided, filtered): s = ReferenceAcquisitionStep.__new__(ReferenceAcquisitionStep) cps = {"grounding_plan": {"data": {"decided": plan_decided}}, "entity_filter": {"data": {"filtered_entities": filtered}}} s._load_prev_checkpoint = lambda sid: cps.get(sid) s.project_config = {"project_id": "p"} return s def _d(sid, form, difficulty="hard", route="research"): return {"research_subject_id": sid, "_surface_form": form, "route": route, "generation_difficulty": difficulty} class TestWhoGetsBought: def test_a_hard_target_that_survived_is_bought(self): st = _step([_d("rs_a", "1983년 오백원 지폐")], {"props": [{"short_id": "P02", "name": "1983년 오백원 지폐", "research_subject_id": "rs_a"}]}) got = st._targets() assert [t["research_subject_id"] for t in got] == ["rs_a"] assert got[0]["entity_key"] == "props" def test_uncertain_counts_too(self): st = _step([_d("rs_a", "옛 승차권", "uncertain")], {"props": [{"short_id": "P01", "name": "옛 승차권", "research_subject_id": "rs_a"}]}) assert len(st._targets()) == 1 def test_a_not_hard_target_is_not_bought(self): """★흔한 것까지 사면 시간이 통째로 샌다.""" st = _step([_d("rs_a", "자전거 체인", "not_hard")], {"props": [{"short_id": "P01", "name": "자전거 체인", "research_subject_id": "rs_a"}]}) assert st._targets() == [] def test_research_route_alone_does_not_buy(self): """★★실측에서 아홉 축이 전부 research 였다 — 대리로 쓰면 전부 산다.""" st = _step([_d("rs_a", "자전거 체인", "not_hard", route="research")], {"props": [{"short_id": "P01", "name": "자전거 체인", "research_subject_id": "rs_a"}]}) assert st._targets() == [] def test_a_hard_target_the_filter_removed_is_not_bought(self): """★필터가 지운 것을 사면 **최종 집합이 아닌 것**을 사는 것이다.""" st = _step([_d("rs_a", "옛 지폐")], {"props": []}) assert st._targets() == [] def test_nothing_to_buy_is_a_normal_outcome(self): st = _step([], {"props": [{"short_id": "P01", "name": "가방"}]}) out = st._execute() assert out["failed_count"] == 0 assert out["data"]["target_count"] == 0 assert out["data"]["acquisition_contract"] class TestItBindsBySubjectIdNotByName: def test_a_materialized_row_carries_its_subject_id(self): st = _step([_d("rs_a", "옛 지폐")], {"props": [{"short_id": "P02", "name": "옛 지폐", "research_subject_id": "rs_a", "grounding_materialized": True}]}) assert len(st._targets()) == 1 def test_an_extracted_row_is_matched_by_surface_form(self): """★추출이 건진 행에는 subject id 가 없다 — 이름으로 한 번 더 본다.""" st = _step([_d("rs_a", "회수권 뭉치")], {"props": [{"short_id": "P01", "name": "고무줄로 묶인 낡은 종이 회수권 뭉치"}]}) assert [t["research_subject_id"] for t in st._targets()] == ["rs_a"] def test_an_unrelated_row_is_not_matched(self): st = _step([_d("rs_a", "회수권 뭉치")], {"props": [{"short_id": "P01", "name": "요금통"}]}) assert st._targets() == [] class TestTheStepIsRegisteredWhereItBelongs: def test_it_runs_after_outlook_and_before_the_scene(self): """★★**뒤집은 시험** (2026-09-01 D 활성화). 앞에는 「필터 뒤, 상세 앞」이었다. 아웃룩 실물이 `outlook_phase3` (19.2)에서 생기므로 그 **뒤**로 옮겼다 — 앞에 두면 그 갈래의 의무를 결속할 수 없다(중앙 입력이 그것을 요구하며 선다). `scene_detail`(21.70) **앞**이어야 참조가 씬에 실린다. """ from app.core.step_manifest import STEP_MANIFEST as M m = M["reference_acquisition"] assert M["entity_filter"]["order"] < m["order"] assert M["outlook_phase3"]["order"] < m["order"] assert m["order"] < M["scene_detail"]["order"] def test_it_is_invalidated_when_the_outlook_binding_changes(self): from app.core.step_manifest import STEP_MANIFEST as M deps = M["reference_acquisition"]["depends_on"] assert "outlook_phase3" in deps and "grounding_screen" in deps def test_it_is_invalidated_when_the_final_set_changes(self): from app.core.step_manifest import STEP_MANIFEST assert "entity_filter" in STEP_MANIFEST["reference_acquisition"]["depends_on"] def test_it_does_not_run_in_legacy(self): """★검색·다운로드가 있는 스텝이다 — legacy·shadow 에서는 안 돈다. ★★**뒤집었다** — 앞에는 `if_grounding_v2` 였는데, 그러면 새 판 (`v2_chunk`)에서 **안 돈다**. 이 스텝은 두 갈래 모두에서 돌아야 하므로 「v2 인가」가 아니라 **「참조를 사는 판인가」**를 묻는다. """ from app.core.applicability import APPLICABILITY_VALIDATORS from app.core.grounding_mode import (GROUNDING_MODE_LEGACY, GROUNDING_MODE_SHADOW_PLAN, GROUNDING_MODE_V2, GROUNDING_MODE_V2_CHUNK, buys_v2_research, uses_chunk_producer) from app.core.step_manifest import STEP_MANIFEST name = STEP_MANIFEST["reference_acquisition"]["applicability"] assert name == "if_grounding_reference" assert name in APPLICABILITY_VALIDATORS def _on(mode): return buys_v2_research(mode) or uses_chunk_producer(mode) assert _on(GROUNDING_MODE_V2) and _on(GROUNDING_MODE_V2_CHUNK) assert not _on(GROUNDING_MODE_LEGACY) assert not _on(GROUNDING_MODE_SHADOW_PLAN) def test_a_missing_reference_blocks_downstream(self): """★어렵다고 살려 낸 것을 참조 없이 순수 T2I 로 내리지 않는다.""" from app.core.step_manifest import STEP_MANIFEST assert STEP_MANIFEST["reference_acquisition"][ "allow_partial_downstream"] is False class TestBuyingIsNotWiredYet: """★유료 배선은 **승인과 acceptance 뒤**다. 지금 돌면 서야 한다.""" def test_it_refuses_to_buy_without_the_orchestration(self): st = _step([_d("rs_a", "옛 지폐")], {"props": [{"short_id": "P02", "name": "옛 지폐", "research_subject_id": "rs_a"}]}) with pytest.raises(NotImplementedError) as exc: st._execute() assert "유료 승인" in str(exc.value)