"""POST /api/v1/projects/{pid}/episodes/{eid}/repair-projection 테스트 — W4 P3-3. Scenario coverage: - 사전 'failed' step_run이 있을 때 sync 성공 → 'synced' 전환 - 'failed' step 없음 + sync 성공 → 200 + repaired_steps=[] - episode 없음 → 404 - sync 내부 실패 → 500 + 기존 failed 상태 보존 """ from __future__ import annotations import io import shutil import uuid from pathlib import Path import pytest from fastapi.testclient import TestClient from fpdf import FPDF from sqlalchemy import text as sql_text from app.core.config import settings from app.core.database import Base, engine from app.main import app from tests._safety_guards import safe_drop_all, safe_rmtree def _make_pdf(text="repair test") -> bytes: pdf = FPDF() pdf.add_page() pdf.set_font("Helvetica", size=12) pdf.cell(200, 10, text=text) return pdf.output() @pytest.fixture(autouse=True) def _setup_db(): Base.metadata.create_all(engine) with TestClient(app): pass yield safe_drop_all(engine, Base.metadata) proj_dir = Path(settings.projects_dir) if proj_dir.exists(): safe_rmtree(proj_dir) @pytest.fixture() def client(): with TestClient(app, raise_server_exceptions=False) as c: yield c def _login_admin(client: TestClient): resp = client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin123"}) assert resp.status_code == 200 def _create_project_episode(client: TestClient) -> tuple[str, str]: _login_admin(client) p = client.post("/api/v1/projects/", json={"name": "Repair Test"}) assert p.status_code == 200 pid = p.json()["id"] e = client.post( f"/api/v1/projects/{pid}/episodes/", data={"episode_number": "1", "title": "E1"}, files={"file": ("e1.pdf", io.BytesIO(_make_pdf()), "application/pdf")}, ) assert e.status_code == 200 return pid, e.json()["id"] def _seed_step_run(pid: str, eid: str, step_id: str, sync_status: str | None, sync_error: str | None = None): """step_run row 삽입 (pytest fixture DB 세션 독립).""" from app.core.database import SessionLocal db = SessionLocal() try: db.execute(sql_text( "INSERT INTO step_run " "(id, project_id, episode_id, step_id, status, sync_status, sync_error, " "created_at, updated_at) " "VALUES (:id, :pid, :eid, :sid, 'completed', :ss, :se, '2026-04-22', '2026-04-22')" ), { "id": str(uuid.uuid4()), "pid": pid, "eid": eid, "sid": step_id, "ss": sync_status, "se": sync_error, }) db.commit() finally: db.close() def _query_step(pid: str, eid: str, step_id: str) -> tuple: from app.core.database import SessionLocal db = SessionLocal() try: return db.execute(sql_text( "SELECT sync_status, sync_error, synced_at FROM step_run " "WHERE project_id = :pid AND episode_id = :eid AND step_id = :sid" ), {"pid": pid, "eid": eid, "sid": step_id}).fetchone() finally: db.close() # ── Happy path: failed → synced ── def test_repair_projection_transitions_failed_to_synced(client, monkeypatch): pid, eid = _create_project_episode(client) _seed_step_run(pid, eid, "text_cleanup", "failed", "DB down") _seed_step_run(pid, eid, "scene_director", "synced", None) # 이미 synced는 건드리지 않음 # orchestrate_full_sync 성공 모킹 def fake_sync(pid_, eid_, db, *, step_id=None, repair_step_ids=None): # orchestrate_full_sync는 repair_step_ids가 주어지면 같은 트랜잭션에서 # failed→synced 전환을 수행. 테스트에서는 해당 전환을 시뮬레이션. if repair_step_ids: from sqlalchemy import text as _t from datetime import datetime, timezone now = datetime.now(timezone.utc).isoformat() for sid, prev_err in repair_step_ids.items(): db.execute(_t( "UPDATE step_run SET sync_status='synced', sync_error=NULL, synced_at=:now " "WHERE project_id=:pid AND episode_id=:eid AND step_id=:sid " "AND sync_status='failed' " "AND (sync_error = :prev OR (sync_error IS NULL AND :prev IS NULL))" ), {"pid": pid_, "eid": eid_, "sid": sid, "now": now, "prev": prev_err}) db.commit() return {"entity": {"ok": 1}, "relation": {}, "scene_still": {"stills": 3}, "outlook": {}, "episode": {}} monkeypatch.setattr("app.api.v1.episodes.orchestrate_full_sync", fake_sync, raising=False) # episodes.py는 함수 내부에서 import하므로 source를 동시 patch monkeypatch.setattr( "app.services.checkpoint_sync.orchestrate_full_sync", fake_sync, raising=True, ) resp = client.post(f"/api/v1/projects/{pid}/episodes/{eid}/repair-projection") assert resp.status_code == 200, resp.text body = resp.json() assert body["ok"] is True assert body["repaired_steps"] == ["text_cleanup"] assert body["sync_result"]["scene_still"]["stills"] == 3 # DB 검증 — failed였던 step은 synced + sync_error=NULL + synced_at set row = _query_step(pid, eid, "text_cleanup") assert row[0] == "synced" assert row[1] is None assert row[2] is not None # 이미 synced였던 step은 synced_at이 NULL에서 안 바뀜 (repaired 대상 아님) unchanged = _query_step(pid, eid, "scene_director") assert unchanged[0] == "synced" # ── failed 없음 ── def test_repair_projection_no_failed_steps(client, monkeypatch): pid, eid = _create_project_episode(client) _seed_step_run(pid, eid, "text_cleanup", "synced", None) def fake_sync(pid_, eid_, db, *, step_id=None, repair_step_ids=None): return {"entity": {}, "relation": {}, "scene_still": {}, "outlook": {}, "episode": {}} monkeypatch.setattr( "app.services.checkpoint_sync.orchestrate_full_sync", fake_sync, raising=True, ) resp = client.post(f"/api/v1/projects/{pid}/episodes/{eid}/repair-projection") assert resp.status_code == 200 body = resp.json() assert body["ok"] is True assert body["repaired_steps"] == [] # ── episode 없음 ── def test_repair_projection_episode_not_found(client): _login_admin(client) p = client.post("/api/v1/projects/", json={"name": "Repair 404"}) pid = p.json()["id"] resp = client.post(f"/api/v1/projects/{pid}/episodes/does-not-exist/repair-projection") assert resp.status_code == 404 assert resp.json()["error"]["code"] == "episode.not_found" # ── sync 실패 → 500 ── def test_repair_projection_sync_failure_returns_500(client, monkeypatch): pid, eid = _create_project_episode(client) _seed_step_run(pid, eid, "text_cleanup", "failed", "original error") def fake_sync(pid_, eid_, db, *, step_id=None, repair_step_ids=None): raise RuntimeError("still broken") monkeypatch.setattr( "app.services.checkpoint_sync.orchestrate_full_sync", fake_sync, raising=True, ) resp = client.post(f"/api/v1/projects/{pid}/episodes/{eid}/repair-projection") assert resp.status_code == 500 assert resp.json()["error"]["code"] == "repair.sync_failed" # 이전 failed 상태 유지 (orchestrator가 step_id=None이어서 sync_error 덮지 않음) row = _query_step(pid, eid, "text_cleanup") assert row[0] == "failed" assert row[1] == "original error" # ── 권한 없는 사용자 ── def test_repair_projection_requires_auth(client): """로그인 세션 없이 호출 시 401.""" resp = client.post("/api/v1/projects/any-pid/episodes/any-eid/repair-projection") assert resp.status_code in (401, 403)