Spaces:
Running
Running
| """dependencies.auth ๋จ์ ํ ์คํธ. | |
| ์คํ: cd src/backend && python -m pytest tests/test_auth_dependency.py -v | |
| """ | |
| from __future__ import annotations | |
| import time | |
| from uuid import UUID | |
| import jwt | |
| import pytest | |
| from fastapi import HTTPException | |
| from starlette.requests import Request | |
| from config import get_settings | |
| from dependencies.auth import ( | |
| ANONYMOUS_USER, | |
| get_current_user, | |
| get_current_user_optional, | |
| get_user_from_ws_token, | |
| is_anonymous, | |
| ) | |
| TEST_SECRET = "test-secret-for-hs256-verification-000000" | |
| TEST_USER_ID = "11111111-2222-3333-4444-555555555555" | |
| def _request() -> Request: | |
| """์ต์ ์ค์ฝํ๋ก ๋ง๋ ๋๋ฏธ Request. | |
| get_current_user ๋ request ๋ฅผ session_tracker.record ๋ก๋ง ๋๊ธฐ๊ณ , | |
| record ๋ ํค๋์์ IP ๋ฅผ ๋ชป ์ฐพ์ผ๋ฉด ์ฆ์ ๋ฐํํ๋ฏ๋ก ๋น ํค๋๋ฉด ๋ถ์์ฉ ์์. | |
| """ | |
| return Request( | |
| { | |
| "type": "http", | |
| "method": "GET", | |
| "path": "/", | |
| "headers": [], | |
| "query_string": b"", | |
| "client": None, | |
| } | |
| ) | |
| def _token(payload: dict | None = None, secret: str = TEST_SECRET) -> str: | |
| base = { | |
| "sub": TEST_USER_ID, | |
| "email": "alice@example.com", | |
| "aud": "authenticated", | |
| "exp": int(time.time()) + 3600, | |
| "iat": int(time.time()), | |
| } | |
| if payload: | |
| base.update(payload) | |
| return jwt.encode(base, secret, algorithm="HS256") | |
| def _enable_auth(monkeypatch): | |
| """ํ ์คํธ ์ค auth_enabled=True, jwt secret ์ฃผ์ ํ Settings ์บ์ ๋ฆฌ์ .""" | |
| get_settings.cache_clear() | |
| monkeypatch.setenv("AUTH_ENABLED", "true") | |
| monkeypatch.setenv("SUPABASE_JWT_SECRET", TEST_SECRET) | |
| yield | |
| get_settings.cache_clear() | |
| # --- auth_enabled=True ------------------------------------------------------ | |
| async def test_happy_path(): | |
| user = await get_current_user(_request(), authorization=f"Bearer {_token()}") | |
| assert user.id == UUID(TEST_USER_ID) | |
| assert user.email == "alice@example.com" | |
| assert user.tier == "free" | |
| async def test_missing_header_raises_401(): | |
| with pytest.raises(HTTPException) as exc: | |
| await get_current_user(_request(), authorization=None) | |
| assert exc.value.status_code == 401 | |
| async def test_malformed_bearer_raises_401(): | |
| with pytest.raises(HTTPException) as exc: | |
| await get_current_user(_request(), authorization="NotBearer xyz") | |
| assert exc.value.status_code == 401 | |
| async def test_invalid_signature_raises_401(): | |
| bad = _token(secret="different-secret") | |
| with pytest.raises(HTTPException) as exc: | |
| await get_current_user(_request(), authorization=f"Bearer {bad}") | |
| assert exc.value.status_code == 401 | |
| async def test_expired_token_raises_401(): | |
| expired = _token({"exp": int(time.time()) - 10}) | |
| with pytest.raises(HTTPException) as exc: | |
| await get_current_user(_request(), authorization=f"Bearer {expired}") | |
| assert exc.value.status_code == 401 | |
| async def test_tier_from_app_metadata(): | |
| t = _token({"app_metadata": {"tier": "premium"}}) | |
| user = await get_current_user(_request(), authorization=f"Bearer {t}") | |
| assert user.tier == "premium" | |
| # --- optional ------------------------------------------------------------- | |
| async def test_optional_returns_anon_when_missing(): | |
| user = await get_current_user_optional(_request(), authorization=None) | |
| assert is_anonymous(user) | |
| async def test_optional_validates_when_present(): | |
| user = await get_current_user_optional(_request(), authorization=f"Bearer {_token()}") | |
| assert not is_anonymous(user) | |
| assert user.id == UUID(TEST_USER_ID) | |
| # --- WS token ------------------------------------------------------------- | |
| async def test_ws_token_happy_path(): | |
| user = await get_user_from_ws_token(token=_token()) | |
| assert user.id == UUID(TEST_USER_ID) | |
| async def test_ws_token_missing_raises_401(): | |
| with pytest.raises(HTTPException) as exc: | |
| await get_user_from_ws_token(token=None) | |
| assert exc.value.status_code == 401 | |
| # --- auth_enabled=False --------------------------------------------------- | |
| async def test_disabled_returns_anon(monkeypatch): | |
| monkeypatch.setenv("AUTH_ENABLED", "false") | |
| get_settings.cache_clear() | |
| user = await get_current_user(_request(), authorization=None) | |
| assert is_anonymous(user) | |
| user = await get_current_user_optional(_request(), authorization=f"Bearer garbage") | |
| assert is_anonymous(user) | |
| user = await get_user_from_ws_token(token=None) | |
| assert is_anonymous(user) | |
| assert ANONYMOUS_USER.tier == "free" | |