genichinandClaude Opus 4.8
d93c0bcaf5
dcy-008(confirmed)/REQ-043(Approved)/BAT-018. trex-core dcy-034 FR-7 cold-start seed 의존. AC 11/11, QC PASS. worktree+PR 워크플로 첫 파일럿. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VHtM5Ws1HKCKQ4JGi6GRb5
538 lines
19 KiB
Python
538 lines
19 KiB
Python
"""Tests for GET /v1/portfolios/{portfolio_id}/holdings/cost-basis (REQ-043).
|
|
|
|
Uses TestClient with MagicMock repo injection — no DB required.
|
|
Covers AC-1~11 from REQ-043.
|
|
"""
|
|
|
|
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-holdings-001"
|
|
_DAEMON_ID = "daemon-001"
|
|
_OTHER_DAEMON_ID = "daemon-002"
|
|
_PATH = f"/v1/portfolios/{_PORTFOLIO_ID}/holdings/cost-basis"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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",
|
|
asset_quantity: float = 0.0,
|
|
managed_by_daemon_id: str | None = _DAEMON_ID,
|
|
) -> Portfolio:
|
|
return Portfolio(
|
|
portfolio_id=_PORTFOLIO_ID,
|
|
broker="kis",
|
|
status="enabled",
|
|
name="test-holdings-portfolio",
|
|
symbol=symbol,
|
|
asset_id="asset-001",
|
|
owner_user_id=owner,
|
|
cash_balance=1_000_000.0,
|
|
asset_quantity=asset_quantity,
|
|
tag_ids=[],
|
|
managed_by_daemon_id=managed_by_daemon_id,
|
|
)
|
|
|
|
|
|
def _make_entry(
|
|
*,
|
|
ledger_entry_id: int,
|
|
side: str,
|
|
quantity: float,
|
|
price: float,
|
|
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=0.0,
|
|
tax_amount=0.0,
|
|
executed_at=executed_at,
|
|
recorded_at=executed_at,
|
|
)
|
|
|
|
|
|
def _make_service_repo(
|
|
*,
|
|
daemon_id: str | None = _DAEMON_ID,
|
|
portfolio: Portfolio | None = None,
|
|
trades: list[LedgerEntry] | None = None,
|
|
) -> MagicMock:
|
|
"""Repo mock with service token principal."""
|
|
repo = MagicMock()
|
|
principal = TokenPrincipal(
|
|
token_id="t-svc",
|
|
user_id=_USER_ID,
|
|
permission="read",
|
|
token_type="service",
|
|
runtime_id="rt-1",
|
|
daemon_id=daemon_id,
|
|
)
|
|
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 _make_user_repo(
|
|
*,
|
|
user_id: str = _USER_ID,
|
|
permission: str = "read",
|
|
portfolio: Portfolio | None = None,
|
|
trades: list[LedgerEntry] | None = None,
|
|
) -> MagicMock:
|
|
"""Repo mock with user (DB) token principal."""
|
|
repo = MagicMock()
|
|
principal = TokenPrincipal(
|
|
token_id="t-usr",
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC-1: service token + buy 100@10,000 + buy 100@12,000 → WAC=11,000, qty=200
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAC1WacBuysOnly:
|
|
"""AC-1: buy-only portfolio → WAC = weighted average, no warning."""
|
|
|
|
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),
|
|
),
|
|
]
|
|
|
|
def _make_portfolio_with_qty(self) -> Portfolio:
|
|
return _make_portfolio(asset_quantity=200.0)
|
|
|
|
def test_status_200(self) -> None:
|
|
settings = _make_settings()
|
|
repo = _make_service_repo(
|
|
portfolio=self._make_portfolio_with_qty(),
|
|
trades=self._make_trades(),
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.status_code == 200
|
|
|
|
def test_current_quantity_200(self) -> None:
|
|
settings = _make_settings()
|
|
repo = _make_service_repo(
|
|
portfolio=self._make_portfolio_with_qty(),
|
|
trades=self._make_trades(),
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.json()["current_quantity"] == pytest.approx(200.0)
|
|
|
|
def test_cost_basis_per_unit_is_11000(self) -> None:
|
|
settings = _make_settings()
|
|
repo = _make_service_repo(
|
|
portfolio=self._make_portfolio_with_qty(),
|
|
trades=self._make_trades(),
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.json()["cost_basis_per_unit"] == pytest.approx(11_000.0)
|
|
|
|
def test_total_cost_basis_is_2_200_000(self) -> None:
|
|
settings = _make_settings()
|
|
repo = _make_service_repo(
|
|
portfolio=self._make_portfolio_with_qty(),
|
|
trades=self._make_trades(),
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.json()["total_cost_basis"] == pytest.approx(2_200_000.0)
|
|
|
|
def test_data_completeness_warning_is_null(self) -> None:
|
|
settings = _make_settings()
|
|
repo = _make_service_repo(
|
|
portfolio=self._make_portfolio_with_qty(),
|
|
trades=self._make_trades(),
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.json()["data_completeness_warning"] is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC-2: buy 200@10,000 + sell 150 → remaining qty=50, WAC unchanged=10,000
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAC2PartialSellWacUnchanged:
|
|
"""AC-2: partial sell → WAC unchanged, quantity reduced."""
|
|
|
|
def _make_trades(self) -> list[LedgerEntry]:
|
|
return [
|
|
_make_entry(
|
|
ledger_entry_id=1, side="buy", quantity=200, price=10_000,
|
|
executed_at=datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc),
|
|
),
|
|
_make_entry(
|
|
ledger_entry_id=2, side="sell", quantity=150, price=12_000,
|
|
executed_at=datetime(2026, 6, 10, 9, 0, tzinfo=timezone.utc),
|
|
),
|
|
]
|
|
|
|
def _make_portfolio_with_qty(self) -> Portfolio:
|
|
return _make_portfolio(asset_quantity=50.0)
|
|
|
|
def test_current_quantity_50(self) -> None:
|
|
settings = _make_settings()
|
|
repo = _make_service_repo(
|
|
portfolio=self._make_portfolio_with_qty(),
|
|
trades=self._make_trades(),
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["current_quantity"] == pytest.approx(50.0)
|
|
|
|
def test_wac_unchanged_at_10000(self) -> None:
|
|
settings = _make_settings()
|
|
repo = _make_service_repo(
|
|
portfolio=self._make_portfolio_with_qty(),
|
|
trades=self._make_trades(),
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.json()["cost_basis_per_unit"] == pytest.approx(10_000.0)
|
|
|
|
def test_total_cost_basis_is_500_000(self) -> None:
|
|
settings = _make_settings()
|
|
repo = _make_service_repo(
|
|
portfolio=self._make_portfolio_with_qty(),
|
|
trades=self._make_trades(),
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.json()["total_cost_basis"] == pytest.approx(500_000.0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC-3: user read_token (owner) → 200 + normal response
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_ac3_user_owner_read_token_returns_200() -> None:
|
|
"""AC-3: portfolio owner with user read token → 200."""
|
|
settings = _make_settings()
|
|
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),
|
|
),
|
|
]
|
|
repo = _make_user_repo(
|
|
user_id=_USER_ID,
|
|
portfolio=_make_portfolio(owner=_USER_ID, asset_quantity=100.0),
|
|
trades=trades,
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["current_quantity"] == pytest.approx(100.0)
|
|
assert body["cost_basis_per_unit"] == pytest.approx(10_000.0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC-4: user read_token non-owner → 403
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_ac4_user_non_owner_returns_403() -> None:
|
|
"""AC-4: non-owner user token → 403."""
|
|
settings = _make_settings()
|
|
repo = _make_user_repo(
|
|
user_id=_OTHER_USER_ID,
|
|
portfolio=_make_portfolio(owner=_USER_ID),
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC-5: service token daemon_id != managed_by_daemon_id → 403
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_ac5_daemon_mismatch_returns_403() -> None:
|
|
"""AC-5: service token daemon_id != portfolio.managed_by_daemon_id → 403."""
|
|
settings = _make_settings()
|
|
# Portfolio is assigned to DAEMON_ID, but token is for OTHER_DAEMON_ID
|
|
repo = _make_service_repo(
|
|
daemon_id=_OTHER_DAEMON_ID,
|
|
portfolio=_make_portfolio(managed_by_daemon_id=_DAEMON_ID),
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC-6: service token, portfolio.managed_by_daemon_id=null → 403
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_ac6_null_daemon_returns_403() -> None:
|
|
"""AC-6: managed_by_daemon_id=null → 403 (daemon 미배정)."""
|
|
settings = _make_settings()
|
|
repo = _make_service_repo(
|
|
daemon_id=_DAEMON_ID,
|
|
portfolio=_make_portfolio(managed_by_daemon_id=None),
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC-7: non-existent portfolio_id → 404
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_ac7_nonexistent_portfolio_returns_404() -> None:
|
|
"""AC-7: portfolio not found → 404."""
|
|
settings = _make_settings()
|
|
repo = _make_service_repo()
|
|
repo.get.side_effect = ValueError("포트폴리오를 찾을 수 없습니다")
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC-8: no buy/sell history (current_quantity=0) → 200 + null fields
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_ac8_empty_ledger_returns_200_null_fields() -> None:
|
|
"""AC-8: no history → 200 + current_quantity=0 + null cost basis fields."""
|
|
settings = _make_settings()
|
|
repo = _make_service_repo(
|
|
portfolio=_make_portfolio(asset_quantity=0.0),
|
|
trades=[],
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["current_quantity"] == pytest.approx(0.0)
|
|
assert body["cost_basis_per_unit"] is None
|
|
assert body["total_cost_basis"] is None
|
|
assert body["data_completeness_warning"] is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC-9: sell without prior buy → data_completeness_warning non-null
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_ac9_sell_without_buy_triggers_warning() -> None:
|
|
"""AC-9: sell before any buy → data_completeness_warning non-null."""
|
|
settings = _make_settings()
|
|
trades = [
|
|
_make_entry(
|
|
ledger_entry_id=1, side="sell", quantity=50, price=15_000,
|
|
executed_at=datetime(2026, 6, 5, 9, 0, tzinfo=timezone.utc),
|
|
),
|
|
]
|
|
repo = _make_service_repo(
|
|
portfolio=_make_portfolio(asset_quantity=0.0),
|
|
trades=trades,
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC-10: no auth token → 401
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_ac10_no_auth_token_returns_401() -> None:
|
|
"""AC-10: missing Authorization header → 401."""
|
|
settings = _make_settings()
|
|
repo = _make_service_repo()
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH)
|
|
assert resp.status_code == 401
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC-11: endpoint documented in OpenAPI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_ac11_endpoint_in_openapi() -> None:
|
|
"""AC-11: GET /v1/portfolios/{portfolio_id}/holdings/cost-basis is in OpenAPI spec."""
|
|
settings = _make_settings()
|
|
repo = _make_service_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", {})
|
|
endpoint_path = "/v1/portfolios/{portfolio_id}/holdings/cost-basis"
|
|
assert endpoint_path in paths, f"Expected {endpoint_path!r} in OpenAPI paths"
|
|
assert "get" in paths[endpoint_path]
|
|
# Verify response schema contains expected fields
|
|
get_op = paths[endpoint_path]["get"]
|
|
assert "responses" in get_op
|
|
assert "200" in get_op["responses"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Additional: response contains all required fields
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_response_contains_all_required_fields() -> None:
|
|
"""Response payload contains all REQ-043 specified fields."""
|
|
settings = _make_settings()
|
|
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),
|
|
),
|
|
]
|
|
repo = _make_service_repo(
|
|
portfolio=_make_portfolio(asset_quantity=100.0),
|
|
trades=trades,
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
for field in (
|
|
"portfolio_id",
|
|
"symbol",
|
|
"current_quantity",
|
|
"cost_basis_per_unit",
|
|
"total_cost_basis",
|
|
"cost_basis_from",
|
|
"data_completeness_warning",
|
|
):
|
|
assert field in body, f"Missing field in response: {field}"
|
|
|
|
|
|
def test_cost_basis_from_is_first_buy_date() -> None:
|
|
"""cost_basis_from reflects the earliest buy executed_at."""
|
|
settings = _make_settings()
|
|
first_buy = datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)
|
|
trades = [
|
|
_make_entry(
|
|
ledger_entry_id=1, side="buy", quantity=100, price=10_000,
|
|
executed_at=first_buy,
|
|
),
|
|
_make_entry(
|
|
ledger_entry_id=2, side="buy", quantity=50, price=12_000,
|
|
executed_at=datetime(2026, 6, 5, 9, 0, tzinfo=timezone.utc),
|
|
),
|
|
]
|
|
repo = _make_service_repo(
|
|
portfolio=_make_portfolio(asset_quantity=150.0),
|
|
trades=trades,
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
body = resp.json()
|
|
assert body["cost_basis_from"] is not None
|
|
|
|
|
|
def test_portfolio_id_in_response_matches_request() -> None:
|
|
"""portfolio_id in response equals the path parameter."""
|
|
settings = _make_settings()
|
|
repo = _make_service_repo(
|
|
portfolio=_make_portfolio(asset_quantity=0.0),
|
|
trades=[],
|
|
)
|
|
with _with_repo(repo):
|
|
client = _make_client(settings)
|
|
resp = client.get(_PATH, headers={"Authorization": "Bearer tok"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["portfolio_id"] == _PORTFOLIO_ID
|