526 lines
19 KiB
Python
526 lines
19 KiB
Python
"""Real-PostgreSQL persistence and concurrency verification for daemon assignment (Issue #27 / dcy-0004).
|
|
|
|
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 threading
|
|
import uuid
|
|
from contextlib import ExitStack
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.config import Settings, get_settings
|
|
from app.domain.errors import DaemonInactiveError, DaemonInUseError, DaemonNotFoundError
|
|
from app.main import create_app
|
|
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",
|
|
)
|
|
|
|
|
|
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_daemon_fixture(repo: PostgresPortfolioRepository) -> tuple[str, str, str, str]:
|
|
repo.ensure_schema()
|
|
strategy = repo.create_strategy_set(f"daemon-{uuid.uuid4().hex}", "{}")
|
|
asset_id = str(uuid.uuid4())
|
|
owner_user_id = f"daemon-owner-{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"DA{uuid.uuid4().hex[:10]}"),
|
|
)
|
|
daemon = repo.create_daemon(owner_user_id, daemon_slug=f"d-{uuid.uuid4().hex[:10]}")
|
|
return strategy.id, asset_id, owner_user_id, daemon.id
|
|
|
|
|
|
def _cleanup(
|
|
repo: PostgresPortfolioRepository,
|
|
portfolio_ids: list[str],
|
|
strategy_id: str,
|
|
asset_id: str,
|
|
owner_user_id: str,
|
|
daemon_id: str,
|
|
extra_user_ids: list[str] | None = None,
|
|
) -> None:
|
|
with repo._connect().cursor() as cur:
|
|
for portfolio_id in portfolio_ids:
|
|
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 auth_tokens WHERE daemon_id = %s", (daemon_id,))
|
|
cur.execute("DELETE FROM daemons WHERE id = %s", (daemon_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,))
|
|
for user_id in extra_user_ids or []:
|
|
cur.execute("DELETE FROM users WHERE id = %s", (user_id,))
|
|
repo.close()
|
|
|
|
|
|
def _make_client() -> TestClient:
|
|
get_settings.cache_clear()
|
|
settings = Settings(
|
|
allowed_origins_raw="http://localhost:3000",
|
|
admin_tokens_raw="admin-tok",
|
|
trade_write_tokens_raw="user-tok",
|
|
)
|
|
app = create_app(settings_factory=lambda: settings)
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
return TestClient(app)
|
|
|
|
|
|
def _with_repo(repo: PostgresPortfolioRepository) -> ExitStack:
|
|
stack = ExitStack()
|
|
stack.enter_context(
|
|
patch("app.api.v1_portfolios.get_portfolio_repository_from_settings", return_value=repo)
|
|
)
|
|
return stack
|
|
|
|
|
|
def _auth(token: str) -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def test_http_post_returns_daemon_and_list_get_preserves_it() -> None:
|
|
repo = _repo()
|
|
strategy_id, asset_id, owner_user_id, daemon_id = _prepare_daemon_fixture(repo)
|
|
client = _make_client()
|
|
|
|
try:
|
|
with _with_repo(repo):
|
|
resp = client.post(
|
|
"/v1/portfolios",
|
|
json={
|
|
"asset_id": asset_id,
|
|
"broker": "kis",
|
|
"strategy_id": strategy_id,
|
|
"name": "daemon-assignment-check",
|
|
"user_id": owner_user_id,
|
|
"managed_by_daemon_id": daemon_id,
|
|
},
|
|
headers=_auth("admin-tok"),
|
|
)
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
assert body["managed_by_daemon_id"] == daemon_id
|
|
portfolio_id = body["portfolio_id"]
|
|
|
|
stored = repo.get(portfolio_id)
|
|
assert stored.managed_by_daemon_id == daemon_id
|
|
|
|
with _with_repo(repo):
|
|
list_resp = client.get("/v1/portfolios", headers=_auth("admin-tok"))
|
|
assert list_resp.status_code == 200
|
|
matched = [p for p in list_resp.json()["portfolios"] if p["portfolio_id"] == portfolio_id]
|
|
assert matched and matched[0]["managed_by_daemon_id"] == daemon_id
|
|
|
|
_cleanup(repo, [portfolio_id], strategy_id, asset_id, owner_user_id, daemon_id)
|
|
finally:
|
|
pass
|
|
|
|
|
|
def test_validation_failure_leaves_no_partial_portfolio() -> None:
|
|
repo = _repo()
|
|
strategy_id, asset_id, owner_user_id, daemon_id = _prepare_daemon_fixture(repo)
|
|
portfolio_id = str(uuid.uuid4())
|
|
inactive = _repo()
|
|
try:
|
|
inactive.update_daemon(daemon_id, is_active=False)
|
|
|
|
with pytest.raises(DaemonInactiveError):
|
|
repo.create_portfolio(
|
|
portfolio_id,
|
|
owner_user_id=owner_user_id,
|
|
broker="kis",
|
|
status="enabled",
|
|
name="no-partial-inactive",
|
|
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],
|
|
managed_by_daemon_id=daemon_id,
|
|
)
|
|
with repo._connect().cursor() as cur:
|
|
cur.execute("SELECT COUNT(*) FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
|
|
assert cur.fetchone()[0] == 0
|
|
|
|
foreign_user_id = f"foreign-owner-{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)",
|
|
(foreign_user_id, foreign_user_id),
|
|
)
|
|
foreign_daemon = repo.create_daemon(foreign_user_id, daemon_slug=f"foreign-{uuid.uuid4().hex[:10]}")
|
|
with pytest.raises(DaemonNotFoundError):
|
|
repo.create_portfolio(
|
|
portfolio_id,
|
|
owner_user_id=owner_user_id,
|
|
broker="kis",
|
|
status="enabled",
|
|
name="no-partial-foreign",
|
|
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],
|
|
managed_by_daemon_id=foreign_daemon.id,
|
|
)
|
|
with repo._connect().cursor() as cur:
|
|
cur.execute("SELECT COUNT(*) FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
|
|
assert cur.fetchone()[0] == 0
|
|
|
|
with repo._connect().cursor() as cur:
|
|
cur.execute("DELETE FROM auth_tokens WHERE daemon_id = %s", (foreign_daemon.id,))
|
|
cur.execute("DELETE FROM daemons WHERE id = %s", (foreign_daemon.id,))
|
|
cur.execute("DELETE FROM users WHERE id = %s", (foreign_user_id,))
|
|
_cleanup(repo, [], strategy_id, asset_id, owner_user_id, daemon_id)
|
|
finally:
|
|
inactive.close()
|
|
|
|
|
|
def test_assignment_and_deactivation_race_never_commits_inactive_assignment() -> None:
|
|
repo = _repo()
|
|
strategy_id, asset_id, owner_user_id, daemon_id = _prepare_daemon_fixture(repo)
|
|
portfolio_id = str(uuid.uuid4())
|
|
barrier = threading.Barrier(3)
|
|
outcomes: list[str] = []
|
|
|
|
def assign_daemon() -> None:
|
|
worker = _repo()
|
|
try:
|
|
barrier.wait()
|
|
worker.create_portfolio(
|
|
portfolio_id,
|
|
owner_user_id=owner_user_id,
|
|
broker="kis",
|
|
status="enabled",
|
|
name="daemon-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],
|
|
managed_by_daemon_id=daemon_id,
|
|
)
|
|
outcomes.append("assigned")
|
|
except DaemonInactiveError:
|
|
outcomes.append("assignment-rejected")
|
|
finally:
|
|
worker.close()
|
|
|
|
def deactivate_daemon() -> None:
|
|
worker = _repo()
|
|
try:
|
|
barrier.wait()
|
|
worker.update_daemon(daemon_id, is_active=False)
|
|
outcomes.append("deactivated")
|
|
finally:
|
|
worker.close()
|
|
|
|
threads = [threading.Thread(target=assign_daemon), threading.Thread(target=deactivate_daemon)]
|
|
for thread in threads:
|
|
thread.start()
|
|
barrier.wait()
|
|
for thread in threads:
|
|
thread.join(timeout=10)
|
|
assert not thread.is_alive()
|
|
|
|
assert sorted(outcomes) in (
|
|
["assigned", "deactivated"],
|
|
["assignment-rejected", "deactivated"],
|
|
)
|
|
with repo._connect().cursor() as cur:
|
|
cur.execute("SELECT COUNT(*) FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
|
|
portfolio_exists = cur.fetchone()[0] == 1
|
|
if portfolio_exists:
|
|
assert repo.get(portfolio_id).managed_by_daemon_id == daemon_id
|
|
|
|
with repo._connect().cursor() as cur:
|
|
if portfolio_exists:
|
|
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 auth_tokens WHERE daemon_id = %s", (daemon_id,))
|
|
cur.execute("DELETE FROM daemons WHERE id = %s", (daemon_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,))
|
|
repo.close()
|
|
|
|
|
|
def test_patch_reassignment_persists_new_daemon() -> None:
|
|
repo = _repo()
|
|
strategy_id, asset_id, owner_user_id, daemon_a = _prepare_daemon_fixture(repo)
|
|
daemon_b = repo.create_daemon(owner_user_id, daemon_slug=f"d-{uuid.uuid4().hex[:10]}")
|
|
portfolio_id = str(uuid.uuid4())
|
|
try:
|
|
repo.create_portfolio(
|
|
portfolio_id,
|
|
owner_user_id=owner_user_id,
|
|
broker="kis",
|
|
status="enabled",
|
|
name="reassign-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],
|
|
managed_by_daemon_id=daemon_a,
|
|
)
|
|
assert repo.get(portfolio_id).managed_by_daemon_id == daemon_a
|
|
|
|
repo.update_portfolio(portfolio_id, managed_by_daemon_id=daemon_b.id)
|
|
assert repo.get(portfolio_id).managed_by_daemon_id == daemon_b.id
|
|
|
|
with repo._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 auth_tokens WHERE daemon_id = %s", (daemon_b.id,))
|
|
cur.execute("DELETE FROM daemons WHERE id = %s", (daemon_b.id,))
|
|
_cleanup(repo, [], strategy_id, asset_id, owner_user_id, daemon_a)
|
|
finally:
|
|
pass
|
|
|
|
|
|
def test_referenced_daemon_owner_transfer_is_atomic() -> None:
|
|
repo = _repo()
|
|
strategy_id, asset_id, owner_a, daemon_id = _prepare_daemon_fixture(repo)
|
|
owner_b = f"daemon-owner-{uuid.uuid4().hex}"
|
|
portfolio_id = str(uuid.uuid4())
|
|
try:
|
|
with repo._connect().cursor() as cur:
|
|
cur.execute(
|
|
"INSERT INTO users (id, username, permission, is_active) VALUES (%s, %s, 'trade-write', TRUE)",
|
|
(owner_b, owner_b),
|
|
)
|
|
repo.upsert_broker_credential(
|
|
owner_a,
|
|
"kis",
|
|
account_ref="owner-a-account",
|
|
encrypted_payload="ciphertext",
|
|
daemon_id=daemon_id,
|
|
)
|
|
repo.create_portfolio(
|
|
portfolio_id,
|
|
owner_user_id=owner_a,
|
|
broker="kis",
|
|
status="enabled",
|
|
name="owner-transfer-blocked",
|
|
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],
|
|
managed_by_daemon_id=daemon_id,
|
|
)
|
|
|
|
with pytest.raises(DaemonInUseError):
|
|
repo.update_daemon_owner(daemon_id, owner_user_id=owner_b)
|
|
|
|
assert repo.get_daemon_by_id(daemon_id).owner_user_id == owner_a
|
|
assert repo.get(portfolio_id).managed_by_daemon_id == daemon_id
|
|
credential = repo.get_broker_credential_by_daemon(daemon_id, "kis")
|
|
assert credential is not None
|
|
assert credential.user_id == owner_a
|
|
|
|
_cleanup(repo, [portfolio_id], strategy_id, asset_id, owner_a, daemon_id, [owner_b])
|
|
finally:
|
|
pass
|
|
|
|
|
|
def test_unreferenced_daemon_owner_transfer_invalidates_credentials() -> None:
|
|
repo = _repo()
|
|
strategy_id, asset_id, owner_a, daemon_id = _prepare_daemon_fixture(repo)
|
|
owner_b = f"daemon-owner-{uuid.uuid4().hex}"
|
|
try:
|
|
with repo._connect().cursor() as cur:
|
|
cur.execute(
|
|
"INSERT INTO users (id, username, permission, is_active) VALUES (%s, %s, 'trade-write', TRUE)",
|
|
(owner_b, owner_b),
|
|
)
|
|
repo.upsert_broker_credential(
|
|
owner_a,
|
|
"kis",
|
|
account_ref="owner-a-account",
|
|
encrypted_payload="ciphertext",
|
|
daemon_id=daemon_id,
|
|
)
|
|
|
|
moved = repo.update_daemon_owner(daemon_id, owner_user_id=owner_b)
|
|
|
|
assert moved.owner_user_id == owner_b
|
|
assert repo.get_broker_credential_by_daemon(daemon_id, "kis") is None
|
|
|
|
_cleanup(repo, [], strategy_id, asset_id, owner_a, daemon_id, [owner_b])
|
|
finally:
|
|
pass
|
|
|
|
|
|
def test_assignment_and_owner_transfer_race_preserves_same_owner_invariant() -> None:
|
|
repo = _repo()
|
|
strategy_id, asset_id, owner_a, daemon_id = _prepare_daemon_fixture(repo)
|
|
owner_b = f"daemon-owner-{uuid.uuid4().hex}"
|
|
portfolio_id = str(uuid.uuid4())
|
|
barrier = threading.Barrier(3)
|
|
outcomes: list[str] = []
|
|
errors: list[BaseException] = []
|
|
with repo._connect().cursor() as cur:
|
|
cur.execute(
|
|
"INSERT INTO users (id, username, permission, is_active) VALUES (%s, %s, 'trade-write', TRUE)",
|
|
(owner_b, owner_b),
|
|
)
|
|
|
|
def assign_daemon() -> None:
|
|
worker = _repo()
|
|
try:
|
|
barrier.wait()
|
|
worker.create_portfolio(
|
|
portfolio_id,
|
|
owner_user_id=owner_a,
|
|
broker="kis",
|
|
status="enabled",
|
|
name="owner-transfer-race",
|
|
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],
|
|
managed_by_daemon_id=daemon_id,
|
|
)
|
|
outcomes.append("assigned")
|
|
except DaemonNotFoundError:
|
|
outcomes.append("assignment-rejected")
|
|
except BaseException as exc: # pragma: no cover - populated only on failure
|
|
errors.append(exc)
|
|
finally:
|
|
worker.close()
|
|
|
|
def transfer_owner() -> None:
|
|
worker = _repo()
|
|
try:
|
|
barrier.wait()
|
|
worker.update_daemon_owner(daemon_id, owner_user_id=owner_b)
|
|
outcomes.append("transferred")
|
|
except DaemonInUseError:
|
|
outcomes.append("transfer-rejected")
|
|
except BaseException as exc: # pragma: no cover - populated only on failure
|
|
errors.append(exc)
|
|
finally:
|
|
worker.close()
|
|
|
|
threads = [threading.Thread(target=assign_daemon), threading.Thread(target=transfer_owner)]
|
|
for thread in threads:
|
|
thread.start()
|
|
barrier.wait()
|
|
for thread in threads:
|
|
thread.join(timeout=10)
|
|
assert not thread.is_alive()
|
|
|
|
assert errors == []
|
|
assert sorted(outcomes) in (
|
|
["assigned", "transfer-rejected"],
|
|
["assignment-rejected", "transferred"],
|
|
)
|
|
daemon = repo.get_daemon_by_id(daemon_id)
|
|
portfolio = repo.get(portfolio_id) if "assigned" in outcomes else None
|
|
if portfolio is not None:
|
|
assert daemon.owner_user_id == portfolio.owner_user_id == owner_a
|
|
assert portfolio.managed_by_daemon_id == daemon_id
|
|
else:
|
|
assert daemon.owner_user_id == owner_b
|
|
|
|
_cleanup(repo, [portfolio_id], strategy_id, asset_id, owner_a, daemon_id, [owner_b])
|
|
|
|
|
|
def test_deactivation_preserves_assignment_and_revokes_bound_service_token() -> None:
|
|
repo = _repo()
|
|
strategy_id, asset_id, owner_user_id, daemon_id = _prepare_daemon_fixture(repo)
|
|
portfolio_id = str(uuid.uuid4())
|
|
try:
|
|
repo.create_portfolio(
|
|
portfolio_id,
|
|
owner_user_id=owner_user_id,
|
|
broker="kis",
|
|
status="enabled",
|
|
name="deactivation-preserves-reference",
|
|
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],
|
|
managed_by_daemon_id=daemon_id,
|
|
)
|
|
token = repo.create_auth_token(
|
|
owner_user_id,
|
|
name=f"daemon-token-{uuid.uuid4().hex}",
|
|
token_hash=uuid.uuid4().hex,
|
|
token_prefix="pfh",
|
|
permission="trade-write",
|
|
token_type="service",
|
|
daemon_id=daemon_id,
|
|
)
|
|
|
|
repo.update_daemon(daemon_id, is_active=False)
|
|
assert repo.revoke_tokens_by_daemon_id(daemon_id) == 1
|
|
|
|
assert repo.get(portfolio_id).managed_by_daemon_id == daemon_id
|
|
with repo._connect().cursor() as cur:
|
|
cur.execute("SELECT revoked_at IS NOT NULL FROM auth_tokens WHERE id = %s", (token.id,))
|
|
assert cur.fetchone()[0] is True
|
|
|
|
_cleanup(repo, [portfolio_id], strategy_id, asset_id, owner_user_id, daemon_id)
|
|
finally:
|
|
pass
|