portfolio-hub/tests/test_strategy_catalog_postgres_integration.py

"""Real-PostgreSQL concurrency and migration verification for Issue #26.

Run with PORTFOLIOHUB_RUN_POSTGRES_INTEGRATION=1 against a disposable/local
PostgreSQL endpoint. Normal unit-test environments skip this module.
"""

from __future__ import annotations

import hashlib
import os
import subprocess
import threading
import uuid
from pathlib import Path

import psycopg
import pytest
from psycopg import sql

from app.domain.errors import StrategyNotFoundError
from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioRepository

pytestmark = pytest.mark.skipif(
    os.getenv("PORTFOLIOHUB_RUN_POSTGRES_INTEGRATION") != "1",
    reason="set PORTFOLIOHUB_RUN_POSTGRES_INTEGRATION=1 to run real PostgreSQL tests",
)

_ROOT = Path(__file__).parents[1]


def _db_values(dbname: str | None = None) -> dict[str, object]:
    return {
        "host": os.getenv("TIMESCALEDB_HOST", "127.0.0.1"),
        "port": int(os.getenv("TIMESCALEDB_PORT", "5432")),
        "dbname": dbname or os.getenv("TIMESCALEDB_NAME", "portfolio_hub_migration_test"),
        "user": os.getenv("TIMESCALEDB_USER", "portfolio_hub_user"),
        "password": os.getenv("TIMESCALEDB_PASSWORD", "change-me"),
        "sslmode": os.getenv("TIMESCALEDB_SSLMODE", "prefer"),
    }


def _repo(dbname: str | None = None) -> PostgresPortfolioRepository:
    values = _db_values(dbname)
    return PostgresPortfolioRepository(
        PortfolioDBConfig(
            host=str(values["host"]),
            port=int(values["port"]),
            dbname=str(values["dbname"]),
            user=str(values["user"]),
            password=str(values["password"]),
            sslmode=str(values["sslmode"]),
        )
    )


def _prepare_catalog_fixture(repo: PostgresPortfolioRepository) -> tuple[str, str, str]:
    repo.ensure_schema()
    strategy = repo.create_strategy_set(f"concurrency-{uuid.uuid4().hex}", "{}")
    asset_id = str(uuid.uuid4())
    owner_user_id = f"qc-user-{uuid.uuid4().hex}"
    with repo._connect().cursor() as cur:
        cur.execute(
            "INSERT INTO users (id, username, permission, is_active) VALUES (%s, %s, 'trade-write', TRUE)",
            (owner_user_id, owner_user_id),
        )
        cur.execute(
            """
            INSERT INTO assets (asset_id, symbol, markets, status)
            VALUES (%s, %s, ARRAY['KRX'], 'active')
            """,
            (asset_id, f"QC{uuid.uuid4().hex[:10]}"),
        )
    return strategy.id, asset_id, owner_user_id


