"""StepRunner.invalidate_downstream — consumes_downstream 1급 필드 검증. v0.5.23 도입 (v0.5.15의 preserve_on_upstream_force 대체). 2-pass 의존성 선언은 **상류 입장**에서 consumes_downstream: [step_id] 명시. cascade 시 ref가 소비한다고 선언한 하류는 파일 보존(DB step_run=stale). """ from __future__ import annotations import json import logging from pathlib import Path from unittest.mock import MagicMock import pytest from app.core.step_runner import StepRunner @pytest.fixture def fake_runner(tmp_path, monkeypatch): """StepRunner 인스턴스를 mocking — DB/파일 경로 제어.""" inst = StepRunner.__new__(StepRunner) inst.step_id = "upstream" inst.project_id = "p1" inst.episode_id = "e1" inst.db = MagicMock() inst.run_id = "r1" inst.manifest = {} inst.project_config = {} # settings.projects_dir 패치 → tmp_path 하위로 from app.core.config import settings monkeypatch.setattr(settings, "projects_dir", str(tmp_path)) return inst, tmp_path def _make_cp_file(tmp_path: Path, pid: str, eid: str, sid: str) -> Path: cp_dir = tmp_path / pid / "checkpoints" / "episodes" / eid / sid cp_dir.mkdir(parents=True, exist_ok=True) cp_file = cp_dir / "manifest.json" cp_file.write_text(json.dumps({"data": {}}), encoding="utf-8") return cp_file def _patch_consumes_from_manifest(monkeypatch, manifest: dict) -> None: """step_catalog.get_consumes_downstream를 fake_manifest 기반 closure로 치환. step_runner가 step_catalog 헬퍼를 경유하므로(M1 수용), 테스트가 STEP_MANIFEST만 monkeypatch하면 catalog는 빌드 시점 실 데이터를 계속 가리켜 동기화 안 됨. step_runner가 lazy import로 `from app.core.step_catalog import get_consumes_downstream` 를 호출할 때 patched 버전을 받아가도록 본 모듈 네임스페이스에 직접 패치. """ monkeypatch.setattr( "app.core.step_catalog.get_consumes_downstream", lambda sid: manifest.get(sid, {}).get("consumes_downstream", []), ) def test_consumed_downstream_file_preserved(fake_runner, monkeypatch, caplog): """ref(상류)가 consumes_downstream에 선언한 하류는 체크포인트 파일 보존. DB status는 stale로 업데이트 + info 로그 emit.""" inst, tmp_path = fake_runner # upstream(ref)이 "target_consumed"를 역참조 소비한다고 선언. _m = { "upstream": { "depends_on": [], "consumes_downstream": ["target_consumed"], }, "target_consumed": {"depends_on": ["upstream"]}, "target_deleted": {"depends_on": ["upstream"]}, } monkeypatch.setattr("app.core.step_manifest.STEP_MANIFEST", _m) _patch_consumes_from_manifest(monkeypatch, _m) monkeypatch.setattr( "app.core.step_runner.get_all_downstream_recursive", lambda sid: ["target_consumed", "target_deleted"], ) cp_preserved = _make_cp_file(tmp_path, "p1", "e1", "target_consumed") cp_deleted = _make_cp_file(tmp_path, "p1", "e1", "target_deleted") with caplog.at_level(logging.INFO, logger="app.core.step_runner"): inst.invalidate_downstream() # consumes 목록에 있는 step은 파일 그대로 assert cp_preserved.exists(), "consumes_downstream에 선언된 하류 파일이 삭제됨" # 목록에 없는 step은 archive + unlink assert not cp_deleted.exists(), "consumes 외 하류 파일이 보존됨" # DB는 둘 다 stale UPDATE — 2회 assert inst.db.execute.call_count == 2 # info 로그: "Preserved stale checkpoint" 1건 + 소비 주체 명시 preserved_logs = [ r for r in caplog.records if "Preserved stale checkpoint" in r.message and r.levelname == "INFO" ] assert len(preserved_logs) == 1 assert "target_consumed" in preserved_logs[0].message assert "upstream" in preserved_logs[0].message # consumed by ref def test_no_consumes_declaration_deletes_all(fake_runner, monkeypatch): """ref에 consumes_downstream 선언이 없으면 모든 하류 파일 삭제 (기본 동작).""" inst, tmp_path = fake_runner _m = { "upstream": {"depends_on": []}, # consumes_downstream 키 자체 없음 "target": {"depends_on": ["upstream"]}, } monkeypatch.setattr("app.core.step_manifest.STEP_MANIFEST", _m) _patch_consumes_from_manifest(monkeypatch, _m) monkeypatch.setattr( "app.core.step_runner.get_all_downstream_recursive", lambda sid: ["target"], ) cp = _make_cp_file(tmp_path, "p1", "e1", "target") inst.invalidate_downstream() assert not cp.exists() def test_empty_consumes_list_deletes_all(fake_runner, monkeypatch): """consumes_downstream이 빈 리스트여도 모든 하류 파일 삭제.""" inst, tmp_path = fake_runner _m = { "upstream": {"depends_on": [], "consumes_downstream": []}, "target": {"depends_on": ["upstream"]}, } monkeypatch.setattr("app.core.step_manifest.STEP_MANIFEST", _m) _patch_consumes_from_manifest(monkeypatch, _m) monkeypatch.setattr( "app.core.step_runner.get_all_downstream_recursive", lambda sid: ["target"], ) cp = _make_cp_file(tmp_path, "p1", "e1", "target") inst.invalidate_downstream() assert not cp.exists() def test_delete_checkpoints_false_preserves_all(fake_runner, monkeypatch): """delete_checkpoints=False면 consumes 무관하게 전부 파일 보존 (editorial cascade 경로 — consumes 체크 자체를 건너뛰어야 함).""" inst, tmp_path = fake_runner _m = { "upstream": { "depends_on": [], "consumes_downstream": ["target_consumed"], }, "target_consumed": {"depends_on": ["upstream"]}, "target_normal": {"depends_on": ["upstream"]}, } monkeypatch.setattr("app.core.step_manifest.STEP_MANIFEST", _m) _patch_consumes_from_manifest(monkeypatch, _m) monkeypatch.setattr( "app.core.step_runner.get_all_downstream_recursive", lambda sid: ["target_consumed", "target_normal"], ) cp1 = _make_cp_file(tmp_path, "p1", "e1", "target_consumed") cp2 = _make_cp_file(tmp_path, "p1", "e1", "target_normal") inst.invalidate_downstream(delete_checkpoints=False) # 둘 다 파일 유지 assert cp1.exists() assert cp2.exists() # DB는 둘 다 stale UPDATE assert inst.db.execute.call_count == 2 def test_no_downstream_no_commit(fake_runner, monkeypatch): """downstream이 비어 있으면 db.commit 호출 없음.""" inst, _ = fake_runner _m = {"upstream": {"depends_on": []}} monkeypatch.setattr("app.core.step_manifest.STEP_MANIFEST", _m) _patch_consumes_from_manifest(monkeypatch, _m) monkeypatch.setattr( "app.core.step_runner.get_all_downstream_recursive", lambda sid: [], ) inst.invalidate_downstream() inst.db.commit.assert_not_called() def test_target_step_id_override_reads_from_ref(fake_runner, monkeypatch, caplog): """target_step_id 전달 시 cascade 기준을 그 step으로 전환하고, consumes_downstream도 해당 step의 선언을 읽어야 한다 (editorial cascade 호환).""" inst, tmp_path = fake_runner _m = { "upstream": {"depends_on": []}, # 무관 "other_step": { "consumes_downstream": ["other_consumed"], }, "other_consumed": {"depends_on": ["other_step"]}, "other_normal": {"depends_on": ["other_step"]}, } monkeypatch.setattr("app.core.step_manifest.STEP_MANIFEST", _m) _patch_consumes_from_manifest(monkeypatch, _m) recorded_refs = [] def _fake_downstream(sid): recorded_refs.append(sid) return ["other_consumed", "other_normal"] monkeypatch.setattr( "app.core.step_runner.get_all_downstream_recursive", _fake_downstream, ) cp_consumed = _make_cp_file(tmp_path, "p1", "e1", "other_consumed") cp_normal = _make_cp_file(tmp_path, "p1", "e1", "other_normal") with caplog.at_level(logging.INFO, logger="app.core.step_runner"): inst.invalidate_downstream(target_step_id="other_step") # get_all_downstream_recursive가 other_step으로 호출됨 assert recorded_refs == ["other_step"] # other_step의 consumes_downstream 기준으로 보존/삭제 분기 assert cp_consumed.exists(), "other_step이 consumes한 하류 파일 삭제됨" assert not cp_normal.exists(), "other_step이 consumes하지 않은 하류 파일 보존됨" # 로그에 ref가 other_step으로 반영 preserved_logs = [ r for r in caplog.records if "Preserved stale checkpoint" in r.message and r.levelname == "INFO" ] assert len(preserved_logs) == 1 assert "other_consumed" in preserved_logs[0].message assert "other_step" in preserved_logs[0].message