"""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 # =========================================================================== # 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