portfolio-hub/tests/test_portfolio_ownership.py

from __future__ import annotations

from datetime import datetime, timezone

from fastapi.testclient import TestClient

import app.api.portfolios as portfolios_api
import app.core.auth as auth_core
from app.core.config import get_settings
from app.core.security import hash_token
from app.domain.models import LedgerEntry, Portfolio, StrategySet, TokenPrincipal
from app.main import app


class FakeRepo:
    def __init__(self, *, portfolio: Portfolio, principals: dict[str, TokenPrincipal]) -> None:
        self._portfolio = portfolio
        self._principals = principals
        self._strategy = StrategySet(id="set-1", name="target_value", config="{}", nick="target_value:demo")

    def close(self) -> None:
        return None

    def list_users(self):
        return []

    def get_token_principal(self, token_hash: str):
        return self._principals.get(token_hash)

    def get(self, portfolio_id: str) -> Portfolio:
        if self._portfolio.portfolio_id != portfolio_id:
            raise ValueError("포트폴리오를 찾을 수 없습니다")
        return self._portfolio

    def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]:
        return []

    def list_portfolios(
        self,
        *,
        broker: str | None = None,
        status: str | None = None,
        owner_user_id: str | None = None,
        include_tags: list[str] | None = None,
        exclude_tags: list[str] | None = None,
        match_mode: str | None = None,
    ):
        if owner_user_id is not None and self._portfolio.owner_user_id != owner_user_id:
            return []
        return [self._portfolio]

    def add_tag(self, portfolio_id: str, *, tag: str):
        raise NotImplementedError()

    def list_tags(self, portfolio_id: str):
        return []

    def remove_tag(self, portfolio_id: str, *, tag: str) -> None:
        raise NotImplementedError()

    def get_strategy_set(self, id: str):
        if id == self._strategy.id:
            return self._strategy
        return None


def _portfolio(owner_user_id: str) -> Portfolio:
    return Portfolio(
        portfolio_id="p1",
        broker="kis",
        status="enabled",
        name="Samsung Electronics",
        symbol="005930",
        owner_user_id=owner_user_id,
        cash_balance=1_000_000.0,
        asset_quantity=1.0,
        price=70_000.0,
        profit_buffer=0.0,
        active_strategy_id="set-1",
        avabile_strategies=["set-1"],
        created_at=datetime.now(timezone.utc),
        updated_at=datetime.now(timezone.utc),
    )


def _configure_client(monkeypatch, repo: FakeRepo) -> TestClient:
    monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true")
    monkeypatch.delenv("PORTFOLIOHUB_READ_TOKENS", raising=False)
    monkeypatch.delenv("PORTFOLIOHUB_TRADE_WRITE_TOKENS", raising=False)
    monkeypatch.delenv("PORTFOLIOHUB_ADMIN_TOKENS", raising=False)
    get_settings.cache_clear()
    app.dependency_overrides.clear()
    monkeypatch.setattr(portfolios_api, "get_portfolio_repository_from_settings", lambda settings: repo)
    monkeypatch.setattr(auth_core, "get_portfolio_repository_from_settings", lambda settings: repo)
    return TestClient(app)


def test_owner_can_read_owned_portfolio_on_db_auth(monkeypatch) -> None:
    token = "owner-db-token"
    principal = TokenPrincipal(
        token_id="t1",
        user_id="owner-1",
        permission="read",
        token_type="client",
    )
    repo = FakeRepo(portfolio=_portfolio("owner-1"), principals={hash_token(token): principal})
    client = _configure_client(monkeypatch, repo)

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

    app.dependency_overrides.clear()
    assert response.status_code == 200
    assert response.json()["portfolio"]["owner_user_id"] == "owner-1"


def test_non_owner_gets_403_on_db_auth(monkeypatch) -> None:
    token = "other-db-token"
    principal = TokenPrincipal(
        token_id="t2",
        user_id="other-user",
        permission="read",
        token_type="client",
    )
    repo = FakeRepo(portfolio=_portfolio("owner-1"), principals={hash_token(token): principal})
    client = _configure_client(monkeypatch, repo)

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

    app.dependency_overrides.clear()
    assert response.status_code == 403
    assert response.json()["detail"] == "Portfolio ownership mismatch"


def test_owner_bound_bot_token_can_read_owned_portfolio(monkeypatch) -> None:
    token = "owner-bot-token"
    principal = TokenPrincipal(
        token_id="t-bot-1",
        user_id="owner-1",
        permission="read",
        token_type="service",
    )
    repo = FakeRepo(portfolio=_portfolio("owner-1"), principals={hash_token(token): principal})
    client = _configure_client(monkeypatch, repo)

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

    app.dependency_overrides.clear()
    assert response.status_code == 200
    assert response.json()["portfolio"]["owner_user_id"] == "owner-1"


def test_db_enabled_env_token_without_user_context_is_rejected(monkeypatch) -> None:
    repo = FakeRepo(portfolio=_portfolio("owner-1"), principals={})
    monkeypatch.setenv("PORTFOLIOHUB_DB_ENABLED", "true")
    monkeypatch.setenv("PORTFOLIOHUB_TRADE_WRITE_TOKENS", "trade-env-token")
    monkeypatch.delenv("PORTFOLIOHUB_READ_TOKENS", raising=False)
    monkeypatch.delenv("PORTFOLIOHUB_ADMIN_TOKENS", raising=False)
    get_settings.cache_clear()
    app.dependency_overrides.clear()
    monkeypatch.setattr(portfolios_api, "get_portfolio_repository_from_settings", lambda settings: repo)
    monkeypatch.setattr(auth_core, "get_portfolio_repository_from_settings", lambda settings: repo)
    client = TestClient(app)

    response = client.post(
        "/portfolios",
        headers={"Authorization": "Bearer trade-env-token"},
        json={"name": "new-1", "asset_id": "00000000-0000-0000-0000-000000000001", "strategy_id": "set-1"},
    )

    app.dependency_overrides.clear()
    assert response.status_code == 403
    assert response.json()["detail"] == "Ownership context is required"