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>
68 lines
1.7 KiB
Python
68 lines
1.7 KiB
Python
"""PR-Worker-Pool-Registry-1B — /internal/worker/* 권한 분리.
|
|
|
|
worker user 외 모든 사용자 = 403.
|
|
1. voice-memo-bot JWT → 403
|
|
2. 일반 user JWT → 403
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
|
|
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 ensure_user, 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 voice_memo_token(env_setup):
|
|
await ensure_user("voice-memo-bot")
|
|
return mint_access_token("voice-memo-bot")
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def regular_user_token(env_setup):
|
|
await ensure_user("test-regular-user-1b")
|
|
return mint_access_token("test-regular-user-1b")
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client():
|
|
from main import app
|
|
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as ac:
|
|
yield ac
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_voice_memo_bot_jwt_rejected(client, voice_memo_token):
|
|
r = await client.post(
|
|
"/internal/worker/register",
|
|
json={"worker_id": "x", "device_label": "x", "worker_class": "x", "tier": "x"},
|
|
headers={"Authorization": f"Bearer {voice_memo_token}"},
|
|
)
|
|
assert r.status_code == 403, r.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_regular_user_jwt_rejected(client, regular_user_token):
|
|
r = await client.post(
|
|
"/internal/worker/heartbeat",
|
|
json={"worker_id": "x", "status": "available"},
|
|
headers={"Authorization": f"Bearer {regular_user_token}"},
|
|
)
|
|
assert r.status_code == 403, r.text
|