362 lines
13 KiB
Python
362 lines
13 KiB
Python
"""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)
|
|
invested = sum(e.price for e in self.entries if e.action == "INVEST")
|
|
withdrawn = sum(e.price for e in self.entries if e.action == "WITHDRAW")
|
|
profit = sum(e.price for e in self.entries 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_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"
|