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, BotPrincipal
from app.main import app
from app.schemas.auth import CreateBotTokenRequest
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:
    fail_audit: bool = False

    def __post_init__(self) -> None:
        self.bot = BotPrincipal(
            id="bot-1",
            name="bot",
            owner_user_id="owner-1",
            permission="trade-write",
            is_active=True,
        )
        self.tokens: dict[str, AuthToken] = {}
        self._seq = 1

    def get_bot_principal(self, bot_principal_id: str) -> BotPrincipal:
        if bot_principal_id != self.bot.id:
            raise ValueError("봇을 찾을 수 없습니다")
        return self.bot

    def create_auth_token(
        self,
        user_id: str,
        *,
        name: str,
        token_hash: str,
        token_prefix: str,
        permission: str,
        token_type: str,
        bot_principal_id: str | None = None,
        expires_at: datetime | 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,
            bot_principal_id=bot_principal_id,
            expires_at=expires_at,
            created_at=datetime.now(timezone.utc),
        )
        self._seq += 1
        self.tokens[token.id] = token
        return token

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

    def record_audit_event(self, payload: dict[str, object]) -> None:
        if self.fail_audit:
            raise RuntimeError("audit failed")


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


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

    result = service.issue_bot_token("bot-1", CreateBotTokenRequest(name="cli"))

    assert result.auth_token.expires_at is not None
    assert (result.auth_token.expires_at - datetime.now(timezone.utc)).days >= 364


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

    auth_context = AuthContext(
        token="db-token",
        permission=Permission.TRADE_WRITE,
        user_id="owner-1",
        token_source="db",
    )

    try:
        service.issue_owned_bot_token(
            auth_context,
            "bot-1",
            CreateBotTokenRequest(name="cli", no_expiry=True),
        )
    except ValueError as exc:
        assert "비관리자" in str(exc)
    else:
        raise AssertionError("Expected ValueError for non-admin no-expiry")


def test_issue_bot_token_rolls_back_when_audit_fails() -> None:
    repo = _AuthRepo(fail_audit=True)
    service = _service(repo)

    try:
        service.issue_bot_token("bot-1", CreateBotTokenRequest(name="cli"))
    except ValueError as exc:
        assert "발급을 취소" in str(exc)
    else:
        raise AssertionError("Expected ValueError on audit failure")

    assert len(repo.tokens) == 1
    token = next(iter(repo.tokens.values()))
    assert token.revoked_at is not None


def test_revoke_bot_token_keeps_revoke_when_audit_fails() -> None:
    repo = _AuthRepo(fail_audit=True)
    service = _service(repo)
    created = repo.create_auth_token(
        "bot-1",
        name="cli",
        token_hash="hash",
        token_prefix="prefix",
        permission="trade-write",
        token_type="bot",
        bot_principal_id="bot-1",
    )

    try:
        service.revoke_bot_token("bot-1", created.id)
    except ValueError as exc:
        assert "토큰은 폐기" in str(exc)
    else:
        raise AssertionError("Expected ValueError on audit failure")

    assert repo.tokens[created.id].revoked_at is not None