test(reconciliation): verify repository lookup parity

2026-07-15 08:18:32 +09:00
parent 68dd28cf12
commit e95bd4f8b2
2 changed files with 200 additions and 2 deletions
@@ -0,0 +1,198 @@
+"""REQ-121 AC-5 fake-driver parity coverage for managed-slot lookup.
+
+This is a unit contract test: the injected psycopg fake exercises the actual
+PostgreSQL adapter query and row mapping without requiring a PostgreSQL server.
+"""
+from __future__ import annotations
+
+import sys
+import types
+from datetime import datetime, timezone
+from decimal import Decimal
+from uuid import UUID
+
+# This worktree lacks pika, but importing the PostgreSQL adapter traverses the
+# application alarm-emitter import path. Install only the import-time minimum;
+# this test exercises neither RabbitMQ nor a real PostgreSQL connection.
+_pika = types.ModuleType("pika")
+_pika.__path__ = []
+_pika.URLParameters = lambda *args, **kwargs: None
+_pika.BlockingConnection = object
+_pika.BasicProperties = object
+_pika_exceptions = types.ModuleType("pika.exceptions")
+_pika_exceptions.AMQPError = Exception
+_pika.exceptions = _pika_exceptions
+sys.modules.setdefault("pika", _pika)
+sys.modules.setdefault("pika.exceptions", _pika_exceptions)
+
+from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository  # noqa: E402
+from trex_core.domain.slot import Slot, SlotState  # noqa: E402
+from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository  # noqa: E402
+
+
+_SLOT_COLUMNS = (
+    "slot_id, status, symbol, qty, avg_price, limit_price, "
+    "order_id, strategy_id, broker_order_id, portfolio_id, "
+    "sell_strategy_id, forced_close_at, stop_loss_price, take_profit_price, "
+    "created_at, updated_at, instance_key, retired_at, "
+    "timeout_sec, slippage_pct, occupancy_seq"
+)
+_MANAGED_SLOT_SQL = (
+    f"SELECT {_SLOT_COLUMNS} FROM daily_slot "
+    "WHERE instance_key = %s AND symbol = %s "
+    "ORDER BY updated_at DESC NULLS LAST, slot_id ASC LIMIT 1"
+)
+
+
+class _FakeCursor:
+    def __init__(self, rows: list[tuple], executions: list[tuple[str, tuple[str, str]]]) -> None:
+        self._rows = rows
+        self._executions = executions
+        self._row: tuple | None = None
+
+    def __enter__(self) -> _FakeCursor:
+        return self
+
+    def __exit__(self, *args: object) -> None:
+        return None
+
+    def execute(self, query: str, params: tuple[str, str]) -> None:
+        self._executions.append((query, params))
+        instance_key, symbol = params
+        candidates = [
+            row for row in self._rows if row[16] == instance_key and row[2] == symbol
+        ]
+        candidates.sort(key=lambda row: (
+            row[15] is None,
+            -row[15].timestamp() if row[15] is not None else 0,
+            str(UUID(str(row[0]))),
+        ))
+        self._row = candidates[0] if candidates else None
+
+    def fetchone(self) -> tuple | None:
+        return self._row
+
+
+class _FakeConnection:
+    def __init__(self, rows: list[tuple], executions: list[tuple[str, tuple[str, str]]]) -> None:
+        self._rows = rows
+        self._executions = executions
+
+    def __enter__(self) -> _FakeConnection:
+        return self
+
+    def __exit__(self, *args: object) -> None:
+        return None
+
+    def cursor(self) -> _FakeCursor:
+        return _FakeCursor(self._rows, self._executions)
+
+
+class _FakePsycopg:
+    def __init__(self, rows: list[tuple]) -> None:
+        self._rows = rows
+        self.connect_calls: list[tuple[str, dict[str, object]]] = []
+        self.executions: list[tuple[str, tuple[str, str]]] = []
+
+    def connect(self, dsn: str, **kwargs: object) -> _FakeConnection:
+        self.connect_calls.append((dsn, kwargs))
+        return _FakeConnection(self._rows, self.executions)
+
+
+def _slot(slot_id: str, instance_key: str, symbol: str, updated_at: datetime | None) -> Slot:
+    return Slot(
+        slot_id=UUID(slot_id),
+        status=SlotState.HOLDING,
+        symbol=symbol,
+        qty=10.5,
+        avg_price=101.25,
+        limit_price=100.0,
+        order_id="order-1",
+        strategy_id="buy-v1",
+        broker_order_id="broker-1",
+        portfolio_id="portfolio-1",
+        sell_strategy_id="sell-v1",
+        created_at=datetime(2026, 7, 14, tzinfo=timezone.utc),
+        updated_at=updated_at,
+        instance_key=instance_key,
+        timeout_sec=30,
+        slippage_pct=1.5,
+        occupancy_seq=3,
+    )
+
+
+def _row(slot: Slot) -> tuple:
+    """Return a driver-shaped row, including values converted by _row_to_slot."""
+    return (
+        str(slot.slot_id),
+        slot.status.value,
+        slot.symbol,
+        Decimal(str(slot.qty)),
+        Decimal(str(slot.avg_price)),
+        Decimal(str(slot.limit_price)),
+        slot.order_id,
+        slot.strategy_id,
+        slot.broker_order_id,
+        slot.portfolio_id,
+        slot.sell_strategy_id,
+        slot.forced_close_at,
+        None,
+        None,
+        slot.created_at,
+        slot.updated_at,
+        slot.instance_key,
+        slot.retired_at,
+        slot.timeout_sec,
+        Decimal(str(slot.slippage_pct)),
+        slot.occupancy_seq,
+    )
+
+
+def test_find_managed_slot_by_symbol_postgresql_inmemory_parity_with_fake_driver():
+    newest = datetime(2026, 7, 15, 12, tzinfo=timezone.utc)
+    older = datetime(2026, 7, 15, 11, tzinfo=timezone.utc)
+    slots = [
+        # Same newest timestamp: UUID ASC must pick the first fixture slot.
+        _slot("00000000-0000-0000-0000-000000000001", "instance-a", "AAA", newest),
+        _slot("00000000-0000-0000-0000-000000000002", "instance-a", "AAA", newest),
+        _slot("00000000-0000-0000-0000-000000000003", "instance-a", "AAA", older),
+        _slot("00000000-0000-0000-0000-000000000004", "instance-a", "AAA", None),
+        # Same symbol in a different instance must never leak into instance-a.
+        _slot("00000000-0000-0000-0000-000000000005", "instance-b", "AAA", newest),
+        # Same instance with another symbol must not satisfy an AAA query.
+        _slot("00000000-0000-0000-0000-000000000006", "instance-a", "BBB", newest),
+    ]
+    in_memory = InMemorySlotRepository()
+    for slot in slots:
+        in_memory.add_slot(slot)
+
+    fake_psycopg = _FakePsycopg([_row(slot) for slot in slots])
+    postgresql = PostgreSQLSlotRepository("postgresql://fake", psycopg_module=fake_psycopg)
+    lookups = [
+        ("instance-a", "AAA"),
+        ("instance-b", "AAA"),
+        ("instance-a", "BBB"),
+        ("instance-a", "missing"),
+        ("missing-instance", "AAA"),
+    ]
+
+    for instance_key, symbol in lookups:
+        postgres_slot = postgresql.find_managed_slot_by_symbol(instance_key, symbol)
+        in_memory_slot = in_memory.find_managed_slot_by_symbol(instance_key, symbol)
+        assert (postgres_slot.slot_id if postgres_slot else None) == (
+            in_memory_slot.slot_id if in_memory_slot else None
+        )
+
+    selected = postgresql.find_managed_slot_by_symbol("instance-a", "AAA")
+    assert selected is not None
+    assert selected.slot_id == UUID("00000000-0000-0000-0000-000000000001")
+    assert selected.qty == 10.5
+    assert selected.avg_price == 101.25
+    assert selected.slippage_pct == 1.5
+    assert selected.occupancy_seq == 3
+
+    expected_executions = [
+        (_MANAGED_SLOT_SQL, lookup) for lookup in [*lookups, ("instance-a", "AAA")]
+    ]
+    assert fake_psycopg.executions == expected_executions
+    assert fake_psycopg.connect_calls == [("postgresql://fake", {})] * len(expected_executions)