"""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" assert resp.json()["detail"]["message"] == "Insufficient cash balance" def test_manual_ledger_conflict_message_is_sanitized() -> None: repo = _FakeRepo(_portfolio()) client = _make_client(repo) secret = "SENTINEL_SECRET_TOKEN_12345" def _raise_secret(*args: Any, **kwargs: Any) -> LedgerEntry: raise ValueError(f"synthetic backend failure: {secret}") repo.record_manual_ledger = _raise_secret # type: ignore[method-assign] resp = client.post( f"/v1/portfolios/{_PORTFOLIO_ID}/ledger/manual", json={"action": "DIVIDEND", "amount": 1, "executed_at": "2026-09-01"}, headers=_auth(), ) assert resp.status_code == 422 detail = resp.json()["detail"] assert detail["code"] == "MANUAL_LEDGER_CONFLICT" assert detail["message"] == "Manual ledger request could not be processed" assert secret not in resp.text 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