"""Tests for GET /v1/portfolios/{portfolio_id}/pnl/realized (REQ-040, REQ-042). Uses TestClient with MagicMock repo injection — no DB required. WAC computation is exercised end-to-end through the HTTP layer. """ from __future__ import annotations from contextlib import ExitStack from datetime import datetime, timezone from unittest.mock import MagicMock, patch import pytest from fastapi.testclient import TestClient from app.core.config import Settings, get_settings from app.domain.models import LedgerEntry, Portfolio, TokenPrincipal from app.main import create_app # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- _ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=" _USER_ID = "user-001" _OTHER_USER_ID = "user-002" _PORTFOLIO_ID = "pf-pnl-001" _PATH = f"/v1/portfolios/{_PORTFOLIO_ID}/pnl/realized" _DEFAULT_PARAMS = {"from": "2026-06-01", "to": "2026-06-30"} # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_settings() -> Settings: get_settings.cache_clear() return Settings( db_enabled=True, credential_encryption_key=_ENC_KEY, allowed_origins_raw="http://localhost:3000", ) def _make_client(settings: Settings) -> TestClient: app = create_app(settings_factory=lambda: settings) app.dependency_overrides[get_settings] = lambda: settings return TestClient(app) def _make_portfolio(*, owner: str = _USER_ID, symbol: str = "005930") -> Portfolio: return Portfolio( portfolio_id=_PORTFOLIO_ID, broker="kis", status="enabled", name="test-pnl-portfolio", symbol=symbol, asset_id="asset-001", owner_user_id=owner, cash_balance=1_000_000.0, asset_quantity=0.0, tag_ids=[], managed_by_daemon_id=None, ) def _make_entry( *, ledger_entry_id: int, side: str, quantity: float, price: float, fee: float = 0.0, tax: float = 0.0, executed_at: datetime, ) -> LedgerEntry: return LedgerEntry( ledger_entry_id=ledger_entry_id, portfolio_id=_PORTFOLIO_ID, order_id=f"ord-{ledger_entry_id:03d}", execution_id=f"exec-{ledger_entry_id:03d}", side=side, quantity=quantity, price=price, fee_amount=fee, tax_amount=tax, executed_at=executed_at, recorded_at=executed_at, ) def _make_repo( *, portfolio: Portfolio | None = None, trades: list[LedgerEntry] | None = None, user_id: str = _USER_ID, permission: str = "read", ) -> MagicMock: repo = MagicMock() principal = TokenPrincipal( token_id="t1", user_id=user_id, permission=permission, token_type="db", runtime_id=None, daemon_id=None, ) repo.get_token_principal.return_value = principal repo.get.return_value = portfolio or _make_portfolio() repo.get_all_trades_for_pnl.return_value = trades if trades is not None else [] return repo def _with_repo(repo: MagicMock) -> ExitStack: stack = ExitStack() stack.enter_context( patch("app.api.v1_portfolios.get_portfolio_repository_from_settings", return_value=repo) ) stack.enter_context( patch("app.core.auth.get_portfolio_repository_from_settings", return_value=repo) ) return stack # --------------------------------------------------------------------------- # REQ-040 AC-1 / REQ-042 AC-3: WAC 199,300 scenario end-to-end # # buy 100주@10,000 + buy 100주@12,000 → sell 50주@15,000 fee=500 tax=200 # WAC = (100*10,000 + 100*12,000) / 200 = 11,000 # realized_pnl = (15,000 - 11,000) * 50 - 500 - 200 = 199,300 # --------------------------------------------------------------------------- class TestWac199300EndToEnd: def _make_trades(self) -> list[LedgerEntry]: return [ _make_entry( ledger_entry_id=1, side="buy", quantity=100, price=10_000, executed_at=datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc), ), _make_entry( ledger_entry_id=2, side="buy", quantity=100, price=12_000, executed_at=datetime(2026, 6, 5, 9, 0, tzinfo=timezone.utc), ), _make_entry( ledger_entry_id=3, side="sell", quantity=50, price=15_000, fee=500, tax=200, executed_at=datetime(2026, 6, 15, 10, 0, tzinfo=timezone.utc), ), ] def test_total_realized_pnl_is_199300(self) -> None: settings = _make_settings() repo = _make_repo(trades=self._make_trades()) with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 body = resp.json() assert body["total_realized_pnl"] == pytest.approx(199_300.0) def test_cost_basis_per_unit_is_11000(self) -> None: settings = _make_settings() repo = _make_repo(trades=self._make_trades()) with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) body = resp.json() assert len(body["items"]) == 1 assert body["items"][0]["cost_basis_per_unit"] == pytest.approx(11_000.0) def test_items_contain_required_fields(self) -> None: settings = _make_settings() repo = _make_repo(trades=self._make_trades()) with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) item = resp.json()["items"][0] for field in ("executed_at", "quantity", "sell_price", "cost_basis_per_unit", "fee_amount", "tax_amount", "realized_pnl"): assert field in item, f"Missing field: {field}" def test_data_completeness_warning_is_null_when_complete(self) -> None: settings = _make_settings() repo = _make_repo(trades=self._make_trades()) with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) body = resp.json() assert body["data_completeness_warning"] is None def test_cost_basis_from_is_first_buy_date(self) -> None: settings = _make_settings() repo = _make_repo(trades=self._make_trades()) with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) body = resp.json() assert body["cost_basis_from"] is not None # --------------------------------------------------------------------------- # REQ-040 AC-3: from/to missing → 422 # --------------------------------------------------------------------------- def test_pnl_missing_from_returns_422() -> None: settings = _make_settings() repo = _make_repo() with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params={"to": "2026-06-30"}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 422 def test_pnl_missing_to_returns_422() -> None: settings = _make_settings() repo = _make_repo() with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params={"from": "2026-06-01"}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 422 def test_pnl_both_missing_returns_422() -> None: settings = _make_settings() repo = _make_repo() with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 422 # --------------------------------------------------------------------------- # REQ-040 AC-4: no sells in period → 200 + total=0 + items=[] # --------------------------------------------------------------------------- def test_pnl_no_sells_returns_200_with_zero() -> None: trades = [ _make_entry(ledger_entry_id=1, side="buy", quantity=100, price=10_000, executed_at=datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)), ] settings = _make_settings() repo = _make_repo(trades=trades) with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 body = resp.json() assert body["total_realized_pnl"] == pytest.approx(0.0) assert body["items"] == [] def test_pnl_empty_ledger_returns_200_with_zero() -> None: settings = _make_settings() repo = _make_repo(trades=[]) with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 body = resp.json() assert body["total_realized_pnl"] == pytest.approx(0.0) assert body["items"] == [] # --------------------------------------------------------------------------- # REQ-040 AC-5/AC-6: data_completeness_warning (R-1 완화) # --------------------------------------------------------------------------- def test_pnl_sell_without_buy_triggers_warning() -> None: """AC-5: history incomplete → warning non-null.""" trades = [ _make_entry(ledger_entry_id=1, side="sell", quantity=50, price=15_000, fee=500, tax=200, executed_at=datetime(2026, 6, 15, 10, 0, tzinfo=timezone.utc)), ] settings = _make_settings() repo = _make_repo(trades=trades) with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 body = resp.json() assert body["data_completeness_warning"] is not None assert len(body["data_completeness_warning"]) > 0 def test_pnl_complete_history_warning_is_null() -> None: """AC-6: complete history → warning null.""" trades = [ _make_entry(ledger_entry_id=1, side="buy", quantity=100, price=10_000, executed_at=datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)), _make_entry(ledger_entry_id=2, side="sell", quantity=50, price=15_000, fee=100, tax=50, executed_at=datetime(2026, 6, 15, 10, 0, tzinfo=timezone.utc)), ] settings = _make_settings() repo = _make_repo(trades=trades) with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 assert resp.json()["data_completeness_warning"] is None def test_pnl_oversell_triggers_warning() -> None: """AC-5: running_qty < sell qty → warning.""" trades = [ _make_entry(ledger_entry_id=1, side="buy", quantity=10, price=10_000, executed_at=datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)), _make_entry(ledger_entry_id=2, side="sell", quantity=20, price=15_000, executed_at=datetime(2026, 6, 15, 10, 0, tzinfo=timezone.utc)), ] settings = _make_settings() repo = _make_repo(trades=trades) with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 assert resp.json()["data_completeness_warning"] is not None # --------------------------------------------------------------------------- # REQ-040 AC-2: symbol filter # --------------------------------------------------------------------------- def test_pnl_symbol_filter_passed_to_repo() -> None: """Symbol filter param is accepted; repo.get_all_trades_for_pnl is called with symbol.""" settings = _make_settings() repo = _make_repo(trades=[]) with _with_repo(repo): client = _make_client(settings) resp = client.get( _PATH, params={"from": "2026-06-01", "to": "2026-06-30", "symbol": "005930.KS"}, headers={"Authorization": "Bearer tok"}, ) assert resp.status_code == 200 repo.get_all_trades_for_pnl.assert_called_once_with(_PORTFOLIO_ID, symbol="005930.KS") def test_pnl_symbol_filter_excludes_other_symbol_sells() -> None: """Portfolio symbol is '005930'; sells tagged with portfolio symbol only appear.""" portfolio = _make_portfolio(symbol="005930") trades = [ _make_entry(ledger_entry_id=1, side="buy", quantity=100, price=10_000, executed_at=datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)), _make_entry(ledger_entry_id=2, side="sell", quantity=50, price=15_000, executed_at=datetime(2026, 6, 15, 10, 0, tzinfo=timezone.utc)), ] settings = _make_settings() repo = _make_repo(portfolio=portfolio, trades=trades) with _with_repo(repo): client = _make_client(settings) # symbol filter matches portfolio symbol → sell included resp = client.get( _PATH, params={"from": "2026-06-01", "to": "2026-06-30", "symbol": "005930"}, headers={"Authorization": "Bearer tok"}, ) assert resp.status_code == 200 body = resp.json() # portfolio symbol "005930" matches filter "005930" → sell is included assert len(body["items"]) == 1 # --------------------------------------------------------------------------- # REQ-040 AC-7: non-owner token → 403 # --------------------------------------------------------------------------- def test_pnl_non_owner_returns_403() -> None: settings = _make_settings() repo = _make_repo(portfolio=_make_portfolio(owner=_USER_ID), user_id=_OTHER_USER_ID) with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 403 # --------------------------------------------------------------------------- # REQ-040 AC-8: non-existent portfolio → 404 # --------------------------------------------------------------------------- def test_pnl_nonexistent_portfolio_returns_404() -> None: settings = _make_settings() repo = _make_repo() repo.get.side_effect = ValueError("포트폴리오를 찾을 수 없습니다") with _with_repo(repo): client = _make_client(settings) resp = client.get(_PATH, params=_DEFAULT_PARAMS, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 404 # --------------------------------------------------------------------------- # REQ-040 AC-9: endpoint documented in OpenAPI # --------------------------------------------------------------------------- def test_pnl_endpoint_in_openapi() -> None: settings = _make_settings() repo = _make_repo() with _with_repo(repo): client = _make_client(settings) resp = client.get("/openapi.json") assert resp.status_code == 200 openapi = resp.json() paths = openapi.get("paths", {}) pnl_path = f"/v1/portfolios/{{portfolio_id}}/pnl/realized" assert pnl_path in paths, f"Expected {pnl_path!r} in OpenAPI paths" assert "get" in paths[pnl_path]