Files
hyungi_document_server/tests/test_laptop_worker_bot_auth.py
T
Hyungi Ahn f60d6e52fc feat(worker-pool): Registry-1B Pull 활성화 (auth + worker_jobs + 5 endpoint)
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>
2026-05-19 08:54:07 +09:00

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