753 lines
27 KiB
Python
753 lines
27 KiB
Python
"""Tests for BAT-001: asset catalog DB foundation (REQ-009, REQ-011, REQ-014).
|
|
|
|
DB-integration ACs (schema inspection, FK/RESTRICT constraints, Alembic migration
|
|
data integrity) require a live PostgreSQL database and are documented in
|
|
docs/batches/bat-001_20260519_asset-catalog-db-foundation/verification.md.
|
|
The tests here cover all domain-model, schema, and app-level ACs that run without DB.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from app.domain.models import Asset, MarketCode, Portfolio
|
|
from app.repositories.portfolio_repository import PortfolioRepository
|
|
from app.schemas.portfolio import CreatePortfolioRequest, UpdatePortfolioRequest
|
|
from app.schemas.v1 import AssetItem
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-009 AC-4: MarketCode rejects invalid codes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_market_code_rejects_unknown_code() -> None:
|
|
with pytest.raises(ValueError):
|
|
MarketCode("UNKNOWN")
|
|
|
|
|
|
def test_market_code_rejects_empty_string() -> None:
|
|
with pytest.raises(ValueError):
|
|
MarketCode("")
|
|
|
|
|
|
def test_market_code_all_valid_members() -> None:
|
|
expected = {"KRX", "NXT", "SOR", "NASDAQ", "NYSE", "UPBIT", "BITHUMB"}
|
|
actual = {m.value for m in MarketCode}
|
|
assert actual == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-009 AC-5: MarketCode normalizes lowercase input
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_market_code_accepts_uppercase_krx() -> None:
|
|
assert MarketCode("KRX") == MarketCode.KRX
|
|
|
|
|
|
def test_market_code_normalizes_lowercase_krx() -> None:
|
|
assert MarketCode("krx".upper()) == MarketCode.KRX
|
|
|
|
|
|
def test_market_code_normalizes_lowercase_nasdaq() -> None:
|
|
assert MarketCode("nasdaq".upper()) == MarketCode.NASDAQ
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-009: Asset domain model exists with correct fields
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_asset_model_has_required_fields() -> None:
|
|
now = datetime.now(timezone.utc)
|
|
asset = Asset(
|
|
asset_id="00000000-0000-0000-0000-000000000001",
|
|
symbol="005930",
|
|
markets=["KRX"],
|
|
display_name="Samsung Electronics",
|
|
status="active",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
assert asset.asset_id == "00000000-0000-0000-0000-000000000001"
|
|
assert asset.symbol == "005930"
|
|
assert asset.markets == ["KRX"]
|
|
assert asset.display_name == "Samsung Electronics"
|
|
assert asset.status == "active"
|
|
|
|
|
|
def test_asset_model_markets_defaults_to_empty_list() -> None:
|
|
asset = Asset(asset_id="a1", symbol="005930", markets=[])
|
|
assert asset.markets == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-009: PortfolioRepository Protocol declares asset CRUD methods
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_portfolio_repository_protocol_has_asset_crud() -> None:
|
|
required = {"get_asset_by_id", "get_asset_by_symbol", "list_assets", "insert_asset", "update_asset"}
|
|
declared = set(dir(PortfolioRepository))
|
|
for method in required:
|
|
assert method in declared, f"PortfolioRepository missing method: {method}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-011 AC-4: Schemas have no symbol/market fields — API response clean
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_create_portfolio_request_requires_asset_id_not_symbol() -> None:
|
|
fields = CreatePortfolioRequest.model_fields
|
|
assert "asset_id" in fields, "CreatePortfolioRequest must have asset_id"
|
|
assert "symbol" not in fields, "CreatePortfolioRequest must not have symbol"
|
|
|
|
|
|
def test_update_portfolio_request_has_no_symbol_field() -> None:
|
|
fields = UpdatePortfolioRequest.model_fields
|
|
assert "symbol" not in fields, "UpdatePortfolioRequest must not have symbol"
|
|
|
|
|
|
def test_asset_item_schema_has_no_market_field() -> None:
|
|
fields = AssetItem.model_fields
|
|
assert "market" not in fields, "AssetItem must not have market (use assets.markets)"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-011 AC-4: Portfolio domain model has asset_id, not market
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_portfolio_model_has_asset_id_no_market() -> None:
|
|
import dataclasses
|
|
field_names = {f.name for f in dataclasses.fields(Portfolio)}
|
|
assert "asset_id" in field_names, "Portfolio must have asset_id"
|
|
assert "market" not in field_names, "Portfolio must not have market"
|
|
|
|
|
|
def test_portfolio_model_retains_symbol_for_display() -> None:
|
|
import dataclasses
|
|
field_names = {f.name for f in dataclasses.fields(Portfolio)}
|
|
assert "symbol" in field_names, "Portfolio retains symbol (populated via JOIN)"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-011 AC-5: Same user + same asset_id can have two distinct portfolios
|
|
# (no UNIQUE constraint on owner_user_id + asset_id)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class _MinimalRepo:
|
|
"""Minimal in-memory repo to test duplicate portfolio creation."""
|
|
|
|
def __init__(self) -> None:
|
|
self._portfolios: list[Portfolio] = []
|
|
self._seq = 0
|
|
|
|
def create_portfolio(
|
|
self,
|
|
portfolio_id: str,
|
|
*,
|
|
name: str,
|
|
owner_user_id: str,
|
|
broker: str,
|
|
status: str,
|
|
asset_id: str,
|
|
cash_balance: float = 0.0,
|
|
asset_quantity: float = 0.0,
|
|
asset_price: float = 0.0,
|
|
profit_buffer: float = 0.0,
|
|
active_strategy_id: str | None = None,
|
|
avabile_strategies: list[str] | None = None,
|
|
) -> Portfolio:
|
|
if any(p.owner_user_id == owner_user_id and p.name == name for p in self._portfolios):
|
|
raise ValueError(f"이미 사용 중인 이름입니다: {name}")
|
|
p = Portfolio(
|
|
portfolio_id=portfolio_id,
|
|
broker=broker,
|
|
status=status,
|
|
name=name,
|
|
symbol=asset_id,
|
|
asset_id=asset_id,
|
|
owner_user_id=owner_user_id,
|
|
cash_balance=cash_balance,
|
|
asset_quantity=asset_quantity,
|
|
price=asset_price,
|
|
profit_buffer=profit_buffer,
|
|
active_strategy_id=active_strategy_id,
|
|
avabile_strategies=avabile_strategies,
|
|
)
|
|
self._portfolios.append(p)
|
|
return p
|
|
|
|
|
|
def test_same_user_can_create_two_portfolios_for_same_asset_id() -> None:
|
|
_TEST_ASSET_ID = "00000000-0000-0000-0000-000000000001"
|
|
repo = _MinimalRepo()
|
|
|
|
repo.create_portfolio("p1", name="Portfolio_A", owner_user_id="user-1", broker="kis",
|
|
status="enabled", asset_id=_TEST_ASSET_ID)
|
|
repo.create_portfolio("p2", name="Portfolio_B", owner_user_id="user-1", broker="kis",
|
|
status="enabled", asset_id=_TEST_ASSET_ID)
|
|
|
|
assert len(repo._portfolios) == 2
|
|
ids = {p.portfolio_id for p in repo._portfolios}
|
|
assert ids == {"p1", "p2"}
|
|
for p in repo._portfolios:
|
|
assert p.asset_id == _TEST_ASSET_ID
|
|
assert p.owner_user_id == "user-1"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-014 AC documentation: migration script existence and structure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_migration_rev_a_file_exists() -> None:
|
|
import os
|
|
migrations_dir = os.path.join(os.path.dirname(__file__), "..", "migrations", "versions")
|
|
rev_a_files = [f for f in os.listdir(migrations_dir) if "0017" in f and "assets" in f]
|
|
assert len(rev_a_files) == 1, f"rev-A migration (0017 assets) not found in {migrations_dir}"
|
|
|
|
|
|
def test_migration_rev_b_file_exists() -> None:
|
|
import os
|
|
migrations_dir = os.path.join(os.path.dirname(__file__), "..", "migrations", "versions")
|
|
rev_b_files = [f for f in os.listdir(migrations_dir) if "0018" in f]
|
|
assert len(rev_b_files) == 1, f"rev-B migration (0018) not found in {migrations_dir}"
|
|
|
|
|
|
def test_migration_rev_a_creates_assets_table() -> None:
|
|
import os
|
|
migrations_dir = os.path.join(os.path.dirname(__file__), "..", "migrations", "versions")
|
|
rev_a = next(f for f in os.listdir(migrations_dir) if "0017" in f)
|
|
content = open(os.path.join(migrations_dir, rev_a)).read()
|
|
assert "create_table" in content
|
|
assert '"assets"' in content
|
|
assert "asset_id" in content
|
|
assert "symbol" in content
|
|
assert "markets" in content
|
|
assert "idx_assets_markets_gin" in content
|
|
|
|
|
|
def test_migration_rev_b_drops_symbol_and_market_columns() -> None:
|
|
import os
|
|
migrations_dir = os.path.join(os.path.dirname(__file__), "..", "migrations", "versions")
|
|
rev_b = next(f for f in os.listdir(migrations_dir) if "0018" in f)
|
|
content = open(os.path.join(migrations_dir, rev_b)).read()
|
|
assert "drop_column" in content
|
|
assert '"symbol"' in content or "'symbol'" in content
|
|
assert "asset_id" in content
|
|
assert "fk_portfolios_asset_id" in content
|
|
|
|
|
|
def test_migration_rev_b_has_data_migration_insert() -> None:
|
|
import os
|
|
migrations_dir = os.path.join(os.path.dirname(__file__), "..", "migrations", "versions")
|
|
rev_b = next(f for f in os.listdir(migrations_dir) if "0018" in f)
|
|
content = open(os.path.join(migrations_dir, rev_b)).read()
|
|
assert "INSERT INTO assets" in content
|
|
assert "ARRAY_AGG" in content
|
|
|
|
|
|
# ===========================================================================
|
|
# REQ-010: Admin asset catalog CRUD
|
|
# ===========================================================================
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-010 CLI: FakeRepo for asset CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_ASSET_UUID = "00000000-0000-0000-0000-000000000099"
|
|
|
|
|
|
class _AssetFakeRepo:
|
|
"""Minimal in-memory repo supporting asset CRUD for CLI tests."""
|
|
|
|
def __init__(self) -> None:
|
|
self._assets: dict[str, Asset] = {}
|
|
self._seq = 1
|
|
|
|
def insert_asset(
|
|
self,
|
|
*,
|
|
symbol: str,
|
|
markets: list[str],
|
|
display_name: str | None = None,
|
|
status: str = "active",
|
|
) -> Asset:
|
|
now = __import__("datetime").datetime.now(__import__("datetime").timezone.utc)
|
|
asset_id = f"00000000-0000-0000-0000-{self._seq:012d}"
|
|
self._seq += 1
|
|
asset = Asset(
|
|
asset_id=asset_id, symbol=symbol, markets=markets,
|
|
display_name=display_name, status=status,
|
|
created_at=now, updated_at=now,
|
|
)
|
|
self._assets[asset_id] = asset
|
|
return asset
|
|
|
|
def update_asset(
|
|
self,
|
|
asset_id: str,
|
|
*,
|
|
markets: list[str] | None = None,
|
|
display_name: str | None = None,
|
|
status: str | None = None,
|
|
) -> Asset:
|
|
asset = self._assets.get(asset_id)
|
|
if asset is None:
|
|
raise ValueError(f"asset not found: {asset_id}")
|
|
import dataclasses
|
|
fields = dataclasses.asdict(asset)
|
|
if markets is not None:
|
|
fields["markets"] = markets
|
|
if display_name is not None:
|
|
fields["display_name"] = display_name
|
|
if status is not None:
|
|
fields["status"] = status
|
|
updated = Asset(**fields)
|
|
self._assets[asset_id] = updated
|
|
return updated
|
|
|
|
def list_assets(self) -> list[Asset]:
|
|
return list(self._assets.values())
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
def ensure_schema(self) -> None:
|
|
pass
|
|
|
|
|
|
def _build_asset_runner(repo: _AssetFakeRepo):
|
|
from typer.testing import CliRunner
|
|
from app.cli.main import create_app
|
|
from app.core.config import Settings
|
|
|
|
settings = Settings(allowed_origins_raw="http://localhost:3000")
|
|
app = create_app(
|
|
settings_factory=lambda: settings,
|
|
repository_factory=lambda _: repo,
|
|
)
|
|
return CliRunner(), app
|
|
|
|
|
|
# REQ-010 AC-1: pfhub asset create
|
|
def test_asset_cli_create_success() -> None:
|
|
repo = _AssetFakeRepo()
|
|
runner, app = _build_asset_runner(repo)
|
|
result = runner.invoke(app, ["asset", "create", "005930", "--market", "KRX", "--force"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "005930" in result.output
|
|
|
|
|
|
# REQ-010 AC-2: pfhub asset list — active/inactive filter
|
|
def test_asset_cli_list_returns_all() -> None:
|
|
repo = _AssetFakeRepo()
|
|
repo.insert_asset(symbol="AAPL", markets=["NASDAQ"], status="active")
|
|
repo.insert_asset(symbol="BTC", markets=["UPBIT"], status="inactive")
|
|
runner, app = _build_asset_runner(repo)
|
|
result = runner.invoke(app, ["asset", "list"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "AAPL" in result.output
|
|
assert "BTC" in result.output
|
|
|
|
|
|
def test_asset_cli_list_active_only_filter() -> None:
|
|
repo = _AssetFakeRepo()
|
|
repo.insert_asset(symbol="AAPL", markets=["NASDAQ"], status="active")
|
|
repo.insert_asset(symbol="BTC", markets=["UPBIT"], status="inactive")
|
|
runner, app = _build_asset_runner(repo)
|
|
result = runner.invoke(app, ["asset", "list", "--active-only"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "AAPL" in result.output
|
|
assert "BTC" not in result.output
|
|
|
|
|
|
# REQ-010 AC-3: pfhub asset update
|
|
def test_asset_cli_update_success() -> None:
|
|
repo = _AssetFakeRepo()
|
|
asset = repo.insert_asset(symbol="005930", markets=["KRX"])
|
|
runner, app = _build_asset_runner(repo)
|
|
result = runner.invoke(app, [
|
|
"asset", "update", asset.asset_id,
|
|
"--display-name", "Samsung Electronics",
|
|
"--force",
|
|
])
|
|
assert result.exit_code == 0, result.output
|
|
assert "005930" in result.output
|
|
|
|
|
|
# REQ-010 AC-4: pfhub asset deactivate
|
|
def test_asset_cli_deactivate_success() -> None:
|
|
repo = _AssetFakeRepo()
|
|
asset = repo.insert_asset(symbol="005930", markets=["KRX"])
|
|
runner, app = _build_asset_runner(repo)
|
|
result = runner.invoke(app, ["asset", "deactivate", asset.asset_id, "--force"])
|
|
assert result.exit_code == 0, result.output
|
|
assert repo._assets[asset.asset_id].status == "inactive"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-010 HTTP: admin catalog endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_catalog_http_client(monkeypatch=None):
|
|
"""Build TestClient with mock admin-auth for catalog endpoint tests."""
|
|
from unittest.mock import MagicMock, patch
|
|
from fastapi.testclient import TestClient
|
|
from app.main import create_app
|
|
from app.core.config import Settings, get_settings
|
|
from app.domain.models import TokenPrincipal
|
|
|
|
_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
|
|
settings = Settings(
|
|
allowed_origins_raw="http://localhost:3000",
|
|
db_enabled=True,
|
|
credential_encryption_key=_ENC_KEY,
|
|
)
|
|
admin_principal = TokenPrincipal(
|
|
token_id="t-admin", user_id="admin-1", permission="admin",
|
|
token_type="admin", runtime_id="rt-1",
|
|
)
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = admin_principal
|
|
return mock_repo, settings
|
|
|
|
|
|
def _make_asset_obj(asset_id: str = _ASSET_UUID, symbol: str = "005930") -> Asset:
|
|
from datetime import datetime, timezone
|
|
return Asset(
|
|
asset_id=asset_id,
|
|
symbol=symbol,
|
|
markets=["KRX"],
|
|
display_name="Samsung",
|
|
status="active",
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
|
|
|
|
def test_asset_http_create_returns_201() -> None:
|
|
from unittest.mock import MagicMock, patch
|
|
from fastapi.testclient import TestClient
|
|
from app.main import create_app
|
|
from app.core.config import Settings, get_settings
|
|
from app.domain.models import TokenPrincipal
|
|
|
|
_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
|
|
settings = Settings(
|
|
allowed_origins_raw="http://localhost:3000",
|
|
db_enabled=True,
|
|
credential_encryption_key=_ENC_KEY,
|
|
)
|
|
admin_principal = TokenPrincipal(
|
|
token_id="t-admin", user_id="admin-1", permission="admin",
|
|
token_type="admin", runtime_id="rt-1",
|
|
)
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = admin_principal
|
|
mock_repo.insert_asset.return_value = _make_asset_obj()
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_assets.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
app = create_app(settings_factory=lambda: settings)
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
client = TestClient(app)
|
|
resp = client.post(
|
|
"/v1/assets",
|
|
json={"symbol": "005930", "markets": ["KRX"]},
|
|
headers={"Authorization": "Bearer admin-token"},
|
|
)
|
|
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
assert body["symbol"] == "005930"
|
|
assert body["status"] == "active"
|
|
assert "asset_id" in body
|
|
|
|
|
|
def test_asset_http_update_returns_200() -> None:
|
|
from unittest.mock import MagicMock, patch
|
|
from fastapi.testclient import TestClient
|
|
from app.main import create_app
|
|
from app.core.config import Settings, get_settings
|
|
from app.domain.models import TokenPrincipal
|
|
|
|
_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
|
|
settings = Settings(
|
|
allowed_origins_raw="http://localhost:3000",
|
|
db_enabled=True,
|
|
credential_encryption_key=_ENC_KEY,
|
|
)
|
|
admin_principal = TokenPrincipal(
|
|
token_id="t-admin", user_id="admin-1", permission="admin",
|
|
token_type="admin", runtime_id="rt-1",
|
|
)
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = admin_principal
|
|
mock_repo.update_asset.return_value = _make_asset_obj()
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_assets.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
app = create_app(settings_factory=lambda: settings)
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
client = TestClient(app)
|
|
resp = client.patch(
|
|
f"/v1/assets/{_ASSET_UUID}",
|
|
json={"display_name": "Samsung Electronics"},
|
|
headers={"Authorization": "Bearer admin-token"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["asset_id"] == _ASSET_UUID
|
|
|
|
|
|
def test_asset_http_deactivate_returns_200() -> None:
|
|
from unittest.mock import MagicMock, patch
|
|
from fastapi.testclient import TestClient
|
|
from app.main import create_app
|
|
from app.core.config import Settings, get_settings
|
|
from app.domain.models import TokenPrincipal
|
|
|
|
_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
|
|
settings = Settings(
|
|
allowed_origins_raw="http://localhost:3000",
|
|
db_enabled=True,
|
|
credential_encryption_key=_ENC_KEY,
|
|
)
|
|
admin_principal = TokenPrincipal(
|
|
token_id="t-admin", user_id="admin-1", permission="admin",
|
|
token_type="admin", runtime_id="rt-1",
|
|
)
|
|
inactive_asset = _make_asset_obj()
|
|
import dataclasses
|
|
inactive_asset = dataclasses.replace(inactive_asset, status="inactive")
|
|
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = admin_principal
|
|
mock_repo.update_asset.return_value = inactive_asset
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_assets.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
app = create_app(settings_factory=lambda: settings)
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
client = TestClient(app)
|
|
resp = client.patch(
|
|
f"/v1/assets/{_ASSET_UUID}/deactivate",
|
|
headers={"Authorization": "Bearer admin-token"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "inactive"
|
|
|
|
|
|
def test_asset_http_non_admin_403() -> None:
|
|
from unittest.mock import MagicMock, patch
|
|
from fastapi.testclient import TestClient
|
|
from app.main import create_app
|
|
from app.core.config import Settings, get_settings
|
|
from app.domain.models import TokenPrincipal
|
|
|
|
_ENC_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
|
|
settings = Settings(
|
|
allowed_origins_raw="http://localhost:3000",
|
|
db_enabled=True,
|
|
credential_encryption_key=_ENC_KEY,
|
|
)
|
|
user_principal = TokenPrincipal(
|
|
token_id="t-user", user_id="user-1", permission="read",
|
|
token_type="user", runtime_id="rt-1",
|
|
)
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = user_principal
|
|
|
|
with patch("app.core.auth.get_portfolio_repository_from_settings", return_value=mock_repo), \
|
|
patch("app.api.v1_assets.get_portfolio_repository_from_settings", return_value=mock_repo):
|
|
app = create_app(settings_factory=lambda: settings)
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
client = TestClient(app)
|
|
resp = client.post(
|
|
"/v1/assets",
|
|
json={"symbol": "005930", "markets": ["KRX"]},
|
|
headers={"Authorization": "Bearer user-token"},
|
|
)
|
|
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ===========================================================================
|
|
# REQ-012: Portfolio create catalog validation
|
|
# ===========================================================================
|
|
|
|
class _PortfolioServiceFakeRepo:
|
|
"""FakeRepo for PortfolioService catalog validation tests."""
|
|
|
|
def __init__(self, asset: Asset | None = None) -> None:
|
|
self._asset = asset
|
|
self._strategy_sets: dict = {}
|
|
self._portfolios: dict = {}
|
|
|
|
def get_asset_by_id(self, asset_id: str) -> Asset | None:
|
|
if self._asset and self._asset.asset_id == asset_id:
|
|
return self._asset
|
|
return None
|
|
|
|
def get_asset_by_symbol(self, symbol: str) -> Asset | None:
|
|
if self._asset and self._asset.symbol == symbol:
|
|
return self._asset
|
|
return None
|
|
|
|
def get_or_create_strategy_set(self, name: str, config: str):
|
|
from app.domain.models import StrategySet
|
|
key = f"{name}:{config}"
|
|
if key not in self._strategy_sets:
|
|
self._strategy_sets[key] = StrategySet(id=f"ss-{len(self._strategy_sets)}", name=name, config=config, nick=name)
|
|
return self._strategy_sets[key]
|
|
|
|
def create_portfolio(self, portfolio_id: str, *, name, owner_user_id, broker, status, asset_id, **kwargs):
|
|
from app.domain.models import Portfolio
|
|
portfolio = Portfolio(
|
|
portfolio_id=portfolio_id, broker=broker, status=status, name=name,
|
|
symbol=asset_id, asset_id=asset_id, owner_user_id=owner_user_id,
|
|
cash_balance=kwargs.get("cash_balance", 0.0),
|
|
asset_quantity=kwargs.get("asset_quantity", 0.0),
|
|
price=kwargs.get("asset_price", 0.0),
|
|
profit_buffer=kwargs.get("profit_buffer", 0.0),
|
|
active_strategy_id=kwargs.get("active_strategy_id"),
|
|
)
|
|
self._portfolios[portfolio_id] = portfolio
|
|
return portfolio
|
|
|
|
def get(self, portfolio_id: str):
|
|
p = self._portfolios.get(portfolio_id)
|
|
if p is None:
|
|
raise ValueError(f"portfolio not found: {portfolio_id}")
|
|
return p
|
|
|
|
def get_strategy_set(self, id: str):
|
|
return None
|
|
|
|
def list_recent_trades(self, portfolio_id: str, *, limit: int = 5):
|
|
return []
|
|
|
|
def list_comments(self, portfolio_id: str):
|
|
return []
|
|
|
|
def list_strategy_sets(self):
|
|
return []
|
|
|
|
def list_portfolios(self, **kwargs):
|
|
return []
|
|
|
|
def list_ledger_entries(self, portfolio_id: str, **kwargs):
|
|
return []
|
|
|
|
def list_broker_credentials(self, **kwargs):
|
|
return []
|
|
|
|
|
|
_CATALOG_ASSET_ID = "aaaaaaaa-0000-0000-0000-000000000001"
|
|
|
|
|
|
def _active_asset() -> Asset:
|
|
return Asset(asset_id=_CATALOG_ASSET_ID, symbol="005930", markets=["KRX"], status="active")
|
|
|
|
|
|
def _inactive_asset() -> Asset:
|
|
return Asset(asset_id=_CATALOG_ASSET_ID, symbol="005930", markets=["KRX"], status="inactive")
|
|
|
|
|
|
# REQ-012 AC-1: active asset → portfolio create succeeds
|
|
def test_portfolio_create_with_active_asset() -> None:
|
|
from app.services.portfolio_service import PortfolioService
|
|
from app.schemas.portfolio import CreatePortfolioRequest
|
|
|
|
repo = _PortfolioServiceFakeRepo(asset=_active_asset())
|
|
service = PortfolioService(repo)
|
|
result = service.create_portfolio(
|
|
CreatePortfolioRequest(
|
|
name="Test_Portfolio",
|
|
asset_id=_CATALOG_ASSET_ID,
|
|
broker="kis",
|
|
),
|
|
owner_user_id="user-1",
|
|
)
|
|
assert result is not None
|
|
|
|
|
|
# REQ-012 AC-2: unknown asset_id → "asset not found" error
|
|
def test_portfolio_create_unknown_asset() -> None:
|
|
from app.services.portfolio_service import PortfolioService
|
|
from app.schemas.portfolio import CreatePortfolioRequest
|
|
|
|
repo = _PortfolioServiceFakeRepo(asset=None)
|
|
service = PortfolioService(repo)
|
|
with pytest.raises(ValueError, match="asset not found"):
|
|
service.create_portfolio(
|
|
CreatePortfolioRequest(
|
|
name="Test_Portfolio",
|
|
asset_id="nonexistent-uuid",
|
|
broker="kis",
|
|
),
|
|
owner_user_id="user-1",
|
|
)
|
|
|
|
|
|
# REQ-012 AC-3: inactive asset → "asset is inactive" error
|
|
def test_portfolio_create_inactive_asset() -> None:
|
|
from app.services.portfolio_service import PortfolioService
|
|
from app.schemas.portfolio import CreatePortfolioRequest
|
|
|
|
repo = _PortfolioServiceFakeRepo(asset=_inactive_asset())
|
|
service = PortfolioService(repo)
|
|
with pytest.raises(ValueError, match="asset is inactive"):
|
|
service.create_portfolio(
|
|
CreatePortfolioRequest(
|
|
name="Test_Portfolio",
|
|
asset_id=_CATALOG_ASSET_ID,
|
|
broker="kis",
|
|
),
|
|
owner_user_id="user-1",
|
|
)
|
|
|
|
|
|
# REQ-012 AC-4: --symbol resolves asset_id and creates portfolio
|
|
def test_portfolio_create_by_symbol() -> None:
|
|
from typer.testing import CliRunner
|
|
from app.cli.main import create_app
|
|
from app.core.config import Settings
|
|
from app.domain.models import User, StrategySet
|
|
|
|
class _FullFakeRepo(_PortfolioServiceFakeRepo):
|
|
def __init__(self):
|
|
super().__init__(asset=_active_asset())
|
|
self._users = {"user-1": User(id="user-1", username="alice", permission="read", is_active=True)}
|
|
self._users_by_username = {"alice": self._users["user-1"]}
|
|
|
|
def get_user_by_username(self, username: str):
|
|
return self._users_by_username.get(username)
|
|
|
|
def get_portfolio_by_name_and_user(self, owner_user_id: str, name: str):
|
|
return next(
|
|
(p for p in self._portfolios.values()
|
|
if p.owner_user_id == owner_user_id and p.name == name),
|
|
None,
|
|
)
|
|
|
|
repo = _FullFakeRepo()
|
|
settings = Settings(allowed_origins_raw="http://localhost:3000")
|
|
app = create_app(
|
|
settings_factory=lambda: settings,
|
|
repository_factory=lambda _: repo,
|
|
)
|
|
runner = CliRunner()
|
|
result = runner.invoke(app, [
|
|
"portfolio", "create", "Test_Portfolio",
|
|
"--user", "alice",
|
|
"--symbol", "005930",
|
|
"--force",
|
|
])
|
|
assert result.exit_code == 0, result.output
|
|
assert len(repo._portfolios) == 1
|
|
assert list(repo._portfolios.values())[0].asset_id == _CATALOG_ASSET_ID
|