f60d6e52fc
worker-pool-policy §B 1B 영역 완료. 1A scaffold (mig 270~274 + 503 stub) 위에:
- mig 275/276: worker_jobs (status CHECK + user_id=owner) + pending partial index
- create_laptop_worker_bot_token + require_worker_user dependency (voice-memo 동형)
- /internal/worker/{register,heartbeat,claim,result,drain} 5 endpoint 실 구현
- /claim FOR UPDATE SKIP LOCKED + 204 body 0
- /result 소유권 검증 (worker_id 매칭, 404) + failed 재시도 (attempts/max)
- explicit failure 시 request.result 무시 (DB result NULL 유지)
- 테스트 22 항목 7 파일
policy §B.2 5 invariant 보존: voice-memo wrapper 변경 0, drain advisory,
result raw JSONB, ProcessingQueue 무변경, 운영 자동 분기 변경 0.
활용처 (recap context + /jobs/recap + payload 100KB guard) = Registry-1C 영역.
stale recovery / 노트북 client / canonical promote = Notebook-Pilot-1 영역.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
146 lines
5.5 KiB
Python
146 lines
5.5 KiB
Python
"""PR-Worker-Pool-Registry-1B invariant 1 강제 — voice-memo-bot wrapper 회귀 0.
|
|
|
|
테스트 5개:
|
|
1. voice-memo-bot legacy 발급 그대로 (env=true, username=voice-memo-bot → token)
|
|
2. voice-memo 분기 우선 평가 (둘 다 env true 라도 voice-memo 가 먼저 hit)
|
|
3. laptop-worker-bot env disabled → None
|
|
4. laptop-worker-bot env enabled + username match → token (TTL 365d)
|
|
5. password_changed_at 갱신 시 401 (legacy NULL 호환 포함, decode_token + verify_password_changed_at)
|
|
|
|
DB 의존 0 (token 함수 + decode_token + monkeypatch only).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
@pytest.fixture
|
|
def env_clean(monkeypatch):
|
|
for k in (
|
|
"VOICE_MEMO_BOT_TOKEN_ENABLED",
|
|
"VOICE_MEMO_BOT_USERNAME",
|
|
"VOICE_MEMO_BOT_TOKEN_EXPIRE_DAYS",
|
|
"LAPTOP_WORKER_BOT_TOKEN_ENABLED",
|
|
"LAPTOP_WORKER_BOT_USERNAME",
|
|
"LAPTOP_WORKER_BOT_TOKEN_EXPIRE_DAYS",
|
|
):
|
|
monkeypatch.delenv(k, raising=False)
|
|
return monkeypatch
|
|
|
|
|
|
def test_voice_memo_bot_legacy_unchanged(env_clean):
|
|
"""invariant 1 ①: voice-memo-bot env=true 시 365d token 발급 그대로."""
|
|
from core.auth import create_voice_memo_bot_token, decode_token
|
|
|
|
env_clean.setenv("VOICE_MEMO_BOT_TOKEN_ENABLED", "true")
|
|
token = create_voice_memo_bot_token("voice-memo-bot")
|
|
assert token is not None
|
|
payload = decode_token(token)
|
|
assert payload is not None
|
|
assert payload["sub"] == "voice-memo-bot"
|
|
assert payload["type"] == "access"
|
|
|
|
|
|
def test_voice_memo_branch_preferred_over_laptop(env_clean):
|
|
"""invariant 1 ②: 두 env 동시 true 일 때 voice-memo 분기가 먼저 평가.
|
|
|
|
실 호출 site (app/api/auth.py login) 의 분기 순서 검증:
|
|
bot_token = create_voice_memo_bot_token(...) # 먼저
|
|
if bot_token is not None: return ...
|
|
laptop_bot_token = create_laptop_worker_bot_token(...) # 그 다음
|
|
voice-memo-bot username 으로 호출하면 voice-memo 분기에서 return,
|
|
laptop_bot_token 은 평가되지 않는다. 본 테스트는 username 별 결과로 검증.
|
|
"""
|
|
from core.auth import create_laptop_worker_bot_token, create_voice_memo_bot_token
|
|
|
|
env_clean.setenv("VOICE_MEMO_BOT_TOKEN_ENABLED", "true")
|
|
env_clean.setenv("LAPTOP_WORKER_BOT_TOKEN_ENABLED", "true")
|
|
|
|
# voice-memo-bot username 에서는 voice-memo 가 hit, laptop 은 hit X
|
|
assert create_voice_memo_bot_token("voice-memo-bot") is not None
|
|
assert create_laptop_worker_bot_token("voice-memo-bot") is None
|
|
|
|
# laptop-worker-bot username 에서는 반대
|
|
assert create_voice_memo_bot_token("laptop-worker-bot") is None
|
|
assert create_laptop_worker_bot_token("laptop-worker-bot") is not None
|
|
|
|
|
|
def test_laptop_worker_bot_env_disabled_returns_none(env_clean):
|
|
"""정정 #1 ③: env 미설정/false 일 때 None."""
|
|
from core.auth import create_laptop_worker_bot_token
|
|
|
|
# env 없음
|
|
assert create_laptop_worker_bot_token("laptop-worker-bot") is None
|
|
# env false
|
|
env_clean.setenv("LAPTOP_WORKER_BOT_TOKEN_ENABLED", "false")
|
|
assert create_laptop_worker_bot_token("laptop-worker-bot") is None
|
|
# env true + username mismatch
|
|
env_clean.setenv("LAPTOP_WORKER_BOT_TOKEN_ENABLED", "true")
|
|
assert create_laptop_worker_bot_token("not-laptop-worker-bot") is None
|
|
|
|
|
|
def test_laptop_worker_bot_env_enabled_returns_token(env_clean):
|
|
"""정정 #1 ④: env true + username match → 365d token."""
|
|
from core.auth import create_laptop_worker_bot_token, decode_token
|
|
|
|
env_clean.setenv("LAPTOP_WORKER_BOT_TOKEN_ENABLED", "true")
|
|
env_clean.setenv("LAPTOP_WORKER_BOT_TOKEN_EXPIRE_DAYS", "365")
|
|
token = create_laptop_worker_bot_token("laptop-worker-bot")
|
|
assert token is not None
|
|
payload = decode_token(token)
|
|
assert payload is not None
|
|
assert payload["sub"] == "laptop-worker-bot"
|
|
assert payload["type"] == "access"
|
|
now_ts = int(datetime.now(timezone.utc).timestamp())
|
|
exp = int(payload["exp"])
|
|
diff_days = (exp - now_ts) / 86400
|
|
# 365d ± 1d 허용 (테스트 실행 시간 marginal)
|
|
assert 364 <= diff_days <= 366, f"expected ~365d TTL, got {diff_days:.2f}d"
|
|
|
|
|
|
def test_laptop_worker_bot_password_changed_at_invalidates(env_clean):
|
|
"""invariant 1 ⑤: password_changed_at 갱신 시 401 (legacy NULL 호환 포함)."""
|
|
from fastapi import HTTPException
|
|
|
|
from core.auth import (
|
|
create_laptop_worker_bot_token,
|
|
decode_token,
|
|
verify_password_changed_at,
|
|
)
|
|
|
|
env_clean.setenv("LAPTOP_WORKER_BOT_TOKEN_ENABLED", "true")
|
|
token = create_laptop_worker_bot_token("laptop-worker-bot")
|
|
assert token is not None
|
|
payload = decode_token(token)
|
|
|
|
class FakeUser:
|
|
password_changed_at: datetime | None
|
|
|
|
def __init__(self, dt):
|
|
self.password_changed_at = dt
|
|
|
|
# ⑤-a: NULL → skip (legacy 호환)
|
|
verify_password_changed_at(payload, FakeUser(None)) # raise 없음
|
|
|
|
# ⑤-b: password 변경이 token iat 보다 이전 → 통과
|
|
past = datetime.now(timezone.utc) - timedelta(hours=1)
|
|
verify_password_changed_at(payload, FakeUser(past)) # raise 없음
|
|
|
|
# ⑤-c: password 변경이 token iat 보다 이후 → 401
|
|
future = datetime.now(timezone.utc) + timedelta(hours=1)
|
|
with pytest.raises(HTTPException) as exc:
|
|
verify_password_changed_at(payload, FakeUser(future))
|
|
assert exc.value.status_code == 401
|