portfolio-hub/tests/test_cli.py

from __future__ import annotations

import json
from datetime import datetime, timezone

import app.cli.commands.health as health_commands
from typer.testing import CliRunner

from app.cli.main import create_app
from app.core.config import Settings
from app.domain.models import LedgerEntry, Portfolio, StrategySet, User


class FakeRepo:
    def __init__(self, portfolio: Portfolio | None = None) -> None:
        self._portfolio = portfolio
        self.entries: list[LedgerEntry] = []
        self.strategy_set = StrategySet(id="set-1", name="target_value", config="{}", nick="target_value:99914b932bd3")
        self._strategy_sets: dict[str, StrategySet] = {self.strategy_set.id: self.strategy_set}
        self._users: dict[str, User] = {}
        self._user_seq = 1
        self.closed = False

    def close(self) -> None:
        self.closed = True

    def ensure_schema(self) -> None:
        return None

    def ping(self) -> None:
        return None

    def get_user(self, user_id: str) -> User:
        user = self._users.get(user_id)
        if user is None:
            raise ValueError("사용자를 찾을 수 없습니다")
        return user

    def list_users(self) -> list[User]:
        return sorted(self._users.values(), key=lambda user: user.username)

    def create_user(
        self,
        username: str,
        *,
        email: str | None = None,
        password_hash: str | None = None,
        permission: str = "read",
        is_active: bool = True,
    ) -> User:
        if any(user.username == username for user in self._users.values()):
            raise ValueError(f"이미 존재하는 username입니다: {username}")
        if email is not None and any(user.email == email for user in self._users.values()):
            raise ValueError(f"이미 존재하는 email입니다: {email}")
        now = datetime.now(timezone.utc)
        user = User(
            id=f"user-{self._user_seq}",
            username=username,
            email=email,
            permission=permission,
            is_active=is_active,
            password_hash=password_hash,
            created_at=now,
            updated_at=now,
        )
        self._user_seq += 1
        self._users[user.id] = user
        return user

    def update_user(
        self,
        user_id: str,
        *,
        username: str | None = None,
        email: str | None = None,
        password_hash: str | None = None,
        permission: str | None = None,
        is_active: bool | None = None,
    ) -> User:
        user = self.get_user(user_id)
        if username is not None and any(item.id != user_id and item.username == username for item in self._users.values()):
            raise ValueError(f"이미 존재하는 username입니다: {username}")
        if email is not None and any(item.id != user_id and item.email == email for item in self._users.values()):
            raise ValueError(f"이미 존재하는 email입니다: {email}")
        if username is not None:
            user.username = username
        if email is not None:
            user.email = email
        if password_hash is not None:
            user.password_hash = password_hash
        if permission is not None:
            user.permission = permission
        if is_active is not None:
            user.is_active = is_active
        user.updated_at = datetime.now(timezone.utc)
        return user

    def delete_user(self, user_id: str) -> None:
        if user_id not in self._users:
            raise ValueError("사용자를 찾을 수 없습니다")
        del self._users[user_id]

    def get(self, portfolio_id: str) -> Portfolio:
        if self._portfolio is None or self._portfolio.portfolio_id != portfolio_id:
            raise ValueError("포트폴리오를 찾을 수 없습니다")
        return self._portfolio

    def get_or_create(
        self,
        portfolio_id: str,
        symbol: str,
        *,
        broker: str = "kis",
        status: str = "enabled",
        active_strategy_id: str | None = None,
        avabile_strategies: list[str] | None = None,
        initial_cash: float = 0.0,
    ) -> Portfolio:
        if self._portfolio is None:
            self._portfolio = Portfolio(
                portfolio_id=portfolio_id,
                broker=broker,
                status=status,
                symbol=symbol,
                cash_balance=initial_cash,
                active_strategy_id=active_strategy_id,
                avabile_strategies=avabile_strategies,
            )
        return self._portfolio

    def create_portfolio(
        self,
        portfolio_id: str,
        *,
        broker: str,
        status: str,
        symbol: str,
        cash_balance: float,
        asset_quantity: float,
        asset_price: float,
        profit_buffer: float,
        active_strategy_id: str | None,
        avabile_strategies: list[str] | None,
    ) -> Portfolio:
        self._portfolio = Portfolio(
            portfolio_id=portfolio_id,
            broker=broker,
            status=status,
            symbol=symbol,
            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,
        )
        return self._portfolio

    def update_portfolio(self, portfolio_id: str, **fields: object) -> Portfolio:
        portfolio = self.get(portfolio_id)
        for key, value in fields.items():
            if value is not None:
                setattr(portfolio, key, value)
        return portfolio

    def list_portfolios(self, *, broker: str | None = None, status: str | None = None) -> list[Portfolio]:
        if self._portfolio is None:
            return []
        if broker is not None and self._portfolio.broker != broker:
            return []
        if status is not None and self._portfolio.status != status:
            return []
        return [self._portfolio]

    def get_or_create_strategy_set(self, name: str, config: str) -> StrategySet:
        self.strategy_set = StrategySet(id="set-1", name=name, config=config, nick=f"{name}:99914b932bd3")
        self._strategy_sets[self.strategy_set.id] = self.strategy_set
        return self.strategy_set

    def get_strategy_set(self, id: str) -> StrategySet | None:
        return self.strategy_set if self.strategy_set.id == id else None

    def list_strategy_sets(self) -> list[StrategySet]:
        from datetime import datetime, timezone
        return sorted(self._strategy_sets.values(), key=lambda s: s.created_at or datetime.now(timezone.utc), reverse=True)

    def delete_strategy_set(self, id: str) -> None:
        if id in self._strategy_sets:
            del self._strategy_sets[id]

    def save(
        self,
        portfolio: Portfolio,
        *,
        current_price: float,
        action: str,
        trade_price: float,
        quantity: float,
        fee_amount: float,
        tax_amount: float,
        total_value_after: float,
        strategy_set_id: str = "manual",
        note: str = "",
    ) -> None:
        self._portfolio = portfolio
        self.entries.insert(
            0,
            LedgerEntry(
                portfolio_id=portfolio.portfolio_id,
                recorded_at=datetime.now(timezone.utc),
                action=action,
                price=trade_price,
                quantity=quantity,
                fee_amount=fee_amount,
                tax_amount=tax_amount,
                cash_balance_after=portfolio.cash_balance,
                asset_quantity_after=portfolio.asset_quantity,
                total_value_after=total_value_after,
                profit_buffer_after=portfolio.profit_buffer,
                strategy_set_id=strategy_set_id,
                note=note,
            ),
        )

    def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]:
        return self.entries[:limit]

    def reset(self, portfolio_id: str) -> Portfolio:
        portfolio = self.get(portfolio_id)
        portfolio.cash_balance = 0.0
        portfolio.asset_quantity = 0.0
        portfolio.price = 0.0
        portfolio.profit_buffer = 0.0
        portfolio.active_strategy_id = None
        self.entries = []
        return portfolio

    def adjust_buffer(self, portfolio_id: str, amount: float, action: str, note: str = "") -> Portfolio:
        portfolio = self.get(portfolio_id)
        if action == "INVEST":
            portfolio.profit_buffer += amount
        else:
            portfolio.profit_buffer -= amount
        return portfolio

    def transfer_buffer(self, portfolio_id: str, amount: float, direction: str, note: str = "") -> Portfolio:
        portfolio = self.get(portfolio_id)
        if direction == "to-buffer":
            portfolio.cash_balance -= amount
            portfolio.profit_buffer += amount
        else:
            portfolio.cash_balance += amount
            portfolio.profit_buffer -= amount
        return portfolio

    def delete(self, portfolio_id: str, *, clear_ledger: bool = True) -> None:
        self._portfolio = None
        if clear_ledger:
            self.entries = []


