Hyungi Ahn
0cbd97fcba
각 helper 가 자체 engine + NullPool 사용 (connection 격리). fixture chain 의 asyncpg "another operation in progress" race 회피. 호출 site 단순화. 같은 파일 sequential 실행 시 module-level app + global engine pool 충돌은 별 follow-up `PR-Worker-Pool-Test-Fixture-Isolation` (P3) 영역. 단독 PASS 검증: auth 5/5 + smoke 3/3 + ownership 1/1. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
"""PR-Worker-Pool-Registry-1B — worker_jobs ORM/schema 단위 검증.
|
|
|
|
3 항목 (endpoint test 와 중복 회피):
|
|
1. CHECK constraint — status enum 4 외 값 INSERT → IntegrityError (정정 #5)
|
|
2. partial unique-index 존재 검증 (idx_worker_jobs_pending_type)
|
|
3. ON DELETE SET NULL — worker_capabilities 삭제 시 worker_jobs.worker_id 자동 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 sqlalchemy import text
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from _worker_pool_helpers import (
|
|
cleanup_worker_capabilities,
|
|
cleanup_worker_jobs,
|
|
ensure_user,
|
|
fetch_worker_job,
|
|
get_database_url,
|
|
insert_worker_capability,
|
|
insert_worker_job,
|
|
)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def owner_id():
|
|
return await ensure_user("test-owner-smoke-1b")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_constraint_rejects_unknown_status(owner_id):
|
|
"""정정 #5: status 가 enum 4 외 값이면 IntegrityError."""
|
|
engine = create_async_engine(get_database_url())
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
try:
|
|
with pytest.raises(IntegrityError):
|
|
async with sm() as session:
|
|
await session.execute(
|
|
text(
|
|
"INSERT INTO worker_jobs (user_id, job_type, status) "
|
|
"VALUES (:u, 'test-smoke-1b-bad', 'running')"
|
|
),
|
|
{"u": owner_id},
|
|
)
|
|
await session.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_partial_pending_index_exists(owner_id):
|
|
"""idx_worker_jobs_pending_type partial index 존재 검증."""
|
|
engine = create_async_engine(get_database_url())
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
try:
|
|
async with sm() as session:
|
|
res = await session.execute(
|
|
text(
|
|
"SELECT indexname FROM pg_indexes "
|
|
"WHERE tablename = 'worker_jobs' "
|
|
"AND indexname = 'idx_worker_jobs_pending_type'"
|
|
)
|
|
)
|
|
assert res.scalar_one_or_none() == "idx_worker_jobs_pending_type"
|
|
|
|
# 정의에 WHERE status = 'pending' 포함 확인
|
|
res2 = await session.execute(
|
|
text(
|
|
"SELECT indexdef FROM pg_indexes "
|
|
"WHERE indexname = 'idx_worker_jobs_pending_type'"
|
|
)
|
|
)
|
|
indexdef = res2.scalar_one()
|
|
assert "WHERE" in indexdef
|
|
assert "pending" in indexdef
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_delete_set_null_when_capability_dropped(owner_id):
|
|
"""worker_capabilities 삭제 시 worker_jobs.worker_id 자동 NULL."""
|
|
wid = f"test-smoke-1b-del-{uuid.uuid4().hex[:8]}"
|
|
await insert_worker_capability(wid, owner_id)
|
|
job_id = await insert_worker_job(
|
|
owner_id,
|
|
"test-smoke-1b-del",
|
|
status="completed",
|
|
worker_id=wid,
|
|
attempts=1,
|
|
)
|
|
try:
|
|
# cleanup_worker_capabilities 도 worker_heartbeats CASCADE 처리
|
|
await cleanup_worker_capabilities(wid)
|
|
job = await fetch_worker_job(job_id)
|
|
assert job is not None
|
|
assert job["worker_id"] is None # ON DELETE SET NULL
|
|
finally:
|
|
await cleanup_worker_jobs("test-smoke-1b")
|