portfolio-hub/tests/test_auth.py

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone

from fastapi.testclient import TestClient

from app.core.auth import AuthContext, Permission
from app.core.config import Settings
from app.core.config import get_settings
from app.domain.models import AuthToken
from app.main import app
from app.services.auth_service import AuthService


def _client_with_tokens() -> TestClient:
    get_settings.cache_clear()
    app.dependency_overrides.clear()
    return TestClient(app)


def test_read_route_requires_token(monkeypatch) -> None:
    monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
    monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token")
    get_settings.cache_clear()
    client = _client_with_tokens()

    response = client.get("/portfolios")

    assert response.status_code == 401
    assert response.json()["detail"] == "Missing bearer token"


def test_read_route_accepts_read_token(monkeypatch) -> None:
    monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
    monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token")
    get_settings.cache_clear()
    client = _client_with_tokens()

    response = client.get("/portfolios", headers={"Authorization": "Bearer read-token"})

    assert response.status_code == 501
    assert response.json()["detail"] == "Portfolio persistence is not implemented yet"


def test_trade_route_rejects_read_token(monkeypatch) -> None:
    monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
    monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token")
    monkeypatch.setenv("PORTFOLIOHUB_TRADE_WRITE_TOKENS", "trade-token")
    get_settings.cache_clear()
    client = _client_with_tokens()

    response = client.post(
        "/portfolios/p1/buy",
        headers={"Authorization": "Bearer read-token"},
        json={"quantity": 1, "price": 1000, "fee": 0, "tax": 0, "note": ""},
    )

    assert response.status_code == 403
    assert response.json()["detail"] == "Insufficient permission"


def test_admin_route_accepts_admin_token(monkeypatch) -> None:
    monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "false")
    monkeypatch.setenv("PORTFOLIOHUB_ADMIN_TOKENS", "admin-token")
    get_settings.cache_clear()
    client = _client_with_tokens()

    response = client.delete("/portfolios/p1", headers={"Authorization": "Bearer admin-token"})

    assert response.status_code == 501
    assert response.json()["detail"] == "Portfolio persistence is not implemented yet"


@dataclass
class _AuthRepo:
    def __post_init__(self) -> None:
        self.users: dict[str, object] = {}
        self.tokens: dict[str, AuthToken] = {}
        self._seq = 1

    def get_user_by_username(self, username: str):
        return self.users.get(username)

    def get_service_token_by_name(self, name: str) -> AuthToken | None:
        for token in self.tokens.values():
            if token.name == name and token.token_type == "service":
                return token
        return None

    def create_auth_token(
        self,
        user_id: str,
        *,
        name: str,
        token_hash: str,
        token_prefix: str,
        permission: str,
        token_type: str,
        expires_at: datetime | None = None,
        runtime_id: str | None = None,
    ) -> AuthToken:
        token = AuthToken(
            id=f"token-{self._seq}",
            user_id=user_id,
            name=name,
            token_prefix=token_prefix,
            permission=permission,
            token_type=token_type,
            expires_at=expires_at,
            created_at=datetime.now(timezone.utc),
            runtime_id=runtime_id,
        )
        self._seq += 1
        self.tokens[token.id] = token
        return token

    def revoke_auth_token(self, user_id: str, token_id: str) -> None:
        token = self.tokens.get(token_id)
        if token is None or token.user_id != user_id:
            raise ValueError("토큰을 찾을 수 없습니다")
        token.revoked_at = datetime.now(timezone.utc)


def _service(repo: _AuthRepo) -> AuthService:
    return AuthService(repo, Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000"))


def test_issue_service_token_defaults_to_365_days() -> None:
    from dataclasses import dataclass as _dc
    from app.domain.models import User

    repo = _AuthRepo()
    repo.users["alice"] = User(id="user-1", username="alice", permission="read")
    service = _service(repo)

    result = service.issue_service_token("my-token", "alice")

    assert result.auth_token.expires_at is not None
    assert (result.auth_token.expires_at - datetime.now(timezone.utc)).days >= 364
    assert result.auth_token.token_type == "service"
    assert result.auth_token.name == "my-token"


def test_issue_service_token_no_expiry() -> None:
    from app.domain.models import User

    repo = _AuthRepo()
    repo.users["alice"] = User(id="user-1", username="alice", permission="read")
    service = _service(repo)

    result = service.issue_service_token("my-token", "alice", no_expiry=True)

    assert result.auth_token.expires_at is None


def test_issue_service_token_unknown_user_raises() -> None:
    repo = _AuthRepo()
    service = _service(repo)

    try:
        service.issue_service_token("my-token", "unknown")
    except ValueError as exc:
        assert "unknown" in str(exc)
    else:
        raise AssertionError("Expected ValueError for unknown user")


def test_revoke_service_token_success() -> None:
    from app.domain.models import User

    repo = _AuthRepo()
    repo.users["alice"] = User(id="user-1", username="alice", permission="read")
    service = _service(repo)
    issued = service.issue_service_token("my-token", "alice")

    revoked_at = service.revoke_service_token("my-token")

    assert revoked_at is not None
    assert repo.tokens[issued.auth_token.id].revoked_at is not None


def test_revoke_service_token_not_found_raises() -> None:
    repo = _AuthRepo()
    service = _service(repo)

    try:
        service.revoke_service_token("nonexistent")
    except ValueError as exc:
        assert "nonexistent" in str(exc)
    else:
        raise AssertionError("Expected ValueError for missing service token")