def _settings() -> Settings:
    return Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000")


def build_runner(repo: FakeRepo) -> CliRunner:
    app = create_app(settings_factory=_settings, repository_factory=lambda settings: repo)
    runner = CliRunner()
    runner.app = app
    return runner


def _portfolio() -> Portfolio:
    return Portfolio(
        portfolio_id="005930:tv",
        broker="kis",
        status="enabled",
        symbol="005930",
        cash_balance=1_000_000.0,
        asset_quantity=1.0,
        price=70_000.0,
        profit_buffer=50_000.0,
        active_strategy_id="set-1",
        avabile_strategies=["set-1"],
    )


def test_cli_list_renders_table() -> None:
    repo = FakeRepo(_portfolio())
    runner = build_runner(repo)

    result = runner.invoke(runner.app, ["portfolio", "list"])

    assert result.exit_code == 0
    assert "005930:tv" in result.stdout
    assert "Portfolios" in result.stdout


def test_cli_status_supports_json_output() -> None:
    repo = FakeRepo(_portfolio())
    runner = build_runner(repo)

    result = runner.invoke(runner.app, ["--format", "json", "portfolio", "status", "005930:tv"])

    assert result.exit_code == 0
    payload = json.loads(result.stdout)
    assert payload["portfolio"]["portfolio_id"] == "005930:tv"


def test_cli_create_buy_and_delete_flow() -> None:
    repo = FakeRepo()
    runner = build_runner(repo)

    create_result = runner.invoke(
        runner.app,
        [
            "portfolio",
            "create",
            "005930:tv",
            "--symbol",
            "005930",
            "--cash-balance",
            "1000000",
            "--strategy-params",
            '{"target_asset_ratio": 0.5}',
            "--force",
        ],
    )
    buy_result = runner.invoke(
        runner.app,
        [
            "portfolio",
            "buy",
            "005930:tv",
            "--quantity",
            "2",
            "--price",
            "50000",
            "--fee",
            "100",
            "--force",
        ],
    )
    assert repo.entries[0].action == "BUY"
    delete_result = runner.invoke(runner.app, ["portfolio", "delete", "005930:tv", "--keep-ledger", "--force"])

    assert create_result.exit_code == 0
    assert buy_result.exit_code == 0
    assert delete_result.exit_code == 0
    assert "Deleted portfolio: 005930:tv" in delete_result.stdout


