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)
1196 lines
46 KiB
Python
1196 lines
46 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:
|
|
from app.domain.models import StrategySet
|
|
|
|
self._asset = asset
|
|
strategy = StrategySet(id="set-1", name="target_value", config="{}", nick="target_value:default")
|
|
self._strategy_sets: dict = {strategy.id: strategy}
|
|
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 self._strategy_sets.get(id)
|
|
|
|
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, *, include_inactive: bool = True):
|
|
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",
|
|
strategy_id="set-1",
|
|
),
|
|
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",
|
|
strategy_id="set-1",
|
|
),
|
|
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",
|
|
strategy_id="set-1",
|
|
),
|
|
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",
|
|
"--strategy-id",
|
|
"set-1",
|
|
"--force",
|
|
])
|
|
assert result.exit_code == 0, result.output
|
|
assert len(repo._portfolios) == 1
|
|
assert list(repo._portfolios.values())[0].asset_id == _CATALOG_ASSET_ID
|
|
|
|
|
|
# ===========================================================================
|
|
# BAT-012: REQ-002 (assets.tag_ids) + REQ-010 AC-9~12 (GET /v1/assets)
|
|
# ===========================================================================
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-002 AC-1, AC-2: migration file structure for assets.tag_ids
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_migration_assets_tag_ids_file_exists() -> None:
|
|
import os
|
|
migrations_dir = os.path.join(os.path.dirname(__file__), "..", "migrations", "versions")
|
|
matches = [f for f in os.listdir(migrations_dir) if "0023" in f and "assets_tag_ids" in f]
|
|
assert len(matches) == 1, f"assets_tag_ids migration (0023) not found in {migrations_dir}"
|
|
|
|
|
|
def test_migration_assets_tag_ids_adds_column_and_gin_index() -> None:
|
|
import os
|
|
migrations_dir = os.path.join(os.path.dirname(__file__), "..", "migrations", "versions")
|
|
rev = next(f for f in os.listdir(migrations_dir) if "0023" in f and "assets_tag_ids" in f)
|
|
content = open(os.path.join(migrations_dir, rev)).read()
|
|
# AC-1: tag_ids UUID[] NOT NULL DEFAULT '{}' on assets
|
|
assert "add_column" in content
|
|
assert '"assets"' in content
|
|
assert '"tag_ids"' in content
|
|
assert "ARRAY" in content
|
|
assert "UUID" in content
|
|
assert "nullable=False" in content
|
|
assert "'{}'::uuid[]" in content
|
|
# AC-2: idx_assets_tag_ids_gin GIN index
|
|
assert "idx_assets_tag_ids_gin" in content
|
|
assert 'postgresql_using="gin"' in content
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-002 AC-3, AC-4: archived 표시 + 빈 배열 기본값 (Asset model + schema)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_asset_model_tag_ids_defaults_to_empty_list() -> None:
|
|
asset = Asset(asset_id="a-1", symbol="AAPL", markets=["NASDAQ"])
|
|
assert asset.tag_ids == []
|
|
|
|
|
|
def test_asset_model_accepts_tag_ids() -> None:
|
|
asset = Asset(
|
|
asset_id="a-1",
|
|
symbol="AAPL",
|
|
markets=["NASDAQ"],
|
|
tag_ids=["00000000-0000-0000-0000-00000000aaaa"],
|
|
)
|
|
assert asset.tag_ids == ["00000000-0000-0000-0000-00000000aaaa"]
|
|
|
|
|
|
def test_asset_catalog_response_schema_has_tag_ids_field() -> None:
|
|
from app.schemas.v1 import AssetCatalogResponse, TagRefItem
|
|
fields = AssetCatalogResponse.model_fields
|
|
assert "tag_ids" in fields
|
|
# Default factory should yield empty list
|
|
inst = AssetCatalogResponse(asset_id="a", symbol="s", markets=[], status="active")
|
|
assert inst.tag_ids == []
|
|
|
|
|
|
def test_build_tag_refs_archived_for_inactive_tags() -> None:
|
|
from app.api.v1_assets import _build_tag_refs
|
|
tag_active_map = {
|
|
"tid-active": True,
|
|
"tid-inactive": False,
|
|
}
|
|
refs = _build_tag_refs(["tid-active", "tid-inactive"], tag_active_map)
|
|
assert len(refs) == 2
|
|
assert refs[0].tag_id == "tid-active"
|
|
assert refs[0].archived is False
|
|
assert refs[1].tag_id == "tid-inactive"
|
|
assert refs[1].archived is True
|
|
|
|
|
|
def test_build_tag_refs_archived_for_missing_tags() -> None:
|
|
"""tag_id not present in tag master (deleted/never existed) → archived=True (이력 유지)."""
|
|
from app.api.v1_assets import _build_tag_refs
|
|
refs = _build_tag_refs(["tid-unknown"], {})
|
|
assert len(refs) == 1
|
|
assert refs[0].tag_id == "tid-unknown"
|
|
assert refs[0].archived is True
|
|
|
|
|
|
def test_build_tag_refs_empty_returns_empty_list() -> None:
|
|
from app.api.v1_assets import _build_tag_refs
|
|
assert _build_tag_refs([], {}) == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-010 AC-9~12: GET /v1/assets admin catalog list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_catalog_get_client(principal_kwargs: dict, assets: list[Asset], tags: dict[str, "object"] | None = None):
|
|
"""Build TestClient with mocked auth + list_assets + get_tags_by_ids."""
|
|
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,
|
|
)
|
|
principal = TokenPrincipal(**principal_kwargs)
|
|
mock_repo = MagicMock()
|
|
mock_repo.get_token_principal.return_value = principal
|
|
mock_repo.list_assets.return_value = assets
|
|
mock_repo.get_tags_by_ids.return_value = tags or {}
|
|
|
|
patches = (
|
|
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),
|
|
)
|
|
for p in patches:
|
|
p.start()
|
|
|
|
app = create_app(settings_factory=lambda: settings)
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
return TestClient(app), mock_repo, patches
|
|
|
|
|
|
def _make_asset_with_tags(asset_id: str, symbol: str, markets: list[str], status: str = "active", tag_ids: list[str] | None = None) -> Asset:
|
|
from datetime import datetime, timezone
|
|
return Asset(
|
|
asset_id=asset_id,
|
|
symbol=symbol,
|
|
markets=markets,
|
|
display_name=symbol,
|
|
status=status,
|
|
tag_ids=tag_ids or [],
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
|
|
|
|
_ADMIN_KW = dict(token_id="t-admin", user_id="admin-1", permission="admin", token_type="admin", runtime_id="rt-1")
|
|
_USER_KW = dict(token_id="t-user", user_id="user-1", permission="read", token_type="user", runtime_id="rt-1")
|
|
_SERVICE_KW = dict(token_id="t-svc", user_id=None, permission="trade-write", token_type="service", runtime_id="rt-1", daemon_id="d-1")
|
|
|
|
|
|
# REQ-010 AC-9: admin GET 200 + AssetCatalogListResponse shape (portfolio 필드 부재)
|
|
def test_get_v1_assets_admin_returns_catalog_response() -> None:
|
|
from app.domain.models import Tag
|
|
assets = [
|
|
_make_asset_with_tags("a-1", "005930", ["KRX"], status="active", tag_ids=["tag-active"]),
|
|
_make_asset_with_tags("a-2", "AAPL", ["NASDAQ"], status="active", tag_ids=[]),
|
|
]
|
|
tags = {"tag-active": Tag(tag_id="tag-active", name="kospi200", created_by_user_id="admin-1", is_active=True)}
|
|
client, _, patches = _make_catalog_get_client(_ADMIN_KW, assets, tags)
|
|
try:
|
|
resp = client.get("/v1/assets", headers={"Authorization": "Bearer admin"})
|
|
finally:
|
|
for p in patches:
|
|
p.stop()
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert "items" in body and "total" in body
|
|
assert body["total"] == 2
|
|
item = body["items"][0]
|
|
# Catalog-shape only — no portfolio fields
|
|
assert set(item.keys()) >= {"asset_id", "symbol", "markets", "display_name", "status", "tag_ids", "created_at", "updated_at"}
|
|
assert "cash" not in item and "quantity" not in item and "owner_user_id" not in item
|
|
# AC-3 boundary: active tag → archived False
|
|
a1 = next(i for i in body["items"] if i["asset_id"] == "a-1")
|
|
assert a1["tag_ids"] == [{"tag_id": "tag-active", "archived": False}]
|
|
# AC-4: empty default
|
|
a2 = next(i for i in body["items"] if i["asset_id"] == "a-2")
|
|
assert a2["tag_ids"] == []
|
|
|
|
|
|
# REQ-002 AC-3 integration: archived 표시 동반 반환 (inactive tag in GET response)
|
|
def test_get_v1_assets_archived_flag_for_inactive_tag() -> None:
|
|
from app.domain.models import Tag
|
|
assets = [
|
|
_make_asset_with_tags("a-1", "005930", ["KRX"], tag_ids=["tag-inactive"]),
|
|
]
|
|
tags = {"tag-inactive": Tag(tag_id="tag-inactive", name="legacy", created_by_user_id="admin-1", is_active=False)}
|
|
client, _, patches = _make_catalog_get_client(_ADMIN_KW, assets, tags)
|
|
try:
|
|
resp = client.get("/v1/assets", headers={"Authorization": "Bearer admin"})
|
|
finally:
|
|
for p in patches:
|
|
p.stop()
|
|
assert resp.status_code == 200
|
|
item = resp.json()["items"][0]
|
|
assert item["tag_ids"] == [{"tag_id": "tag-inactive", "archived": True}]
|
|
|
|
|
|
# REQ-010 AC-10: status=all/inactive 필터
|
|
def test_get_v1_assets_status_filter_all_includes_inactive() -> None:
|
|
assets = [
|
|
_make_asset_with_tags("a-1", "AAPL", ["NASDAQ"], status="active"),
|
|
_make_asset_with_tags("a-2", "DEFUNCT", ["KRX"], status="inactive"),
|
|
]
|
|
client, _, patches = _make_catalog_get_client(_ADMIN_KW, assets, {})
|
|
try:
|
|
resp = client.get("/v1/assets?status=all", headers={"Authorization": "Bearer admin"})
|
|
finally:
|
|
for p in patches:
|
|
p.stop()
|
|
assert resp.status_code == 200
|
|
symbols = {i["symbol"] for i in resp.json()["items"]}
|
|
assert symbols == {"AAPL", "DEFUNCT"}
|
|
|
|
|
|
def test_get_v1_assets_status_filter_inactive_only() -> None:
|
|
assets = [
|
|
_make_asset_with_tags("a-1", "AAPL", ["NASDAQ"], status="active"),
|
|
_make_asset_with_tags("a-2", "DEFUNCT", ["KRX"], status="inactive"),
|
|
]
|
|
client, _, patches = _make_catalog_get_client(_ADMIN_KW, assets, {})
|
|
try:
|
|
resp = client.get("/v1/assets?status=inactive", headers={"Authorization": "Bearer admin"})
|
|
finally:
|
|
for p in patches:
|
|
p.stop()
|
|
assert resp.status_code == 200
|
|
items = resp.json()["items"]
|
|
assert len(items) == 1
|
|
assert items[0]["symbol"] == "DEFUNCT"
|
|
|
|
|
|
def test_get_v1_assets_status_filter_default_active_only() -> None:
|
|
assets = [
|
|
_make_asset_with_tags("a-1", "AAPL", ["NASDAQ"], status="active"),
|
|
_make_asset_with_tags("a-2", "DEFUNCT", ["KRX"], status="inactive"),
|
|
]
|
|
client, _, patches = _make_catalog_get_client(_ADMIN_KW, assets, {})
|
|
try:
|
|
resp = client.get("/v1/assets", headers={"Authorization": "Bearer admin"})
|
|
finally:
|
|
for p in patches:
|
|
p.stop()
|
|
assert resp.status_code == 200
|
|
items = resp.json()["items"]
|
|
assert len(items) == 1
|
|
assert items[0]["symbol"] == "AAPL"
|
|
|
|
|
|
def test_get_v1_assets_status_filter_rejects_unknown_value() -> None:
|
|
client, _, patches = _make_catalog_get_client(_ADMIN_KW, [], {})
|
|
try:
|
|
resp = client.get("/v1/assets?status=garbage", headers={"Authorization": "Bearer admin"})
|
|
finally:
|
|
for p in patches:
|
|
p.stop()
|
|
assert resp.status_code == 422
|
|
|
|
|
|
# REQ-010 AC-11: market 필터
|
|
def test_get_v1_assets_market_filter() -> None:
|
|
assets = [
|
|
_make_asset_with_tags("a-1", "005930", ["KRX", "NXT"]),
|
|
_make_asset_with_tags("a-2", "AAPL", ["NASDAQ"]),
|
|
]
|
|
client, _, patches = _make_catalog_get_client(_ADMIN_KW, assets, {})
|
|
try:
|
|
resp = client.get("/v1/assets?market=KRX", headers={"Authorization": "Bearer admin"})
|
|
finally:
|
|
for p in patches:
|
|
p.stop()
|
|
assert resp.status_code == 200
|
|
items = resp.json()["items"]
|
|
assert len(items) == 1
|
|
assert items[0]["symbol"] == "005930"
|
|
|
|
|
|
# REQ-044: GET /v1/assets is now require_read — user/service tokens return 200
|
|
def test_get_v1_assets_user_token_returns_200() -> None:
|
|
"""REQ-044: read user token accepted by GET /v1/assets."""
|
|
client, _, patches = _make_catalog_get_client(_USER_KW, [], {})
|
|
try:
|
|
resp = client.get("/v1/assets", headers={"Authorization": "Bearer user"})
|
|
finally:
|
|
for p in patches:
|
|
p.stop()
|
|
assert resp.status_code == 200
|
|
assert resp.json()["total"] == 0
|
|
|
|
|
|
def test_get_v1_assets_service_token_returns_200() -> None:
|
|
"""REQ-044: trade-write service token accepted by GET /v1/assets (trade-write >= read)."""
|
|
client, _, patches = _make_catalog_get_client(_SERVICE_KW, [], {})
|
|
try:
|
|
resp = client.get("/v1/assets", headers={"Authorization": "Bearer svc"})
|
|
finally:
|
|
for p in patches:
|
|
p.stop()
|
|
assert resp.status_code == 200
|
|
assert resp.json()["total"] == 0
|
|
|
|
|
|
def test_get_v1_assets_no_token_returns_401() -> None:
|
|
"""GET /v1/assets without token returns 401 (require_read still gates auth)."""
|
|
client, _, patches = _make_catalog_get_client(_ADMIN_KW, [], {})
|
|
try:
|
|
resp = client.get("/v1/assets") # no Authorization header
|
|
finally:
|
|
for p in patches:
|
|
p.stop()
|
|
assert resp.status_code == 401
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# BAT-013: REQ-005 CHG-20260521-02 — assets 테이블 tag_ids mutation 정합화
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_postgres_repository_has_append_asset_tag_id_method() -> None:
|
|
"""BAT-013 / REQ-005 AC-1: PostgresRepository에 append_asset_tag_id 메서드 존재."""
|
|
from app.repositories.postgres import PostgresPortfolioRepository
|
|
|
|
assert hasattr(PostgresPortfolioRepository, "append_asset_tag_id")
|
|
assert callable(PostgresPortfolioRepository.append_asset_tag_id)
|
|
|
|
|
|
def test_postgres_repository_has_remove_asset_tag_id_method() -> None:
|
|
"""BAT-013 / REQ-005 AC-5: PostgresRepository에 remove_asset_tag_id 메서드 존재."""
|
|
from app.repositories.postgres import PostgresPortfolioRepository
|
|
|
|
assert hasattr(PostgresPortfolioRepository, "remove_asset_tag_id")
|
|
assert callable(PostgresPortfolioRepository.remove_asset_tag_id)
|
|
|
|
|
|
def test_append_asset_tag_id_targets_assets_table_not_portfolios() -> None:
|
|
"""BAT-013 / REQ-005 AC-1: append_asset_tag_id SQL이 assets 테이블을 UPDATE한다 (portfolios 아님)."""
|
|
from unittest.mock import MagicMock, patch
|
|
from app.repositories.postgres import PostgresPortfolioRepository
|
|
|
|
repo = PostgresPortfolioRepository.__new__(PostgresPortfolioRepository)
|
|
repo.ensure_schema = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.rowcount = 1
|
|
mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
|
|
mock_cursor.__exit__ = MagicMock(return_value=None)
|
|
mock_conn = MagicMock()
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
with patch.object(PostgresPortfolioRepository, "_connect", return_value=mock_conn):
|
|
repo.append_asset_tag_id("asset-uuid-1", "tag-uuid-1")
|
|
|
|
sql_executed = mock_cursor.execute.call_args_list[0][0][0]
|
|
assert "UPDATE assets" in sql_executed
|
|
assert "UPDATE portfolios" not in sql_executed
|
|
assert "WHERE asset_id" in sql_executed
|
|
|
|
|
|
def test_remove_asset_tag_id_targets_assets_table_not_portfolios() -> None:
|
|
"""BAT-013 / REQ-005 AC-5: remove_asset_tag_id SQL이 assets 테이블을 UPDATE한다 (portfolios 아님)."""
|
|
from unittest.mock import MagicMock, patch
|
|
from app.repositories.postgres import PostgresPortfolioRepository
|
|
|
|
repo = PostgresPortfolioRepository.__new__(PostgresPortfolioRepository)
|
|
repo.ensure_schema = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.rowcount = 1
|
|
mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
|
|
mock_cursor.__exit__ = MagicMock(return_value=None)
|
|
mock_conn = MagicMock()
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
with patch.object(PostgresPortfolioRepository, "_connect", return_value=mock_conn):
|
|
repo.remove_asset_tag_id("asset-uuid-1", "tag-uuid-1")
|
|
|
|
sql_executed = mock_cursor.execute.call_args_list[0][0][0]
|
|
assert "UPDATE assets" in sql_executed
|
|
assert "UPDATE portfolios" not in sql_executed
|
|
assert "WHERE asset_id" in sql_executed
|
|
|
|
|
|
def test_append_asset_tag_id_raises_when_asset_not_found() -> None:
|
|
"""BAT-013 / REQ-005 EC-1: 존재하지 않는 asset_id는 ValueError."""
|
|
from unittest.mock import MagicMock, patch
|
|
from app.repositories.postgres import PostgresPortfolioRepository
|
|
|
|
repo = PostgresPortfolioRepository.__new__(PostgresPortfolioRepository)
|
|
repo.ensure_schema = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.rowcount = 0
|
|
mock_cursor.fetchone.return_value = None
|
|
mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
|
|
mock_cursor.__exit__ = MagicMock(return_value=None)
|
|
mock_conn = MagicMock()
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
with patch.object(PostgresPortfolioRepository, "_connect", return_value=mock_conn):
|
|
with pytest.raises(ValueError, match="asset을 찾을 수 없습니다"):
|
|
repo.append_asset_tag_id("missing-asset-uuid", "tag-uuid-1")
|
|
|
|
|
|
def test_remove_asset_tag_id_raises_when_asset_not_found() -> None:
|
|
"""BAT-013 / REQ-005 EC-1: DELETE 시 asset 부재면 ValueError."""
|
|
from unittest.mock import MagicMock, patch
|
|
from app.repositories.postgres import PostgresPortfolioRepository
|
|
|
|
repo = PostgresPortfolioRepository.__new__(PostgresPortfolioRepository)
|
|
repo.ensure_schema = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.rowcount = 0
|
|
mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
|
|
mock_cursor.__exit__ = MagicMock(return_value=None)
|
|
mock_conn = MagicMock()
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
with patch.object(PostgresPortfolioRepository, "_connect", return_value=mock_conn):
|
|
with pytest.raises(ValueError, match="asset을 찾을 수 없습니다"):
|
|
repo.remove_asset_tag_id("missing-asset-uuid", "tag-uuid-1")
|
|
|
|
|
|
def test_append_asset_tag_id_idempotent_guard_present() -> None:
|
|
"""BAT-013 / REQ-005 EC-4: append_asset_tag_id SQL이 중복 방지 가드(`NOT ... = ANY(tag_ids)`)를 포함한다."""
|
|
from unittest.mock import MagicMock, patch
|
|
from app.repositories.postgres import PostgresPortfolioRepository
|
|
|
|
repo = PostgresPortfolioRepository.__new__(PostgresPortfolioRepository)
|
|
repo.ensure_schema = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.rowcount = 1
|
|
mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
|
|
mock_cursor.__exit__ = MagicMock(return_value=None)
|
|
mock_conn = MagicMock()
|
|
mock_conn.cursor.return_value = mock_cursor
|
|
|
|
with patch.object(PostgresPortfolioRepository, "_connect", return_value=mock_conn):
|
|
repo.append_asset_tag_id("asset-uuid-1", "tag-uuid-1")
|
|
|
|
sql_executed = mock_cursor.execute.call_args_list[0][0][0]
|
|
assert "NOT (" in sql_executed
|
|
assert "= ANY(tag_ids)" in sql_executed
|
|
assert "array_append" in sql_executed
|