o2o-negosium-original/lps/tests/test_worker.py

"""워커 루프 테스트 — drain→DONE, 실패→재시도→dead, reaper 회수 후 재처리, NOTIFY 깨움.
핸들러는 fake(브라우저 없이) — 워커 로직만 결정론적으로 검증한다."""

import asyncio

import pytest_asyncio
from sqlalchemy import text

from common.enums import JobStatus, JobType
from crud.job_crud import JobQueue
from worker.notify import JobListener
from worker.runner import Worker


@pytest_asyncio.fixture
async def q(db_engine):
    async with db_engine.begin() as conn:
        await conn.execute(text("TRUNCATE job"))
    return JobQueue()


async def test_worker_drains_all_to_done(q):
    for i in range(3):
        await q.enqueue(JobType.SEARCH.value, {"product_name": f"item{i}"})

    async def handler(job):
        return {"ok": True, "q": job["payload"]["product_name"]}

    processed = await Worker("w1", q, handler).drain()
    assert processed == 3
    assert (await q.counts())["DONE"] == 3


async def test_worker_failure_retries_then_dead(q):
    await q.enqueue(JobType.SEARCH.value, {"product_name": "x"}, max_attempts=2)

    async def boom(job):
        raise RuntimeError("nope")

    w = Worker("w1", q, boom, backoff_fn=lambda a: 0)  # 백오프 0 → 즉시 재시도 가능
    assert await w.process_one() is True                # 1/2 실패 → PENDING
    assert (await q.counts())["PENDING"] == 1
    assert await w.process_one() is True                # 2/2 실패 → DEAD
    counts = await q.counts()
    assert counts["DEAD"] == 1 and counts["PENDING"] == 0
    assert await w.process_one() is False               # DEAD 는 claim 대상 아님


async def test_reaper_reclaims_then_worker_reprocesses(q):
    jid = await q.enqueue(JobType.SEARCH.value, {"product_name": "x"})
    await q.claim("dead-worker", lease_sec=1)   # 점유 후 사망 흉내
    await asyncio.sleep(1.3)
    assert jid in await q.reap()                 # 회수 → PENDING

    async def handler(job):
        return {"ok": True}

    assert await Worker("w2", q, handler).process_one() is True
    assert (await q.counts())["DONE"] == 1


async def test_job_deadline_cancels_hung_handler(q):
    """핸들러 행 → 데드라인 초과 시 취소·fail 처리(재큐/DEAD)돼야 한다. 없으면 heartbeat 가
    lease 를 계속 갱신해 워커 슬롯이 영구 점유된다(2026-07-10 부하테스트 실측)."""
    jid = await q.enqueue(JobType.SEARCH.value, {"product_name": "hang"}, max_attempts=1)

    async def hang(job):
        await asyncio.sleep(3600)

    w = Worker("w1", q, hang, backoff_fn=lambda a: 0, job_deadline_sec=0.2)
    assert await w.process_one() is True            # 행이어도 데드라인에 끊겨 반환된다
    assert (await q.counts())["DEAD"] == 1          # max_attempts=1 → 즉시 DEAD
    job = await q.get(jid)
    assert "JobDeadlineExceeded" in job["last_error"]


async def test_job_deadline_retries_before_dead(q):
    """데드라인 초과도 일반 실패처럼 백오프 재큐를 탄다(시도 소진 전까지)."""
    await q.enqueue(JobType.SEARCH.value, {"product_name": "hang"}, max_attempts=2)

    async def hang(job):
        await asyncio.sleep(3600)

    w = Worker("w1", q, hang, backoff_fn=lambda a: 0, job_deadline_sec=0.2)
    assert await w.process_one() is True
    assert (await q.counts())["PENDING"] == 1       # 1/2 → 재큐
    assert await w.process_one() is True
    assert (await q.counts())["DEAD"] == 1          # 2/2 → DEAD


async def test_ops_counts_long_running_as_stuck(q, db_engine):
    """lease 가 계속 갱신돼도(행 상태의 heartbeat) 실행 10분 초과면 stuck_running 에 잡혀야 한다."""
    await q.enqueue(JobType.SEARCH.value, {"product_name": "x"})
    await q.claim("w1", lease_sec=3600)             # lease 는 멀쩡(만료 안 됨)
    assert (await q.ops())["stuck_running"] == 0
    async with db_engine.begin() as conn:
        await conn.execute(text("UPDATE job SET run_started_at = now() - interval '11 minutes' WHERE status = 2"))
    assert (await q.ops())["stuck_running"] == 1


async def test_browser_reaper_survives_hung_adapter():
    """한 어댑터의 close 행이 정리 루프 전체를 멈추면 안 된다 — 타임아웃 후 다음 어댑터로."""
    from worker_main import run_browser_reaper

    class HungAdapter:
        source = "hung"
        async def close_if_idle(self, idle_sec):
            await asyncio.sleep(3600)

    class OkAdapter:
        source = "ok"
        closed = False
        async def close_if_idle(self, idle_sec):
            self.closed = True

    ok = OkAdapter()
    stop = asyncio.Event()
    task = asyncio.create_task(run_browser_reaper(
        [HungAdapter(), ok], stop, idle_sec=0, interval=0.01, close_timeout=0.05))
    try:
        await asyncio.wait_for(_until(lambda: ok.closed), timeout=3.0)   # 행 어댑터를 지나 ok 까지 도달
    finally:
        stop.set()
        await task


async def _until(cond, poll: float = 0.02):
    while not cond():
        await asyncio.sleep(poll)


async def test_enqueue_notifies_listener(q):
    listener = JobListener()
    await listener.start()
    try:
        await q.enqueue(JobType.SEARCH.value, {"product_name": "x"})  # pg_notify 발생
        assert await listener.wait(3.0) is True                       # 즉시 깨어남
    finally:
        await listener.close()