genichin
6e02d92282
- Add acceptance criteria documentation (AC-0013 through AC-0023) - Add Line-0003 discovery documentation - Implement Strategy Catalog v1 with new database migration - Update portfolio API endpoints and schemas - Refactor strategy and portfolio services - Add strategy management CLI commands - Update domain models with strategy catalog support - Add comprehensive tests for v1 strategies - Update design documentation (data model, interfaces, runtime flows)
641 lines
23 KiB
Python
641 lines
23 KiB
Python
"""Tests for BAT-009: GET /v1/portfolios (REQ-028).
|
|
|
|
REQ-055 AC-1~10 케이스는 파일 하단에 추가됨.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from contextlib import ExitStack
|
|
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 Asset, Portfolio, TokenPrincipal
|
|
from app.main import create_app
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_ASSET_UUID = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
|
|
_DAEMON_ID = "daemon-001"
|
|
_USER_ID = "user-001"
|
|
_OTHER_USER_ID = "user-002"
|
|
|
|
|
|
def _make_settings(**kwargs) -> Settings:
|
|
return Settings(allowed_origins_raw="http://localhost:3000", **kwargs)
|
|
|
|
|
|
def _make_asset(markets: list[str] | None = None, display_name: str | None = "삼성전자") -> Asset:
|
|
return Asset(
|
|
asset_id=_ASSET_UUID,
|
|
symbol="005930",
|
|
markets=markets or ["KRX"],
|
|
display_name=display_name,
|
|
status="active",
|
|
)
|
|
|
|
|
|
def _make_portfolio(
|
|
portfolio_id: str,
|
|
owner: str = _USER_ID,
|
|
status: str = "enabled",
|
|
daemon_id: str | None = None,
|
|
display_name: str | None = "삼성전자",
|
|
active_strategy_id: str | None = None,
|
|
strategy_name: str | None = None,
|
|
strategy_params: dict | None = None,
|
|
strategy_revision: int | None = None,
|
|
markets: list[str] | None = None,
|
|
) -> Portfolio:
|
|
return Portfolio(
|
|
portfolio_id=portfolio_id,
|
|
broker="kiwoom",
|
|
status=status,
|
|
name=f"p-{portfolio_id[:4]}",
|
|
symbol="005930",
|
|
display_name=display_name,
|
|
asset_id=_ASSET_UUID,
|
|
owner_user_id=owner,
|
|
cash_balance=0.0,
|
|
asset_quantity=0.0,
|
|
price=0.0,
|
|
profit_buffer=0.0,
|
|
tag_ids=[],
|
|
managed_by_daemon_id=daemon_id,
|
|
active_strategy_id=active_strategy_id,
|
|
strategy_name=strategy_name,
|
|
strategy_params=strategy_params if strategy_params is not None else {},
|
|
strategy_revision=strategy_revision,
|
|
markets=markets if markets is not None else [],
|
|
)
|
|
|
|
|
|
class _FakeListRepo:
|
|
def __init__(
|
|
self,
|
|
portfolios: list[Portfolio] | None = None,
|
|
token_principal: TokenPrincipal | None = None,
|
|
) -> None:
|
|
self._portfolios: list[Portfolio] = portfolios or []
|
|
self._token_principal = token_principal
|
|
self.return_summaries: dict[str, dict[str, float]] = {}
|
|
self.summary_requests: list[list[str]] = []
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
def get_token_principal(self, token_hash: str) -> TokenPrincipal | None:
|
|
return self._token_principal
|
|
|
|
def list_portfolios(
|
|
self,
|
|
*,
|
|
owner_user_id: str | None = None,
|
|
managed_by_daemon_id: str | None = None,
|
|
market: str | None = None,
|
|
status: str | None = None,
|
|
**_kwargs: Any,
|
|
) -> list[Portfolio]:
|
|
results = self._portfolios
|
|
if owner_user_id is not None:
|
|
results = [p for p in results if p.owner_user_id == owner_user_id]
|
|
if managed_by_daemon_id is not None:
|
|
results = [p for p in results if p.managed_by_daemon_id == managed_by_daemon_id]
|
|
if status is not None:
|
|
results = [p for p in results if p.status == status]
|
|
return results
|
|
|
|
def get_return_summaries(self, portfolio_ids: list[str]) -> dict[str, dict[str, float]]:
|
|
self.summary_requests.append(list(portfolio_ids))
|
|
return {
|
|
portfolio_id: self.return_summaries[portfolio_id]
|
|
for portfolio_id in portfolio_ids
|
|
if portfolio_id in self.return_summaries
|
|
}
|
|
|
|
|
|
def _make_client(
|
|
admin_tokens: list[str] | None = None,
|
|
db_enabled: bool = False,
|
|
) -> TestClient:
|
|
get_settings.cache_clear()
|
|
settings = _make_settings(
|
|
admin_tokens_raw=",".join(admin_tokens or ["admin-tok"]),
|
|
trade_write_tokens_raw="user-tok",
|
|
db_enabled=db_enabled,
|
|
)
|
|
app = create_app(settings_factory=lambda: settings)
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
return TestClient(app)
|
|
|
|
|
|
def _with_repo(repo: _FakeListRepo, *, db_auth: bool = False) -> ExitStack:
|
|
"""Patch repo factory for route handlers and optionally for auth middleware."""
|
|
stack = ExitStack()
|
|
stack.enter_context(
|
|
patch("app.api.v1_portfolios.get_portfolio_repository_from_settings", return_value=repo)
|
|
)
|
|
if db_auth:
|
|
stack.enter_context(
|
|
patch("app.core.auth.get_portfolio_repository_from_settings", return_value=repo)
|
|
)
|
|
return stack
|
|
|
|
|
|
def _auth(token: str) -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-028: GET /v1/portfolios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_get_service_token_daemon_bound_returns_own_portfolios() -> None:
|
|
"""REQ-028 AC-1: service token(daemon-bound) → portfolios with managed_by_daemon_id=auth.daemon_id."""
|
|
p1 = _make_portfolio("p-001", daemon_id=_DAEMON_ID)
|
|
p2 = _make_portfolio("p-002", daemon_id="other-daemon")
|
|
repo = _FakeListRepo(
|
|
portfolios=[p1, p2],
|
|
token_principal=TokenPrincipal(
|
|
token_id="tid", user_id=_USER_ID, permission="trade-write",
|
|
token_type="service", daemon_id=_DAEMON_ID,
|
|
),
|
|
)
|
|
client = _make_client(db_enabled=True)
|
|
|
|
with _with_repo(repo, db_auth=True):
|
|
resp = client.get("/v1/portfolios", headers=_auth("svc-tok"))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 1
|
|
assert data["portfolios"][0]["portfolio_id"] == "p-001"
|
|
assert data["portfolios"][0]["symbol"] == "005930"
|
|
|
|
|
|
def test_get_service_token_no_daemon_returns_403() -> None:
|
|
"""REQ-028 AC-2: service token(no daemon_id) → HTTP 403."""
|
|
repo = _FakeListRepo(
|
|
token_principal=TokenPrincipal(
|
|
token_id="tid", user_id=_USER_ID, permission="trade-write",
|
|
token_type="service", daemon_id=None,
|
|
),
|
|
)
|
|
client = _make_client(db_enabled=True)
|
|
|
|
with _with_repo(repo, db_auth=True):
|
|
resp = client.get("/v1/portfolios", headers=_auth("svc-tok"))
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_get_user_token_returns_own_portfolios() -> None:
|
|
"""REQ-028 AC-3: session token + trade-write → own portfolios only."""
|
|
p1 = _make_portfolio("p-001", owner=_USER_ID)
|
|
p2 = _make_portfolio("p-002", owner=_OTHER_USER_ID)
|
|
repo = _FakeListRepo(
|
|
portfolios=[p1, p2],
|
|
token_principal=TokenPrincipal(
|
|
token_id="tid", user_id=_USER_ID, permission="trade-write",
|
|
token_type="session", daemon_id=None,
|
|
),
|
|
)
|
|
client = _make_client(db_enabled=True)
|
|
|
|
with _with_repo(repo, db_auth=True):
|
|
# Use a token not in env lists so it falls through to DB lookup
|
|
resp = client.get("/v1/portfolios", headers=_auth("session-db-tok"))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 1
|
|
assert data["portfolios"][0]["owner_user_id"] == _USER_ID
|
|
|
|
|
|
def test_get_admin_default_returns_own_portfolios() -> None:
|
|
"""REQ-028 AC-4: admin token (no ?user_id) → own portfolios (env token user_id=None → no filter → all)."""
|
|
p1 = _make_portfolio("p-001", owner=_USER_ID)
|
|
p2 = _make_portfolio("p-002", owner=_OTHER_USER_ID)
|
|
repo = _FakeListRepo(portfolios=[p1, p2])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 2
|
|
|
|
|
|
def test_get_admin_with_user_id_returns_target_portfolios() -> None:
|
|
"""REQ-028 AC-5: admin token + ?user_id=<target> → target user's portfolios."""
|
|
p1 = _make_portfolio("p-001", owner=_USER_ID)
|
|
p2 = _make_portfolio("p-002", owner=_OTHER_USER_ID)
|
|
repo = _FakeListRepo(portfolios=[p1, p2])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get(f"/v1/portfolios?user_id={_OTHER_USER_ID}", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 1
|
|
assert data["portfolios"][0]["owner_user_id"] == _OTHER_USER_ID
|
|
|
|
|
|
def test_get_excludes_disabled_by_default() -> None:
|
|
"""REQ-028 AC-7: no ?include_disabled → disabled portfolios excluded."""
|
|
active = _make_portfolio("p-001", status="enabled")
|
|
disabled = _make_portfolio("p-002", status="disabled")
|
|
repo = _FakeListRepo(portfolios=[active, disabled])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 1
|
|
assert data["portfolios"][0]["status"] == "enabled"
|
|
|
|
|
|
def test_get_include_disabled_returns_all() -> None:
|
|
"""REQ-028 AC-8: ?include_disabled=true → all portfolios returned."""
|
|
active = _make_portfolio("p-001", status="enabled")
|
|
disabled = _make_portfolio("p-002", status="disabled")
|
|
repo = _FakeListRepo(portfolios=[active, disabled])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios?include_disabled=true", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 2
|
|
|
|
|
|
def test_get_client_token_returns_403() -> None:
|
|
"""REQ-028 AC-9: client token → HTTP 403."""
|
|
repo = _FakeListRepo(
|
|
token_principal=TokenPrincipal(
|
|
token_id="tid", user_id=_USER_ID, permission="read",
|
|
token_type="client", daemon_id=None,
|
|
),
|
|
)
|
|
client = _make_client(db_enabled=True)
|
|
|
|
with _with_repo(repo, db_auth=True):
|
|
resp = client.get("/v1/portfolios", headers=_auth("client-tok"))
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_get_no_bearer_returns_401() -> None:
|
|
"""REQ-028 AC-10: no token → HTTP 401."""
|
|
client = _make_client()
|
|
resp = client.get("/v1/portfolios")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_get_response_has_asset_id_field() -> None:
|
|
"""REQ-028 AC-11: response portfolios reference assets.asset_id (UUID)."""
|
|
p = _make_portfolio("p-001")
|
|
repo = _FakeListRepo(portfolios=[p])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
item = resp.json()["portfolios"][0]
|
|
assert item["asset_id"] == _ASSET_UUID
|
|
assert item["symbol"] == "005930"
|
|
|
|
|
|
def test_get_list_includes_return_summary_fields_from_one_filtered_batch() -> None:
|
|
own = _make_portfolio("p-001", owner=_USER_ID)
|
|
other = _make_portfolio("p-002", owner=_OTHER_USER_ID)
|
|
repo = _FakeListRepo(portfolios=[own, other])
|
|
repo.return_summaries = {
|
|
"p-001": {
|
|
"total_invested": 500.0,
|
|
"total_withdrawn": 125.0,
|
|
"net_contributed_capital": 375.0,
|
|
"cumulative_profit_withdrawn": 40.0,
|
|
},
|
|
"p-002": {
|
|
"total_invested": 999.0,
|
|
"total_withdrawn": 0.0,
|
|
"net_contributed_capital": 999.0,
|
|
"cumulative_profit_withdrawn": 0.0,
|
|
},
|
|
}
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get(f"/v1/portfolios?user_id={_USER_ID}", headers=_auth("admin-tok"))
|
|
|
|
assert resp.status_code == 200
|
|
assert repo.summary_requests == [["p-001"]]
|
|
item = resp.json()["portfolios"][0]
|
|
assert item["net_contributed_capital"] == 375.0
|
|
assert item["cumulative_profit_withdrawn"] == 40.0
|
|
|
|
|
|
def test_get_list_defaults_missing_return_summary_to_zero() -> None:
|
|
repo = _FakeListRepo(portfolios=[_make_portfolio("p-001")])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
|
|
assert resp.status_code == 200
|
|
item = resp.json()["portfolios"][0]
|
|
assert item["net_contributed_capital"] == 0.0
|
|
assert item["cumulative_profit_withdrawn"] == 0.0
|
|
|
|
|
|
def test_get_empty_list_uses_empty_batch_without_n_plus_one() -> None:
|
|
repo = _FakeListRepo()
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"portfolios": [], "total": 0}
|
|
assert repo.summary_requests == [[]]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-055: GET /v1/portfolios 응답 전략·시장 메타데이터 additive 필드 (BAT-023)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_STRATEGY_ID = "strat-uuid-001"
|
|
|
|
|
|
def test_req055_ac1_bound_portfolio_returns_strategy_fields() -> None:
|
|
"""REQ-055 AC-1: active_strategy_id 바인딩된 포트폴리오 → strategy_name/strategy_params 정상 반환."""
|
|
p = _make_portfolio(
|
|
"p-001",
|
|
active_strategy_id=_STRATEGY_ID,
|
|
strategy_name="tvr_v1",
|
|
strategy_params={"target_weight": 0.5, "rebalance_band_pct": 0.05},
|
|
strategy_revision=7,
|
|
markets=["KRX"],
|
|
)
|
|
repo = _FakeListRepo(portfolios=[p])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
item = resp.json()["portfolios"][0]
|
|
assert item["active_strategy_id"] == _STRATEGY_ID
|
|
assert item["strategy_name"] == "tvr_v1"
|
|
assert item["strategy_params"] == {"target_weight": 0.5, "rebalance_band_pct": 0.05}
|
|
assert item["strategy_revision"] == 7
|
|
|
|
|
|
def test_req055_ac2_unbound_portfolio_returns_null_strategy() -> None:
|
|
"""REQ-055 AC-2: active_strategy_id=null → strategy_name=null, strategy_params={}."""
|
|
p = _make_portfolio("p-001", active_strategy_id=None, strategy_name=None, strategy_params={})
|
|
repo = _FakeListRepo(portfolios=[p])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
item = resp.json()["portfolios"][0]
|
|
assert item["active_strategy_id"] is None
|
|
assert item["strategy_name"] is None
|
|
assert item["strategy_params"] == {}
|
|
|
|
|
|
def test_req055_ac3_dangling_fk_returns_null_strategy_but_keeps_strategy_id() -> None:
|
|
"""REQ-055 AC-3: dangling FK → active_strategy_id 원래 값 유지, strategy_name=null, strategy_params={}."""
|
|
# dangling FK: active_strategy_id 있으나 LEFT JOIN 결과 ss.name=null, ss.config=null
|
|
# 이 시뮬레이션에서는 domain layer에서 strategy_name=None, strategy_params={}로 설정
|
|
p = _make_portfolio(
|
|
"p-001",
|
|
active_strategy_id=_STRATEGY_ID,
|
|
strategy_name=None, # dangling: JOIN 결과 없음
|
|
strategy_params={},
|
|
markets=[],
|
|
)
|
|
repo = _FakeListRepo(portfolios=[p])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
item = resp.json()["portfolios"][0]
|
|
assert item["active_strategy_id"] == _STRATEGY_ID # 원래 값 유지
|
|
assert item["strategy_name"] is None
|
|
assert item["strategy_params"] == {}
|
|
|
|
|
|
def test_req055_ac4_invalid_config_json_returns_empty_params_and_logs_warning(caplog) -> None:
|
|
"""REQ-055 AC-4: config 파싱 불가 → strategy_params={}, WARNING 로그 1건, HTTP 200."""
|
|
# 이 테스트는 postgres._row_to_portfolio 파싱 로직을 직접 검증한다.
|
|
# WARNING 로그는 app.repositories.postgres 로거에서 발생한다.
|
|
from app.repositories.postgres import PostgresPortfolioRepository
|
|
|
|
# row: index 0=portfolio_id, 1=broker, 2=status, 3=name, 4=symbol, 5=cash, 6=qty, 7=price,
|
|
# 8=profit_buffer, 9=active_strategy_id, 10=avabile_strategies, 11=owner_user_id,
|
|
# 12=created_at, 13=updated_at, 14=last_trade_at, 15=managed_by_daemon_id,
|
|
# 16=asset_id, 17=tag_ids, 18=display_name, 19=ss.name, 20=ss.config, 21=a.markets
|
|
row = (
|
|
"p-001", # [0] portfolio_id
|
|
"kiwoom", # [1] broker
|
|
"enabled", # [2] status
|
|
"test-portfolio", # [3] name
|
|
"005930", # [4] symbol
|
|
0.0, # [5] cash_balance
|
|
0.0, # [6] asset_quantity
|
|
0.0, # [7] asset_price
|
|
0.0, # [8] profit_buffer
|
|
_STRATEGY_ID, # [9] active_strategy_id
|
|
None, # [10] avabile_strategies
|
|
"user-001", # [11] owner_user_id
|
|
None, # [12] created_at
|
|
None, # [13] updated_at
|
|
None, # [14] last_trade_at
|
|
None, # [15] managed_by_daemon_id
|
|
_ASSET_UUID, # [16] asset_id
|
|
[], # [17] tag_ids
|
|
"삼성전자", # [18] display_name
|
|
"tvr_v1", # [19] ss.name
|
|
"{invalid json}", # [20] ss.config — 파싱 불가
|
|
["KRX"], # [21] a.markets
|
|
)
|
|
|
|
with caplog.at_level(logging.WARNING, logger="app.repositories.postgres"):
|
|
portfolio = PostgresPortfolioRepository._row_to_portfolio(row)
|
|
|
|
assert portfolio.strategy_params == {}
|
|
assert any("파싱 실패" in record.message for record in caplog.records)
|
|
warning_records = [r for r in caplog.records if r.levelno >= logging.WARNING]
|
|
assert len(warning_records) >= 1
|
|
|
|
|
|
def test_req055_ac5_markets_with_values() -> None:
|
|
"""REQ-055 AC-5: assets.markets=['KRX','NXT'] → markets=['KRX','NXT']."""
|
|
p = _make_portfolio("p-001", markets=["KRX", "NXT"])
|
|
repo = _FakeListRepo(portfolios=[p])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
item = resp.json()["portfolios"][0]
|
|
assert item["markets"] == ["KRX", "NXT"]
|
|
|
|
|
|
def test_req055_ac6_empty_markets() -> None:
|
|
"""REQ-055 AC-6: assets.markets=[] → markets=[]."""
|
|
p = _make_portfolio("p-001", markets=[])
|
|
repo = _FakeListRepo(portfolios=[p])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
item = resp.json()["portfolios"][0]
|
|
assert item["markets"] == []
|
|
|
|
|
|
def test_req055_ac7_existing_fields_regression() -> None:
|
|
"""REQ-055 AC-7: 기존 16필드 회귀 없음 — 값 변경 없이 반환."""
|
|
p = _make_portfolio(
|
|
"p-001",
|
|
owner=_USER_ID,
|
|
status="enabled",
|
|
display_name="삼성전자",
|
|
daemon_id=_DAEMON_ID,
|
|
)
|
|
repo = _FakeListRepo(portfolios=[p])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
item = resp.json()["portfolios"][0]
|
|
|
|
# 기존 16필드 검증
|
|
assert item["portfolio_id"] == "p-001"
|
|
assert item["asset_id"] == _ASSET_UUID
|
|
assert item["symbol"] == "005930"
|
|
assert item["display_name"] == "삼성전자"
|
|
assert item["owner_user_id"] == _USER_ID
|
|
assert item["broker"] == "kiwoom"
|
|
assert item["status"] == "enabled"
|
|
assert item["name"] == "p-p-00"
|
|
assert item["cash_balance"] == 0.0
|
|
assert item["asset_quantity"] == 0.0
|
|
assert item["price"] == 0.0
|
|
assert item["profit_buffer"] == 0.0
|
|
assert item["tag_ids"] == []
|
|
assert item["managed_by_daemon_id"] == _DAEMON_ID
|
|
assert "created_at" in item
|
|
assert "updated_at" in item
|
|
|
|
|
|
def test_req055_ac8_token_branch_invariant_service_token() -> None:
|
|
"""REQ-055 AC-8: service token 분기 — REQ-028 토큰 동작 불변 + 신규 필드 포함."""
|
|
p = _make_portfolio(
|
|
"p-001",
|
|
daemon_id=_DAEMON_ID,
|
|
active_strategy_id=_STRATEGY_ID,
|
|
strategy_name="tvr_v1",
|
|
strategy_params={"k": "v"},
|
|
markets=["KRX"],
|
|
)
|
|
repo = _FakeListRepo(
|
|
portfolios=[p],
|
|
token_principal=TokenPrincipal(
|
|
token_id="tid", user_id=_USER_ID, permission="trade-write",
|
|
token_type="service", daemon_id=_DAEMON_ID,
|
|
),
|
|
)
|
|
client = _make_client(db_enabled=True)
|
|
|
|
with _with_repo(repo, db_auth=True):
|
|
resp = client.get("/v1/portfolios", headers=_auth("svc-tok"))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 1
|
|
item = data["portfolios"][0]
|
|
# 토큰 분기 동작 불변 확인
|
|
assert item["portfolio_id"] == "p-001"
|
|
# 신규 필드도 포함되어 있음
|
|
assert item["active_strategy_id"] == _STRATEGY_ID
|
|
assert item["strategy_name"] == "tvr_v1"
|
|
assert item["strategy_params"] == {"k": "v"}
|
|
assert item["markets"] == ["KRX"]
|
|
|
|
|
|
def test_req055_ac9_strategy_params_exact_key_values() -> None:
|
|
"""REQ-055 AC-9: strategy_params 응답 dict가 원본 JSON과 동일 키·값."""
|
|
expected_params = {
|
|
"target_weight": 0.5,
|
|
"rebalance_band_pct": 0.05,
|
|
"min_trade_krw": 10000,
|
|
}
|
|
p = _make_portfolio(
|
|
"p-001",
|
|
active_strategy_id=_STRATEGY_ID,
|
|
strategy_name="tvr_v1",
|
|
strategy_params=expected_params,
|
|
markets=["KRX"],
|
|
)
|
|
repo = _FakeListRepo(portfolios=[p])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
item = resp.json()["portfolios"][0]
|
|
assert item["strategy_params"] == expected_params
|
|
assert item["strategy_params"]["target_weight"] == 0.5
|
|
assert item["strategy_params"]["rebalance_band_pct"] == 0.05
|
|
assert item["strategy_params"]["min_trade_krw"] == 10000
|
|
|
|
|
|
def test_req055_ac10_mixed_list_individual_accuracy() -> None:
|
|
"""REQ-055 AC-10: 혼합 리스트 — 각 포트폴리오의 4 필드가 개별적으로 정확하게 반환."""
|
|
bound = _make_portfolio(
|
|
"p-001",
|
|
active_strategy_id=_STRATEGY_ID,
|
|
strategy_name="tvr_v1",
|
|
strategy_params={"target_weight": 0.5},
|
|
markets=["KRX", "NXT"],
|
|
)
|
|
unbound = _make_portfolio(
|
|
"p-002",
|
|
active_strategy_id=None,
|
|
strategy_name=None,
|
|
strategy_params={},
|
|
markets=["KRX"],
|
|
)
|
|
repo = _FakeListRepo(portfolios=[bound, unbound])
|
|
client = _make_client()
|
|
|
|
with _with_repo(repo):
|
|
resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 2
|
|
|
|
items = {item["portfolio_id"]: item for item in data["portfolios"]}
|
|
|
|
# 바인딩된 포트폴리오
|
|
p1 = items["p-001"]
|
|
assert p1["active_strategy_id"] == _STRATEGY_ID
|
|
assert p1["strategy_name"] == "tvr_v1"
|
|
assert p1["strategy_params"] == {"target_weight": 0.5}
|
|
assert p1["markets"] == ["KRX", "NXT"]
|
|
|
|
# 미바인딩 포트폴리오
|
|
p2 = items["p-002"]
|
|
assert p2["active_strategy_id"] is None
|
|
assert p2["strategy_name"] is None
|
|
assert p2["strategy_params"] == {}
|
|
assert p2["markets"] == ["KRX"]
|