"""Tests for BAT-024 / REQ-056 manual ledger API.""" from __future__ import annotations from contextlib import ExitStack from datetime import datetime, timezone from typing import Any from unittest.mock import patch 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 _PORTFOLIO_ID = "pf-001" _USER_ID = "user-001" _OTHER_USER_ID = "user-002" def _make_settings() -> Settings: return Settings(allowed_origins_raw="http://localhost:3000", db_enabled=True) def _token_principal( user_id: str = _USER_ID, *, token_type: str = "session", permission: str = "read", ) -> TokenPrincipal: return TokenPrincipal( token_id=f"tok-{user_id}-{token_type}", user_id=user_id, permission=permission, token_type=token_type, ) def _portfolio( *, owner: str = _USER_ID, status: str = "enabled", cash: float = 1000.0, quantity: float = 10.0, price: float = 10.0, buffer: float = 500.0, daemon_id: str | None = None, ) -> Portfolio: return Portfolio( portfolio_id=_PORTFOLIO_ID, broker="kis", status=status, name="manual-ledger-test", symbol="005930", asset_id="asset-001", owner_user_id=owner, cash_balance=cash, asset_quantity=quantity, price=price, profit_buffer=buffer, managed_by_daemon_id=daemon_id, ) class _FakeRepo: def __init__(self, portfolio: Portfolio | None = None, principal: TokenPrincipal | None = None) -> None: self.portfolio = portfolio or _portfolio() self.principal = principal or _token_principal() self.entries: list[LedgerEntry] = [] self._next_id = 1 def close(self) -> None: pass def get_token_principal(self, token_hash: str) -> TokenPrincipal | None: return self.principal def get(self, portfolio_id: str) -> Portfolio: if portfolio_id != self.portfolio.portfolio_id: raise ValueError("포트폴리오를 찾을 수 없습니다") return self.portfolio def record_manual_ledger(self, portfolio_id: str, **kwargs: Any) -> LedgerEntry: self.get(portfolio_id) action = kwargs["action"] amount = float(kwargs.get("amount") or 0.0) quantity = float(kwargs.get("quantity") or 0.0) price = float(kwargs.get("price") or 0.0) fee = float(kwargs.get("fee") or 0.0) tax = float(kwargs.get("tax") or 0.0) target = kwargs.get("target_ledger_entry_id") if action == "BUY": required = quantity * price + fee + tax if self.portfolio.cash_balance < required: raise ValueError("INSUFFICIENT_CASH") self.portfolio.cash_balance -= required self.portfolio.asset_quantity += quantity self.portfolio.price = price elif action == "SELL": if self.portfolio.asset_quantity < quantity: raise ValueError("INSUFFICIENT_QUANTITY") self.portfolio.cash_balance += quantity * price - fee - tax self.portfolio.asset_quantity -= quantity self.portfolio.price = price elif action == "INVEST": self.portfolio.profit_buffer += amount price = amount elif action == "BUFFER_OUT": if self.portfolio.profit_buffer < amount: raise ValueError("INSUFFICIENT_BUFFER") self.portfolio.profit_buffer -= amount self.portfolio.cash_balance += amount price = amount elif action == "BUFFER_IN": if self.portfolio.cash_balance < amount: raise ValueError("INSUFFICIENT_CASH") self.portfolio.cash_balance -= amount self.portfolio.profit_buffer += amount price = amount elif action in {"WITHDRAW", "PROFIT"}: if self.portfolio.profit_buffer < amount: raise ValueError("INSUFFICIENT_BUFFER") self.portfolio.profit_buffer -= amount price = amount elif action == "REVERSAL": target_entry = next((e for e in self.entries if e.ledger_entry_id == target), None) if target_entry is None: raise ValueError("LEDGER_ENTRY_NOT_FOUND") if any(e.target_ledger_entry_id == target and e.action == "REVERSAL" for e in self.entries): raise ValueError("LEDGER_ENTRY_ALREADY_REVERSED") price = 0.0 total = self.portfolio.cash_balance + self.portfolio.asset_quantity * self.portfolio.price + self.portfolio.profit_buffer entry = LedgerEntry( ledger_entry_id=self._next_id, portfolio_id=portfolio_id, action=action, side=action.lower(), quantity=quantity, price=price, fee_amount=fee, tax_amount=tax, recorded_at=datetime.now(timezone.utc), cash_balance_after=self.portfolio.cash_balance, asset_quantity_after=self.portfolio.asset_quantity, profit_buffer_after=self.portfolio.profit_buffer, total_value_after=total, note=kwargs.get("note") or "", source="manual", target_ledger_entry_id=target, confirmation_evidence=kwargs.get("confirmation_evidence"), actor_user_id=kwargs.get("actor_user_id"), ) self._next_id += 1 self.entries.append(entry) return entry def list_manual_ledger_entries(self, portfolio_id: str, limit: int = 50) -> list[LedgerEntry]: self.get(portfolio_id) return list(reversed(self.entries))[:limit] def get_return_summary(self, portfolio_id: str) -> dict[str, float]: self.get(portfolio_id) reversed_ids = { e.target_ledger_entry_id for e in self.entries if e.action == "REVERSAL" and e.target_ledger_entry_id is not None } active = [e for e in self.entries if e.ledger_entry_id not in reversed_ids] invested = sum(e.price for e in active if e.action == "INVEST") withdrawn = sum(e.price for e in active if e.action == "WITHDRAW") profit = sum(e.price for e in active if e.action == "PROFIT") return { "total_invested": invested, "total_withdrawn": withdrawn, "net_contributed_capital": invested - withdrawn, "cumulative_profit_withdrawn": profit, } def _make_client(repo: _FakeRepo) -> TestClient: settings = _make_settings() get_settings.cache_clear() app = create_app(settings_factory=lambda: settings) app.dependency_overrides[get_settings] = lambda: settings stack = ExitStack() stack.enter_context(patch("app.core.auth.get_portfolio_repository_from_settings", return_value=repo)) stack.enter_context(patch("app.api.v1_portfolios.get_portfolio_repository_from_settings", return_value=repo)) client = TestClient(app) client._repo_stack = stack # type: ignore[attr-defined] # keep patches alive for the client lifetime return client def _auth() -> dict[str, str]: return {"Authorization": "Bearer user-token"} def test_manual_buy_updates_snapshot_and_returns_total_value_after() -> None: repo = _FakeRepo(_portfolio(cash=1000, quantity=1, price=10, buffer=100)) client = _make_client(repo) resp = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "BUY", "quantity": 2, "price": 100, "fee": 1, "tax": 2}, headers=_auth(), ) assert resp.status_code == 201 data = resp.json() assert data["action"] == "BUY" assert data["cash_balance_after"] == 797.0 assert data["asset_quantity_after"] == 3.0 assert data["profit_buffer_after"] == 100.0 assert data["total_value_after"] == 1197.0 def test_manual_write_rejects_service_token() -> None: repo = _FakeRepo(principal=_token_principal(token_type="service")) client = _make_client(repo) resp = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "INVEST", "amount": 100}, headers=_auth(), ) assert resp.status_code == 403 def test_daemon_managed_requires_confirmation_then_records_evidence() -> None: repo = _FakeRepo(_portfolio(daemon_id="daemon-1")) client = _make_client(repo) first = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "PROFIT", "amount": 100}, headers=_auth(), ) assert first.status_code == 409 assert first.json()["detail"]["code"] == "MANUAL_WRITE_REQUIRES_CONFIRMATION" second = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "PROFIT", "amount": 100, "confirm_daemon_managed_manual_write": True}, headers=_auth(), ) assert second.status_code == 201 assert second.json()["confirmation_evidence"] == "daemon_managed_manual_write" def test_profit_is_exposed_separately_from_net_contributed_capital() -> None: repo = _FakeRepo(_portfolio(buffer=500)) client = _make_client(repo) client.post(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "INVEST", "amount": 300}, headers=_auth()) client.post(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "WITHDRAW", "amount": 100}, headers=_auth()) client.post(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "PROFIT", "amount": 50}, headers=_auth()) resp = client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/returns/summary", headers=_auth()) assert resp.status_code == 200 data = resp.json() assert data["net_contributed_capital"] == 200.0 assert data["cumulative_profit_withdrawn"] == 50.0 def test_return_summary_excludes_reversed_invest_withdraw_and_profit() -> None: repo = _FakeRepo(_portfolio(buffer=1_000)) client = _make_client(repo) entries = [] for action, amount in (("INVEST", 300), ("WITHDRAW", 100), ("PROFIT", 50)): response = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": action, "amount": amount}, headers=_auth(), ) assert response.status_code == 201 entries.append(response.json()["ledger_entry_id"]) for target in entries: response = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "REVERSAL", "target_ledger_entry_id": target}, headers=_auth(), ) assert response.status_code == 201 resp = client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/returns/summary", headers=_auth()) assert resp.status_code == 200 assert resp.json() == { "portfolio_id": _PORTFOLIO_ID, "total_invested": 0.0, "total_withdrawn": 0.0, "net_contributed_capital": 0.0, "cumulative_profit_withdrawn": 0.0, } def test_insufficient_cash_rejected_as_409() -> None: repo = _FakeRepo(_portfolio(cash=10, quantity=0, price=0)) client = _make_client(repo) resp = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "BUY", "quantity": 1, "price": 100}, headers=_auth(), ) assert resp.status_code == 409 assert resp.json()["detail"]["code"] == "INSUFFICIENT_CASH" def test_manual_history_returns_total_value_after() -> None: repo = _FakeRepo(_portfolio(buffer=500)) client = _make_client(repo) client.post(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "PROFIT", "amount": 100}, headers=_auth()) resp = client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", headers=_auth()) assert resp.status_code == 200 item = resp.json()["items"][0] assert item["action"] == "PROFIT" assert "total_value_after" in item def test_owner_mismatch_returns_403() -> None: repo = _FakeRepo(_portfolio(owner=_OTHER_USER_ID)) client = _make_client(repo) resp = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "INVEST", "amount": 100}, headers=_auth(), ) assert resp.status_code == 403 def test_admin_session_does_not_bypass_manual_owner_scope() -> None: repo = _FakeRepo( _portfolio(owner=_OTHER_USER_ID), principal=_token_principal(user_id=_USER_ID, token_type="session", permission="admin"), ) client = _make_client(repo) resp = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "INVEST", "amount": 100}, headers=_auth(), ) assert resp.status_code == 403 def test_admin_session_does_not_bypass_manual_read_owner_scope() -> None: repo = _FakeRepo( _portfolio(owner=_OTHER_USER_ID), principal=_token_principal(user_id=_USER_ID, token_type="session", permission="admin"), ) client = _make_client(repo) history = client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", headers=_auth()) summary = client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/returns/summary", headers=_auth()) assert history.status_code == 403 assert summary.status_code == 403 def test_amount_actions_reject_fee_and_tax_fields() -> None: repo = _FakeRepo(_portfolio()) client = _make_client(repo) for payload in [ {"action": "INVEST", "amount": 100, "fee": 1}, {"action": "INVEST", "amount": 100, "tax": 1}, {"action": "BUFFER_OUT", "amount": 100, "fee": 0}, ]: resp = client.post(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json=payload, headers=_auth()) assert resp.status_code == 422 def test_reversal_rejects_fee_tax_and_missing_target() -> None: repo = _FakeRepo(_portfolio()) client = _make_client(repo) invalid_field = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "REVERSAL", "target_ledger_entry_id": 999, "fee": 1}, headers=_auth(), ) assert invalid_field.status_code == 422 missing_target = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "REVERSAL", "target_ledger_entry_id": 999}, headers=_auth(), ) assert missing_target.status_code == 404 assert missing_target.json()["detail"]["code"] == "LEDGER_ENTRY_NOT_FOUND"