def test_cli_write_command_prompts_before_create() -> None:
    repo = FakeRepo()
    runner = build_runner(repo)

    result = runner.invoke(
        runner.app,
        [
            "portfolio",
            "create",
            "005930:tv",
            "--symbol",
            "005930",
            "--cash-balance",
            "1000000",
        ],
        input="y\n",
    )

    assert result.exit_code == 0
    assert "Create portfolio 005930:tv for symbol 005930?" in result.stdout


def test_cli_write_command_can_abort() -> None:
    repo = FakeRepo()
    runner = build_runner(repo)

    result = runner.invoke(
        runner.app,
        [
            "portfolio",
            "create",
            "005930:tv",
            "--symbol",
            "005930",
            "--cash-balance",
            "1000000",
        ],
        input="n\n",
    )

    assert result.exit_code == 0
    assert repo._portfolio is None


def test_cli_health_check_reports_repository() -> None:
    repo = FakeRepo(_portfolio())
    runner = build_runner(repo)

    result = runner.invoke(runner.app, ["health-check"])

    assert result.exit_code == 0
    assert "repository: FakeRepo" in result.stdout


def test_cli_init_db_reports_created(monkeypatch) -> None:
    repo = FakeRepo(_portfolio())
    runner = build_runner(repo)

    def fake_initialize_database(*, settings, maintenance_db: str = "postgres") -> dict[str, object]:
        return {
            "database": settings.active_db_name,
            "host": settings.active_db_host,
            "port": settings.active_db_port,
            "created": True,
            "schema_initialized": True,
            "migration_applied": "head",
        }

    monkeypatch.setattr(health_commands, "initialize_database", fake_initialize_database)

    result = runner.invoke(runner.app, ["init-db"])

    assert result.exit_code == 0
    assert "created: True" in result.stdout
    assert "schema_initialized: True" in result.stdout
    assert "migration_applied: head" in result.stdout


def test_cli_init_db_supports_json(monkeypatch) -> None:
    repo = FakeRepo(_portfolio())
    runner = build_runner(repo)

    def fake_initialize_database(*, settings, maintenance_db: str = "postgres") -> dict[str, object]:
        return {
            "database": "portfoliohub",
            "host": "localhost",
            "port": 5432,
            "created": False,
            "schema_initialized": True,
            "migration_applied": "head",
        }

    monkeypatch.setattr(health_commands, "initialize_database", fake_initialize_database)

    result = runner.invoke(runner.app, ["--format", "json", "init-db"])

    assert result.exit_code == 0
    payload = json.loads(result.stdout)
    assert payload["database"] == "portfoliohub"
    assert payload["schema_initialized"] is True


def test_cli_migrate_db_reports_head(monkeypatch) -> None:
    repo = FakeRepo(_portfolio())
    runner = build_runner(repo)

    def fake_bootstrap_or_upgrade_to_head(settings) -> None:
        return None

    monkeypatch.setattr(health_commands, "bootstrap_or_upgrade_to_head", fake_bootstrap_or_upgrade_to_head)

    result = runner.invoke(runner.app, ["migrate-db"])

    assert result.exit_code == 0
    assert "migration_applied: head" in result.stdout


def test_cli_user_list_renders_table() -> None:
    repo = FakeRepo()
    repo.create_user("alice", email="alice@example.com")
    runner = build_runner(repo)

    result = runner.invoke(runner.app, ["user", "list"])

    assert result.exit_code == 0
    assert "Users" in result.stdout
    assert "alice" in result.stdout


def test_cli_user_create_supports_json() -> None:
    repo = FakeRepo()
    runner = build_runner(repo)

    result = runner.invoke(
        runner.app,
        [
            "--format",
            "json",
            "user",
            "create",
            "--username",
            "alice",
            "--password",
            "password123",
            "--email",
            "alice@example.com",
            "--force",
        ],
    )

    assert result.exit_code == 0
    payload = json.loads(result.stdout)
    assert payload["username"] == "alice"
    assert payload["email"] == "alice@example.com"


def test_cli_user_update_and_delete_flow() -> None:
    repo = FakeRepo()
    user = repo.create_user("alice", email="alice@example.com")
    runner = build_runner(repo)

    update_result = runner.invoke(
        runner.app,
        ["user", "update", user.id, "--username", "alice-updated", "--force"],
    )
    delete_result = runner.invoke(runner.app, ["user", "delete", user.id, "--force"])

    assert update_result.exit_code == 0
    assert "alice-updated" in update_result.stdout
    assert delete_result.exit_code == 0
    assert f"Deleted user: {user.id}" in delete_result.stdout