101 lines
2.9 KiB
Python
101 lines
2.9 KiB
Python
"""Shared test fixtures.
|
|
|
|
Infrastructure integration tests are gated on env vars:
|
|
- TREX_POSTGRESQL_TEST_DSN : PostgreSQL repository/outbox/migration tests
|
|
- TREX_RABBITMQ_TEST_URL : RabbitMQ transport + intent RPC E2E tests
|
|
- TREX_KIS_BROKER_* : KIS sandbox round-trip test
|
|
|
|
When a gate is unset the corresponding tests are skipped; unit tests
|
|
(in-memory adapters, stub transport) always run.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from trex_core.adapters.broker.kis import KISBroker
|
|
|
|
_PG_TABLES = (
|
|
"broker_account_snapshot",
|
|
"ledger_skip",
|
|
"killswitch_exclusion",
|
|
"snapshot_pause",
|
|
"ledger_outbox",
|
|
"intent_log",
|
|
"slot_transition_log",
|
|
"daily_symbol_entries",
|
|
"daily_slot",
|
|
"reconcile_notification_state",
|
|
"slot_strategy_state",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def kis_balance_broker_factory(monkeypatch):
|
|
"""Build a real KISBroker backed by deterministic synthetic balance pages."""
|
|
|
|
def factory(responses):
|
|
broker = KISBroker(
|
|
"12345678-01",
|
|
app_key="test-app-key",
|
|
app_secret="test-app-secret",
|
|
base_url="https://openapivts.kis.test",
|
|
)
|
|
monkeypatch.setattr(broker, "_headers", lambda _tr: {})
|
|
pending = list(responses)
|
|
calls = []
|
|
|
|
class Response:
|
|
def __init__(self, page):
|
|
self._page = page
|
|
self.headers = {"tr_cont": page.get("tr_cont", "E")}
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {
|
|
"rt_cd": self._page.get("rt_cd", "0"),
|
|
"output1": self._page.get("rows", []),
|
|
"ctx_area_fk100": self._page.get("fk", ""),
|
|
"ctx_area_nk100": self._page.get("nk", ""),
|
|
}
|
|
|
|
def fake_get(*_args, **kwargs):
|
|
calls.append({
|
|
"headers": dict(kwargs["headers"]),
|
|
"params": dict(kwargs["params"]),
|
|
})
|
|
response = pending.pop(0)
|
|
if isinstance(response, BaseException):
|
|
raise response
|
|
return Response(response)
|
|
|
|
monkeypatch.setattr("trex_core.adapters.broker.kis.httpx.get", fake_get)
|
|
return broker, calls
|
|
|
|
return factory
|
|
|
|
|
|
@pytest.fixture
|
|
def pg_dsn() -> str:
|
|
dsn = os.environ.get("TREX_POSTGRESQL_TEST_DSN")
|
|
if not dsn:
|
|
pytest.skip("TREX_POSTGRESQL_TEST_DSN not set; skipping PostgreSQL integration test")
|
|
return dsn
|
|
|
|
|
|
@pytest.fixture
|
|
def pg_clean(pg_dsn: str):
|
|
"""Apply migrations and truncate slot-core tables before each PG test."""
|
|
import psycopg
|
|
|
|
from trex_core.adapters.persistence.migrate import apply_migrations
|
|
|
|
apply_migrations(pg_dsn)
|
|
with psycopg.connect(pg_dsn, autocommit=True) as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(f"TRUNCATE {', '.join(_PG_TABLES)}")
|
|
return pg_dsn
|