"""여러 편을 **순서대로 하나씩** 돌린다 (2026-09-04). ## 왜 순차인가 앞 화가 끝나야 `entity_canon` 이 서고, 그래야 다음 화가 앞 화 명부를 보고 같은 신원으로 이어 붙인다(`episode_carry`). 나란히 돌리면 둘 다 백지에서 시작해 같은 것을 두 신원으로 만든다. 게다가 `short_id` 장부와 canon 이 프로젝트 범위라 동시 쓰기가 부딪힌다. ## 여기서 재는 것 1. 락이 **프로젝트 하나**인가 — 종전에는 에피소드마다라 두 화가 나란히 돌았다 2. 대기열이 **화수 순서**로 도는가 3. `queued` 가 DB 에 **durable** 하게 남는가 (프로세스를 다시 띄워도 보인다) 4. 앞 화가 실패하면 뒤 화가 **선다** (`blocked_by_previous`) 5. 도는 중인 화가 있으면 대기열 등록을 **거절**하는가 Lane: ``-m pg``. """ from __future__ import annotations import uuid import pytest from sqlalchemy import text as sql_text pytestmark = pytest.mark.pg def _seed(session, pid: str, n: int): uid = f"queue-{uuid.uuid4()}" session.execute(sql_text( "INSERT INTO user_account (id, username, display_name, password_hash, " "role, is_active, created_at, updated_at) VALUES " "(:uid, :un, 't', 'x', 'creator', 1, '2026-01-01', '2026-01-01')" ), {"uid": uid, "un": f"u_{uid}"}) session.execute(sql_text( "INSERT INTO project_registry (id, name, created_by, created_at, updated_at) " "VALUES (:pid, 'queue', :uid, '2026-01-01', '2026-01-01')" ), {"pid": pid, "uid": uid}) eids = [] # ★일부러 **거꾸로** 넣는다 — 정렬이 화수 기준인지 보려고. for num in range(n, 0, -1): eid = str(uuid.uuid4()) eids.append((eid, num)) session.execute(sql_text( "INSERT INTO episode (id, project_id, title, episode_number, " "source_filename, source_path, fulltext, status, created_at, updated_at) " "VALUES (:eid, :pid, :t, :n, 'x.txt', 'x/x.txt', '본문', 'uploaded', " "'2026-01-01', '2026-01-01')" ), {"eid": eid, "pid": pid, "t": f"{num}화", "n": num}) session.commit() return [e for e, _ in eids] @pytest.fixture def proj(pg_session): pid = f"p-{uuid.uuid4()}" eids = _seed(pg_session, pid, 3) yield pid, eids from app.core.task_registry import release_episode_run, unregister_task release_episode_run(f"run_all:{pid}") unregister_task(f"run_queue:{pid}") def test_락이_프로젝트_하나다(proj): """★종전에는 에피소드마다라 같은 프로젝트 두 화가 나란히 돌았다.""" from app.core.task_registry import claim_episode_run, release_episode_run pid, _eids = proj key = f"run_all:{pid}" assert claim_episode_run(key, holder="1화:analysis") is True assert claim_episode_run(key, holder="2화:analysis") is False, ( "같은 프로젝트의 두 화가 자리를 같이 잡았다") release_episode_run(key) assert claim_episode_run(key, holder="2화:analysis") is True release_episode_run(key) def test_대기열이_화수_순서로_돈다(pg_session, proj, monkeypatch): """★★끝점 — 실제로 **어떤 순서로** 불렸는지 본다.""" import app.services.analysis_dispatch_service as ads pid, eids = proj called: list = [] def _fake(*, project_id, episode_id, category, mode, db, run_inline=False, **kw): assert run_inline is True, "대기열이 배경 작업으로 던졌다 — 순차가 안 된다" called.append(episode_id) return {"ok": True} monkeypatch.setattr(ads, "dispatch_category_run", _fake) # ★배경 스레드 없이 이 스레드에서 끝까지 — 순서를 확실히 보려고. monkeypatch.setattr(ads, "submit_background_job", lambda *, job_key, target, description="": (target(), True)[1]) out = ads.dispatch_project_queue(pid, eids, "analysis", "resume", pg_session) order = [e["episode_number"] for e in out["episodes"]] assert order == [1, 2, 3], f"화수 순서가 아니다: {order}" nums = {e: n for e, n in zip(eids, (3, 2, 1))} assert [nums[e] for e in called] == [1, 2, 3], "부른 순서가 화수 순이 아니다" def test_queued_가_DB_에_남는다(pg_session, proj, monkeypatch): """메모리 대기열은 재시작에 사라지지만 「무엇을 대기시켰는지」는 남아야 한다.""" import app.services.analysis_dispatch_service as ads pid, eids = proj # 배경 작업을 **안 돌린다** — 등록만 하고 상태를 본다. monkeypatch.setattr(ads, "submit_background_job", lambda *, job_key, target, description="": True) ads.dispatch_project_queue(pid, eids, "analysis", "resume", pg_session) rows = pg_session.execute(sql_text( "SELECT status FROM episode WHERE project_id = :p"), {"p": pid}).fetchall() assert {r[0] for r in rows} == {ads.EPISODE_STATUS_QUEUED}, rows def test_앞_화가_실패하면_뒤_화가_선다(pg_session, proj, monkeypatch): """★자동으로 넘어가면 앞 화 없이 만든 신원이 그대로 굳는다.""" import app.services.analysis_dispatch_service as ads pid, eids = proj nums = {e: n for e, n in zip(eids, (3, 2, 1))} seen: list = [] def _fake(*, project_id, episode_id, category, mode, db, run_inline=False, **kw): seen.append(nums[episode_id]) if nums[episode_id] == 2: raise RuntimeError("2화 실패") return {"ok": True} monkeypatch.setattr(ads, "dispatch_category_run", _fake) monkeypatch.setattr(ads, "submit_background_job", lambda *, job_key, target, description="": (target(), True)[1]) ads.dispatch_project_queue(pid, eids, "analysis", "resume", pg_session) assert seen == [1, 2], f"2화가 실패했는데 3화를 돌렸다: {seen}" # ★다른 세션이 쓴 것은 **다른 세션으로** 읽는다. 이 fixture 세션은 # `_seed` 부터 이어 온 제 트랜잭션 안에 있어 남의 commit 을 안 본다. from app.core.database import SessionLocal probe = SessionLocal() try: st3 = probe.execute(sql_text( "SELECT status FROM episode WHERE project_id = :p " "AND episode_number = 3"), {"p": pid}).fetchone()[0] finally: probe.close() assert st3 == ads.EPISODE_STATUS_BLOCKED, ( f"3화가 「앞 화 때문에」로 안 섰다: {st3}") def test_도는_중이면_대기열_등록을_거절한다(pg_session, proj): from app.core.errors import AppError from app.services.analysis_dispatch_service import dispatch_project_queue pid, eids = proj pg_session.execute(sql_text( "UPDATE episode SET status = 'analyzing' WHERE id = :e"), {"e": eids[0]}) pg_session.commit() with pytest.raises(AppError) as got: dispatch_project_queue(pid, eids, "analysis", "resume", pg_session) assert got.value.code == "step.already_running" def test_이_프로젝트에_없는_에피소드는_거절한다(pg_session, proj): from app.core.errors import AppError from app.services.analysis_dispatch_service import dispatch_project_queue pid, eids = proj with pytest.raises(AppError) as got: dispatch_project_queue(pid, [*eids, str(uuid.uuid4())], "analysis", "resume", pg_session) assert got.value.code == "episode.not_found" def test_대기열이_배경작업_주입을_가리지_않는다(): """★`submit_background_job` 을 함수 안에서 다시 import 하면 그 이름이 지역으로 가려져, 시험이 모듈 전역을 바꿔도 **진짜 스레드**가 돈다. 실측 2026-09-04 — 그래서 시험이 대기열이 끝나기 전에 DB 를 읽고 「안 섰다」로 보였다. 지역 import 결함 부류다. """ import ast import inspect import app.services.analysis_dispatch_service as ads tree = ast.parse(inspect.getsource(ads.dispatch_project_queue)) local = { alias.asname or alias.name for node in ast.walk(tree) if isinstance(node, (ast.Import, ast.ImportFrom)) for alias in node.names } assert "submit_background_job" not in local, ( "dispatch_project_queue 가 submit_background_job 을 지역으로 다시 " "가져온다 — 모듈 전역을 가려 주입이 안 먹는다") def test_화수가_겹치면_시작_전에_선다(pg_session, proj): """★순서를 정할 수 없는 입력으로 돌기 시작하면 되돌릴 수 없다.""" from app.core.errors import AppError from app.services.analysis_dispatch_service import dispatch_project_queue pid, eids = proj pg_session.execute(sql_text( "UPDATE episode SET episode_number = 1 WHERE id = :e"), {"e": eids[0]}) pg_session.commit() with pytest.raises(AppError) as got: dispatch_project_queue(pid, eids, "analysis", "resume", pg_session) assert got.value.code == "queue.duplicate_episode_number" def test_대기열이_프로젝트_자리를_끝까지_들고_있다(pg_session, proj, monkeypatch): """★화 사이에 자리를 놓으면 그 틈에 단일 실행이 끼어들어 순서를 깬다.""" import app.services.analysis_dispatch_service as ads from app.core.task_registry import claim_episode_run pid, eids = proj held: list = [] def _fake(*, project_id, episode_id, category, mode, db, run_inline=False, claim_held_by=None, **kw): # 도는 도중 남이 자리를 잡을 수 있나 — 잡히면 문이 없는 것이다. got = claim_episode_run(f"run_all:{pid}", holder="침입자") held.append(got) assert claim_held_by, "대기열이 자리를 물려주지 않았다" return {"ok": True} monkeypatch.setattr(ads, "dispatch_category_run", _fake) monkeypatch.setattr(ads, "submit_background_job", lambda *, job_key, target, description="": (target(), True)[1]) ads.dispatch_project_queue(pid, eids, "analysis", "resume", pg_session) assert held == [False, False, False], ( f"화가 도는 동안 남이 자리를 잡았다: {held}") def test_등록_실패하면_queued_를_되돌린다(pg_session, proj, monkeypatch): """★안 되돌리면 아무도 안 돌릴 「대기 중」으로 남는다 (Codex 2026-09-04).""" import app.services.analysis_dispatch_service as ads from app.core.errors import AppError pid, eids = proj monkeypatch.setattr(ads, "submit_background_job", lambda *, job_key, target, description="": False) with pytest.raises(AppError) as got: ads.dispatch_project_queue(pid, eids, "analysis", "resume", pg_session) assert got.value.code == "step.already_running" from app.core.database import SessionLocal probe = SessionLocal() try: rows = probe.execute(sql_text( "SELECT status FROM episode WHERE project_id = :p"), {"p": pid}).fetchall() finally: probe.close() assert {r[0] for r in rows} == {"uploaded"}, ( f"등록이 실패했는데 queued 로 남았다: {rows}")