330 lines
12 KiB
Python
330 lines
12 KiB
Python
"""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 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_preserves_existing_nick() -> None:
|
|
owner = _repo()
|
|
strategy_id, asset_id, owner_user_id = _prepare_catalog_fixture(owner)
|
|
initial = owner.get_strategy_set(strategy_id)
|
|
assert initial is not None
|
|
initial_nick = initial.nick
|
|
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
|
|
assert final.nick == initial_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/pfhub"), "migrate-db"],
|
|
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()
|