portfolio-hub/tests/test_asset_catalog.py

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