hyungi_document_server/tests/test_worker_jobs_retry.py

"""PR-Worker-Pool-Registry-1B 정정 #3 — explicit /result failed 재시도.

3 case:
  A. initial attempts=0, max=3 → claim 후 attempts=1 → failed → status='pending' 복귀.
  B. initial attempts=2, max=3 → claim 후 attempts=3 → failed → final 'failed'.
  C. failed 요청에 result={'partial':...} 포함 → DB result IS NULL 유지.
"""

from __future__ import annotations

import os
import sys
import uuid

import pytest
import pytest_asyncio

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))

from httpx import ASGITransport, AsyncClient

from _worker_pool_helpers import (
    cleanup_worker_capabilities,
    cleanup_worker_jobs,
    ensure_user,
    fetch_worker_job,
    insert_worker_capability,
    insert_worker_job,
    mint_access_token,
)


@pytest_asyncio.fixture
async def env_setup(monkeypatch):
    monkeypatch.setenv("LAPTOP_WORKER_BOT_USERNAME", "laptop-worker-bot")


@pytest_asyncio.fixture
async def worker_token(env_setup):
    await ensure_user("laptop-worker-bot")
    return mint_access_token("laptop-worker-bot")


@pytest_asyncio.fixture
async def owner_id():
    return await ensure_user("test-owner-retry-1b")


@pytest_asyncio.fixture
async def worker_id_unique(owner_id):
    wid = f"test-retry-1b-{uuid.uuid4().hex[:8]}"
    await insert_worker_capability(wid, owner_id)
    yield wid
    await cleanup_worker_jobs("test-retry-1b")
    await cleanup_worker_capabilities(wid)


async def _seed_processing_job(
    owner_id: int, worker_id: str, attempts: int, max_attempts: int = 3
) -> int:
    return await insert_worker_job(
        owner_id,
        f"test-retry-1b-{uuid.uuid4().hex[:8]}",
        status="processing",
        worker_id=worker_id,
        attempts=attempts,
        max_attempts=max_attempts,
        claimed=True,
    )


@pytest.mark.asyncio
async def test_case_a_failed_then_pending(worker_token, worker_id_unique, owner_id):
    from main import app

    job_id = await _seed_processing_job(owner_id, worker_id_unique, attempts=1)
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as c:
        r = await c.post(
            "/internal/worker/result",
            json={
                "job_id": job_id,
                "worker_id": worker_id_unique,
                "status": "failed",
                "error_message": "case A error",
            },
            headers={"Authorization": f"Bearer {worker_token}"},
        )
    assert r.status_code == 200, r.text
    job = await fetch_worker_job(job_id)
    assert job["status"] == "pending"
    assert job["worker_id"] is None
    assert job["claimed_at"] is None
    assert job["completed_at"] is None
    assert job["error_message"] == "case A error"


@pytest.mark.asyncio
async def test_case_b_failed_max_attempts_final(worker_token, worker_id_unique, owner_id):
    from main import app

    job_id = await _seed_processing_job(owner_id, worker_id_unique, attempts=3)
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as c:
        r = await c.post(
            "/internal/worker/result",
            json={
                "job_id": job_id,
                "worker_id": worker_id_unique,
                "status": "failed",
                "error_message": "case B final",
            },
            headers={"Authorization": f"Bearer {worker_token}"},
        )
    assert r.status_code == 200
    job = await fetch_worker_job(job_id)
    assert job["status"] == "failed"
    assert job["completed_at"] is not None


@pytest.mark.asyncio
async def test_case_c_failed_ignores_result_field(worker_token, worker_id_unique, owner_id):
    from main import app

    job_id = await _seed_processing_job(owner_id, worker_id_unique, attempts=1)
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as c:
        r = await c.post(
            "/internal/worker/result",
            json={
                "job_id": job_id,
                "worker_id": worker_id_unique,
                "status": "failed",
                "result": {"partial": "should be ignored"},
                "error_message": "case C",
            },
            headers={"Authorization": f"Bearer {worker_token}"},
        )
    assert r.status_code == 200
    job = await fetch_worker_job(job_id)
    assert job["result"] is None  # 정정 #3 엄격 규칙