copilot-swe-agent[bot]andgenichin
b74284e626
Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
724 lines
27 KiB
Python
724 lines
27 KiB
Python
"""Tests for BAT-024 / REQ-056 manual ledger API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import ExitStack
|
|
from datetime import UTC, datetime
|
|
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)
|
|
executed_at = kwargs.get("executed_at")
|
|
confirm_duplicate = bool(kwargs.get("confirm_duplicate") or False)
|
|
note = kwargs.get("note") or ""
|
|
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 == "DIVIDEND":
|
|
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
|
|
}
|
|
duplicate = next(
|
|
(
|
|
e
|
|
for e in self.entries
|
|
if e.action == "DIVIDEND"
|
|
and e.ledger_entry_id not in reversed_ids
|
|
and e.executed_at == executed_at
|
|
and abs(float(e.price) - amount) < 1e-9
|
|
and e.note == note
|
|
),
|
|
None,
|
|
)
|
|
if duplicate is not None and not confirm_duplicate:
|
|
raise ValueError(f"POTENTIAL_DUPLICATE_DIVIDEND:{duplicate.ledger_entry_id}")
|
|
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")
|
|
target_action = target_entry.action or ""
|
|
target_amount = float(target_entry.price)
|
|
target_quantity = float(target_entry.quantity)
|
|
target_fee = float(target_entry.fee_amount)
|
|
target_tax = float(target_entry.tax_amount)
|
|
if target_action == "BUY":
|
|
self.portfolio.cash_balance += target_quantity * target_amount + target_fee + target_tax
|
|
self.portfolio.asset_quantity -= target_quantity
|
|
elif target_action == "SELL":
|
|
required_cash = target_quantity * target_amount - target_fee - target_tax
|
|
if self.portfolio.cash_balance < required_cash:
|
|
raise ValueError("INSUFFICIENT_CASH")
|
|
self.portfolio.cash_balance -= required_cash
|
|
self.portfolio.asset_quantity += target_quantity
|
|
elif target_action == "BUFFER_OUT":
|
|
if self.portfolio.cash_balance < target_amount:
|
|
raise ValueError("INSUFFICIENT_CASH")
|
|
self.portfolio.cash_balance -= target_amount
|
|
self.portfolio.profit_buffer += target_amount
|
|
elif target_action == "BUFFER_IN":
|
|
if self.portfolio.profit_buffer < target_amount:
|
|
raise ValueError("INSUFFICIENT_BUFFER")
|
|
self.portfolio.cash_balance += target_amount
|
|
self.portfolio.profit_buffer -= target_amount
|
|
elif target_action in {"INVEST", "DIVIDEND"}:
|
|
if self.portfolio.profit_buffer < target_amount:
|
|
raise ValueError("INSUFFICIENT_BUFFER")
|
|
self.portfolio.profit_buffer -= target_amount
|
|
elif target_action in {"WITHDRAW", "PROFIT"}:
|
|
self.portfolio.profit_buffer += target_amount
|
|
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,
|
|
executed_at=executed_at,
|
|
recorded_at=datetime.now(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=note,
|
|
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 list_ledger_entries(
|
|
self,
|
|
portfolio_id: str,
|
|
*,
|
|
source: str | None = None,
|
|
before_recorded_at: datetime | None = None,
|
|
before_ledger_entry_id: int | None = None,
|
|
limit: int = 50,
|
|
) -> list[LedgerEntry]:
|
|
self.get(portfolio_id)
|
|
entries = sorted(self.entries, key=lambda entry: (entry.recorded_at, entry.ledger_entry_id), reverse=True)
|
|
if source is not None:
|
|
entries = [entry for entry in entries if entry.source == source]
|
|
if before_recorded_at is not None and before_ledger_entry_id is not None:
|
|
entries = [
|
|
entry
|
|
for entry in entries
|
|
if (entry.recorded_at, entry.ledger_entry_id) < (before_recorded_at, before_ledger_entry_id)
|
|
]
|
|
return 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_dividend_updates_only_profit_buffer_and_records_executed_at_and_tax() -> None:
|
|
repo = _FakeRepo(_portfolio(cash=1000, quantity=10, price=10, buffer=500))
|
|
client = _make_client(repo)
|
|
|
|
resp = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "DIVIDEND", "amount": 120, "tax": 10, "executed_at": "2026-09-01", "note": "Q3"},
|
|
headers=_auth(),
|
|
)
|
|
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
assert body["action"] == "DIVIDEND"
|
|
assert body["amount"] == 120.0
|
|
assert body["tax"] == 10.0
|
|
assert body["executed_at"] == "2026-09-01T00:00:00Z"
|
|
assert body["cash_balance_after"] == 1000.0
|
|
assert body["asset_quantity_after"] == 10.0
|
|
assert body["profit_buffer_after"] == 620.0
|
|
assert body["total_value_after"] == 1720.0
|
|
assert body["source"] == "manual"
|
|
|
|
|
|
def test_dividend_requires_executed_at_and_rejects_fee_field() -> None:
|
|
repo = _FakeRepo(_portfolio())
|
|
client = _make_client(repo)
|
|
|
|
missing_date = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "DIVIDEND", "amount": 50},
|
|
headers=_auth(),
|
|
)
|
|
assert missing_date.status_code == 422
|
|
|
|
invalid_fee = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "DIVIDEND", "amount": 50, "executed_at": "2026-09-01", "fee": 0},
|
|
headers=_auth(),
|
|
)
|
|
assert invalid_fee.status_code == 422
|
|
|
|
|
|
def test_dividend_rejects_sub_minimum_numeric_precision() -> None:
|
|
repo = _FakeRepo(_portfolio())
|
|
client = _make_client(repo)
|
|
|
|
too_precise = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "DIVIDEND", "amount": "0.000000004", "executed_at": "2026-09-01"},
|
|
headers=_auth(),
|
|
)
|
|
assert too_precise.status_code == 422
|
|
|
|
|
|
def test_dividend_rejects_amount_exceeding_numeric_20_8() -> None:
|
|
repo = _FakeRepo(_portfolio())
|
|
client = _make_client(repo)
|
|
|
|
too_large = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "DIVIDEND", "amount": "1000000000000.00000000", "executed_at": "2026-09-01"},
|
|
headers=_auth(),
|
|
)
|
|
assert too_large.status_code == 422
|
|
|
|
|
|
def test_dividend_duplicate_requires_confirmation() -> None:
|
|
repo = _FakeRepo(_portfolio())
|
|
client = _make_client(repo)
|
|
|
|
first = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "DIVIDEND", "amount": 50, "executed_at": "2026-09-01", "note": "Q3"},
|
|
headers=_auth(),
|
|
)
|
|
assert first.status_code == 201
|
|
|
|
duplicate = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "DIVIDEND", "amount": 50, "executed_at": "2026-09-01", "note": "Q3"},
|
|
headers=_auth(),
|
|
)
|
|
assert duplicate.status_code == 409
|
|
assert duplicate.json()["detail"]["code"] == "POTENTIAL_DUPLICATE_DIVIDEND"
|
|
assert duplicate.json()["detail"]["matched_ledger_entry_id"] == first.json()["ledger_entry_id"]
|
|
|
|
confirmed = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={
|
|
"action": "DIVIDEND",
|
|
"amount": 50,
|
|
"executed_at": "2026-09-01",
|
|
"note": "Q3",
|
|
"confirm_duplicate": True,
|
|
},
|
|
headers=_auth(),
|
|
)
|
|
assert confirmed.status_code == 201
|
|
|
|
|
|
def test_confirm_duplicate_false_is_accepted_for_non_dividend_actions() -> None:
|
|
repo = _FakeRepo(_portfolio())
|
|
client = _make_client(repo)
|
|
|
|
invest = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "INVEST", "amount": 100, "confirm_duplicate": False},
|
|
headers=_auth(),
|
|
)
|
|
assert invest.status_code == 201
|
|
|
|
buy = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "BUY", "quantity": 1, "price": 10, "confirm_duplicate": False},
|
|
headers=_auth(),
|
|
)
|
|
assert buy.status_code == 201
|
|
|
|
|
|
def test_confirm_duplicate_true_is_rejected_for_non_dividend_actions() -> None:
|
|
repo = _FakeRepo(_portfolio())
|
|
client = _make_client(repo)
|
|
|
|
invest = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "INVEST", "amount": 100, "confirm_duplicate": True},
|
|
headers=_auth(),
|
|
)
|
|
assert invest.status_code == 422
|
|
|
|
|
|
def test_dividend_reversal_decreases_profit_buffer() -> None:
|
|
repo = _FakeRepo(_portfolio(buffer=200))
|
|
client = _make_client(repo)
|
|
|
|
dividend = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "DIVIDEND", "amount": 80, "executed_at": "2026-09-01", "note": "Q3"},
|
|
headers=_auth(),
|
|
)
|
|
assert dividend.status_code == 201
|
|
assert dividend.json()["profit_buffer_after"] == 280.0
|
|
|
|
reversal = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "REVERSAL", "target_ledger_entry_id": dividend.json()["ledger_entry_id"]},
|
|
headers=_auth(),
|
|
)
|
|
assert reversal.status_code == 201
|
|
assert reversal.json()["profit_buffer_after"] == 200.0
|
|
|
|
|
|
def test_unified_ledger_includes_dividend_fields() -> None:
|
|
repo = _FakeRepo(_portfolio())
|
|
client = _make_client(repo)
|
|
|
|
created = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual",
|
|
json={"action": "DIVIDEND", "amount": 33, "tax": 3, "executed_at": "2026-09-01", "note": "Q3"},
|
|
headers=_auth(),
|
|
)
|
|
assert created.status_code == 201
|
|
|
|
resp = client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger?source=manual", headers=_auth())
|
|
assert resp.status_code == 200
|
|
item = resp.json()["items"][0]
|
|
assert item["source"] == "manual"
|
|
assert item["action"] == "DIVIDEND"
|
|
assert item["amount"] == 33.0
|
|
assert item["price"] == 33.0
|
|
assert item["executed_at"] == "2026-09-01T00:00:00Z"
|
|
assert item["tax_amount"] == 3.0
|
|
assert item["note"] == "Q3"
|
|
assert item["profit_buffer_after"] == created.json()["profit_buffer_after"]
|
|
|
|
|
|
def test_unified_ledger_openapi_contains_nullable_amount() -> None:
|
|
repo = _FakeRepo(_portfolio())
|
|
client = _make_client(repo)
|
|
|
|
response = client.get("/openapi.json")
|
|
assert response.status_code == 200
|
|
schemas = response.json()["components"]["schemas"]
|
|
assert "amount" in schemas["LedgerHistoryItem"]["properties"]
|
|
|
|
|
|
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"
|
|
|
|
|
|
def test_unified_ledger_paginates_mixed_sources_and_hides_side() -> None:
|
|
repo = _FakeRepo()
|
|
client = _make_client(repo)
|
|
manual = client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "INVEST", "amount": 100}, headers=_auth()
|
|
)
|
|
assert manual.status_code == 201
|
|
repo.entries.append(
|
|
LedgerEntry(
|
|
ledger_entry_id=99, portfolio_id=_PORTFOLIO_ID, side="buy", quantity=2, price=10, source="broker",
|
|
recorded_at=datetime.now(UTC), cash_balance_after=900, asset_quantity_after=12,
|
|
)
|
|
)
|
|
first = client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger?limit=1", headers=_auth())
|
|
assert first.status_code == 200
|
|
first_item = first.json()["items"][0]
|
|
assert first_item["source"] == "broker"
|
|
assert first_item["action"] == "BUY"
|
|
assert "side" not in first_item
|
|
assert first.json()["next_cursor"]
|
|
second = client.get(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger?limit=1&cursor={first.json()['next_cursor']}", headers=_auth()
|
|
)
|
|
assert second.status_code == 200
|
|
assert second.json()["items"][0]["ledger_entry_id"] == manual.json()["ledger_entry_id"]
|
|
assert second.json()["next_cursor"] is None
|
|
|
|
|
|
def test_unified_ledger_rejects_tampered_or_different_source_cursor() -> None:
|
|
repo = _FakeRepo()
|
|
client = _make_client(repo)
|
|
for amount in (100, 200):
|
|
assert client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "INVEST", "amount": amount}, headers=_auth()
|
|
).status_code == 201
|
|
cursor = client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger?limit=1", headers=_auth()).json()["next_cursor"]
|
|
assert client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger?limit=1&cursor={cursor[:-1]}x", headers=_auth()).status_code == 422
|
|
assert client.get(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger?limit=1&source=manual&cursor={cursor}", headers=_auth()
|
|
).status_code == 422
|
|
|
|
|
|
def test_unified_ledger_source_filter_unknown_value_is_empty_and_owner_scoped() -> None:
|
|
repo = _FakeRepo()
|
|
client = _make_client(repo)
|
|
assert client.post(
|
|
f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "INVEST", "amount": 100}, headers=_auth()
|
|
).status_code == 201
|
|
filtered = client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger?source=unknown-producer", headers=_auth())
|
|
assert filtered.status_code == 200
|
|
assert filtered.json() == {"items": [], "next_cursor": None}
|
|
|
|
foreign_client = _make_client(_FakeRepo(_portfolio(owner=_OTHER_USER_ID)))
|
|
assert foreign_client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger", headers=_auth()).status_code == 403
|
|
|
|
service_client = _make_client(_FakeRepo(principal=_token_principal(token_type="service")))
|
|
assert service_client.get(f"/v1/portfolios/{_PORTFOLIO_ID}/ledger", headers=_auth()).status_code == 403
|