portfolio-hub/tests/test_users.py

from __future__ import annotations

from datetime import datetime, timezone

from fastapi.testclient import TestClient

import app.api.users as users_api
from app.core.security import hash_password, hash_token
from app.domain.models import AuthToken, TokenPrincipal, User
from app.main import app
from app.core.auth import get_auth_repository
from app.services.auth_service import AuthService, get_auth_service
from app.services.user_service import UserService, get_user_service


class FakeUserRepo:
    def __init__(self) -> None:
        self.users: dict[str, User] = {}
        self.users_by_username: dict[str, User] = {}
        self.tokens: dict[str, AuthToken] = {}
        self.token_principals: dict[str, TokenPrincipal] = {}
        self.next_id = 1
        self.next_token_id = 1

    def get_user(self, user_id: str) -> User:
        user = self.users.get(user_id)
        if user is None:
            raise ValueError("사용자를 찾을 수 없습니다")
        return user

    def get_user_by_username(self, username: str) -> User | None:
        return self.users_by_username.get(username)

    def list_users(self) -> list[User]:
        return sorted(self.users.values(), key=lambda user: user.username)

    def create_user(
        self,
        username: str,
        *,
        email: str | None = None,
        password_hash: str | None = None,
        permission: str = "read",
        is_active: bool = True,
    ) -> User:
        if any(user.username == username for user in self.users.values()):
            raise ValueError(f"이미 존재하는 username입니다: {username}")
        if email is not None and any(user.email == email for user in self.users.values()):
            raise ValueError(f"이미 존재하는 email입니다: {email}")
        now = datetime.now(timezone.utc)
        user = User(
            id=f"user-{self.next_id}",
            username=username,
            email=email,
            permission=permission,
            is_active=is_active,
            password_hash=password_hash,
            created_at=now,
            updated_at=now,
        )
        self.next_id += 1
        self.users[user.id] = user
        self.users_by_username[user.username] = user
        return user

    def update_user(
        self,
        user_id: str,
        *,
        username: str | None = None,
        email: str | None = None,
        password_hash: str | None = None,
        permission: str | None = None,
        is_active: bool | None = None,
    ) -> User:
        user = self.get_user(user_id)
        if username is not None and any(item.id != user_id and item.username == username for item in self.users.values()):
            raise ValueError(f"이미 존재하는 username입니다: {username}")
        if email is not None and any(item.id != user_id and item.email == email for item in self.users.values()):
            raise ValueError(f"이미 존재하는 email입니다: {email}")
        self.users_by_username.pop(user.username, None)
        if username is not None:
            user.username = username
        if email is not None:
            user.email = email
        if password_hash is not None:
            user.password_hash = password_hash
        if permission is not None:
            user.permission = permission
        if is_active is not None:
            user.is_active = is_active
        user.updated_at = datetime.now(timezone.utc)
        self.users_by_username[user.username] = user
        return user

    def delete_user(self, user_id: str) -> None:
        if user_id not in self.users:
            raise ValueError("사용자를 찾을 수 없습니다")
        user = self.users.pop(user_id)
        self.users_by_username.pop(user.username, None)

    def update_user_last_login(self, user_id: str) -> None:
        user = self.get_user(user_id)
        user.last_login_at = datetime.now(timezone.utc)

    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,
    ) -> AuthToken:
        now = datetime.now(timezone.utc)
        token = AuthToken(
            id=f"token-{self.next_token_id}",
            user_id=user_id,
            name=name,
            token_prefix=token_prefix,
            permission=permission,
            token_type=token_type,
            expires_at=expires_at,
            created_at=now,
        )
        self.next_token_id += 1
        self.tokens[token.id] = token
        self.token_principals[token_hash] = TokenPrincipal(
            token_id=token.id,
            user_id=user_id,
            permission=permission,
            token_type=token_type,
        )
        return token

    def list_auth_tokens(self, user_id: str) -> list[AuthToken]:
        return [token for token in self.tokens.values() if token.user_id == user_id]

    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 get_service_token_by_name(self, name: str) -> AuthToken | None:
        return next((t for t in self.tokens.values() if t.name == name and t.token_type == "service"), None)

    def get_token_principal(self, token_hash_value: str) -> TokenPrincipal | None:
        principal = self.token_principals.get(token_hash_value)
        if principal is None:
            return None
        token = self.tokens.get(principal.token_id)
        if token is None or token.revoked_at is not None:
            return None
        if token.expires_at is not None and token.expires_at <= datetime.now(timezone.utc):
            return None
        token.last_used_at = datetime.now(timezone.utc)
        return principal


def _client_with_repo(monkeypatch, repo: FakeUserRepo) -> TestClient:
    monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true")
    monkeypatch.setenv("PORTFOLIOHUB_ADMIN_TOKENS", "admin-token")
    monkeypatch.setenv("PORTFOLIOHUB_READ_TOKENS", "read-token")
    from app.core.config import get_settings

    get_settings.cache_clear()
    app.dependency_overrides.clear()
    app.dependency_overrides[get_user_service] = lambda: UserService(repo)
    app.dependency_overrides[get_auth_repository] = lambda: repo
    app.dependency_overrides[get_auth_service] = lambda: AuthService(repo, get_settings())
    monkeypatch.setattr(users_api, "get_portfolio_repository_from_settings", lambda settings: repo)
    return TestClient(app)


def test_list_users_requires_admin(monkeypatch) -> None:
    client = _client_with_repo(monkeypatch, FakeUserRepo())

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

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


def test_list_users_returns_payload(monkeypatch) -> None:
    repo = FakeUserRepo()
    repo.create_user("alice", email="alice@example.com", password_hash=hash_password("password123"))
    client = _client_with_repo(monkeypatch, repo)

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

    app.dependency_overrides.clear()
    assert response.status_code == 200
    payload = response.json()
    assert len(payload) == 1
    assert payload[0]["username"] == "alice"


def test_create_user_rejects_duplicate_username(monkeypatch) -> None:
    repo = FakeUserRepo()
    repo.create_user("alice", password_hash=hash_password("password123"))
    client = _client_with_repo(monkeypatch, repo)

    response = client.post(
        "/users",
        headers={"Authorization": "Bearer admin-token"},
        json={"username": "alice", "email": "alice2@example.com", "password": "password123"},
    )

    app.dependency_overrides.clear()
    assert response.status_code == 409
    assert "이미 존재하는 username입니다" in response.json()["detail"]


def test_get_update_and_delete_user_flow(monkeypatch) -> None:
    repo = FakeUserRepo()
    user = repo.create_user("alice", email="alice@example.com", password_hash=hash_password("password123"))
    client = _client_with_repo(monkeypatch, repo)

    get_response = client.get(f"/users/{user.id}", headers={"Authorization": "Bearer admin-token"})
    update_response = client.patch(
        f"/users/{user.id}",
        headers={"Authorization": "Bearer admin-token"},
        json={"username": "alice-updated"},
    )
    delete_response = client.delete(f"/users/{user.id}", headers={"Authorization": "Bearer admin-token"})

    app.dependency_overrides.clear()
    assert get_response.status_code == 200
    assert get_response.json()["username"] == "alice"
    assert update_response.status_code == 200
    assert update_response.json()["username"] == "alice-updated"
    assert delete_response.status_code == 204