def test_concurrent_strategy_patch_keeps_derived_nick_consistent() -> None:
    owner = _repo()
    strategy_id, asset_id, owner_user_id = _prepare_catalog_fixture(owner)
    renamed = f"concurrent-renamed-{uuid.uuid4().hex}"
    barrier = threading.Barrier(3)
    errors: list[BaseException] = []

    def patch_name() -> None:
        repo = _repo()
        try:
            barrier.wait()
            repo.update_strategy_set(strategy_id, name=renamed)
        except BaseException as exc:  # pragma: no cover - only populated on failure
            errors.append(exc)
        finally:
            repo.close()

    def patch_config() -> None:
        repo = _repo()
        try:
            barrier.wait()
            repo.update_strategy_set(strategy_id, config='{"window":21}')
        except BaseException as exc:  # pragma: no cover - only populated on failure
            errors.append(exc)
        finally:
            repo.close()

    threads = [threading.Thread(target=patch_name), threading.Thread(target=patch_config)]
    for thread in threads:
        thread.start()
    barrier.wait()
    for thread in threads:
        thread.join(timeout=10)
        assert not thread.is_alive()

    final = owner.get_strategy_set(strategy_id)
    assert errors == []
    assert final is not None
    assert final.name == renamed
    assert final.config == '{"window":21}'
    assert final.revision == 3
    expected_nick = f"{final.name}:{hashlib.sha1(final.config.encode()).hexdigest()[:12]}"
    assert final.nick == expected_nick
    portfolio_id = str(uuid.uuid4())
    owner.create_portfolio(
        portfolio_id,
        owner_user_id=owner_user_id,
        broker="kis",
        status="enabled",
        name="latest-revision-check",
        asset_id=asset_id,
        cash_balance=0,
        asset_quantity=0,
        asset_price=0,
        profit_buffer=0,
        active_strategy_id=strategy_id,
        avabile_strategies=[strategy_id],
    )
    portfolio = owner.get(portfolio_id)
    assert portfolio.strategy_params == {"window": 21}
    assert portfolio.strategy_revision == 3
    with owner._connect().cursor() as cur:
        cur.execute("DELETE FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
    owner.delete_strategy_set(strategy_id)
    with owner._connect().cursor() as cur:
        cur.execute("DELETE FROM assets WHERE asset_id = %s", (asset_id,))
        cur.execute("DELETE FROM users WHERE id = %s", (owner_user_id,))
    owner.close()


def test_portfolio_selection_and_strategy_delete_never_create_orphan() -> None:
    owner = _repo()
    strategy_id, asset_id, owner_user_id = _prepare_catalog_fixture(owner)
    portfolio_id = str(uuid.uuid4())
    barrier = threading.Barrier(3)
    outcomes: list[str] = []

    def select_strategy() -> None:
        repo = _repo()
        try:
            barrier.wait()
            repo.create_portfolio(
                portfolio_id,
                owner_user_id=owner_user_id,
                broker="kis",
                status="enabled",
                name="race-check",
                asset_id=asset_id,
                cash_balance=0,
                asset_quantity=0,
                asset_price=0,
                profit_buffer=0,
                active_strategy_id=strategy_id,
                avabile_strategies=[strategy_id],
            )
            outcomes.append("selected")
        except StrategyNotFoundError:
            outcomes.append("selection-rejected")
        finally:
            repo.close()

    def delete_strategy() -> None:
        repo = _repo()
        try:
            barrier.wait()
            repo.delete_strategy_set(strategy_id)
            outcomes.append("deleted")
        except ValueError as exc:
            assert "in use" in str(exc)
            outcomes.append("delete-rejected")
        finally:
            repo.close()

    threads = [threading.Thread(target=select_strategy), threading.Thread(target=delete_strategy)]
    for thread in threads:
        thread.start()
    barrier.wait()
    for thread in threads:
        thread.join(timeout=10)
        assert not thread.is_alive()

    with owner._connect().cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
        portfolio_exists = cur.fetchone()[0] == 1
        cur.execute("SELECT COUNT(*) FROM strategy_sets WHERE id = %s", (strategy_id,))
        strategy_exists = cur.fetchone()[0] == 1
    assert not (portfolio_exists and not strategy_exists)
    assert sorted(outcomes) in (["delete-rejected", "selected"], ["deleted", "selection-rejected"])

    with owner._connect().cursor() as cur:
        cur.execute("DELETE FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
        cur.execute("DELETE FROM strategy_sets WHERE id = %s", (strategy_id,))
        cur.execute("DELETE FROM assets WHERE asset_id = %s", (asset_id,))
        cur.execute("DELETE FROM users WHERE id = %s", (owner_user_id,))
    owner.close()


def test_stale_trade_save_cannot_restore_a_deleted_strategy_reference() -> None:
    owner = _repo()
    strategy_a, asset_id, owner_user_id = _prepare_catalog_fixture(owner)
    strategy_b = owner.create_strategy_set(f"replacement-{uuid.uuid4().hex}", "{}")
    portfolio_id = str(uuid.uuid4())
    owner.create_portfolio(
        portfolio_id,
        owner_user_id=owner_user_id,
        broker="kis",
        status="enabled",
        name="stale-trade-check",
        asset_id=asset_id,
        cash_balance=1_000,
        asset_quantity=0,
        asset_price=100,
        profit_buffer=0,
        active_strategy_id=strategy_a,
        avabile_strategies=[strategy_a],
    )
    stale = owner.get(portfolio_id)

    concurrent = _repo()
    concurrent.update_portfolio(
        portfolio_id,
        active_strategy_id=strategy_b.id,
        avabile_strategies=[strategy_b.id],
    )
    concurrent.delete_strategy_set(strategy_a)
    concurrent.close()

    stale.cash_balance = 900
    stale.asset_quantity = 1
    owner.save(
        stale,
        current_price=100,
        action="BUY",
        trade_price=100,
        quantity=1,
        fee_amount=0,
        tax_amount=0,
        total_value_after=1_000,
        strategy_set_id=strategy_a,
        note="stale snapshot regression",
    )

    current = owner.get(portfolio_id)
    assert current.active_strategy_id == strategy_b.id
    assert current.avabile_strategies == [strategy_b.id]
    assert owner.get_strategy_set(strategy_a) is None
    with owner._connect().cursor() as cur:
        cur.execute("DELETE FROM portfolio_ledger WHERE portfolio_id = %s", (portfolio_id,))
        cur.execute("DELETE FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
        cur.execute("DELETE FROM strategy_sets WHERE id = %s", (strategy_b.id,))
        cur.execute("DELETE FROM assets WHERE asset_id = %s", (asset_id,))
        cur.execute("DELETE FROM users WHERE id = %s", (owner_user_id,))
    owner.close()


def test_alembic_upgrade_is_repeatable_on_fresh_postgres_database() -> None:
    database_name = f"portfolio_hub_0029_{uuid.uuid4().hex[:12]}"
    maintenance = _db_values("postgres")
    connection = psycopg.connect(**maintenance, autocommit=True)
    try:
        with connection.cursor() as cur:
            cur.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(database_name)))
        env = os.environ.copy()
        env.update(
            {
                "TIMESCALEDB_HOST": str(maintenance["host"]),
                "TIMESCALEDB_PORT": str(maintenance["port"]),
                "TIMESCALEDB_NAME": database_name,
                "TIMESCALEDB_USER": str(maintenance["user"]),
                "TIMESCALEDB_PASSWORD": str(maintenance["password"]),
                "TIMESCALEDB_SSLMODE": str(maintenance["sslmode"]),
            }
        )
        for _ in range(2):
            subprocess.run(
                [str(_ROOT / ".venv/bin/python"), "-m", "alembic", "upgrade", "head"],
                cwd=_ROOT,
                env=env,
                check=True,
                capture_output=True,
                text=True,
            )
        test_db = psycopg.connect(**_db_values(database_name))
        try:
            with test_db.cursor() as cur:
                cur.execute(
                    """
                    SELECT id, status, revision, nick
                    FROM strategy_sets
                    WHERE name = 'target_value' AND config = '{}'
                    """
                )
                rows = cur.fetchall()
                assert rows == [
                    (
                        "00000000-0000-0000-0000-000000000029",
                        "active",
                        1,
                        "target_value:bf21a9e8fbc5",
                    )
                ]
                cur.execute("SELECT version_num FROM alembic_version")
                assert cur.fetchone()[0] == "20260814_0029"
        finally:
            test_db.close()
    finally:
        with connection.cursor() as cur:
            cur.execute(
                "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = %s",
                (database_name,),
            )
            cur.execute(sql.SQL("DROP DATABASE IF EXISTS {}").format(sql.Identifier(database_name)))
        connection.close()