copilot-swe-agent[bot]andgenichin
591db7adb7
Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
117 lines
3.8 KiB
Python
117 lines
3.8 KiB
Python
"""Daemon buy-symbol resolver regression (dcy-011/REQ-007, REL-004 smoke finding).
|
|
|
|
Exercises the REAL Daemon._resolve_symbol against a REAL SnapshotStore — the gap
|
|
that let `asset.symbol` (wrong; SnapshotStore.get returns a SnapshotAsset whose
|
|
symbol is on `.item.symbol`) ship: unit tests injected a stub resolver, so the
|
|
daemon wiring was never run until the REL-004 live smoke timed out every buy.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
from pathlib import Path
|
|
|
|
from trex_core.application.portfolio import PortfolioItem, SnapshotStore
|
|
from trex_core.runtime.daemon import Daemon
|
|
|
|
|
|
class _StoreHolder:
|
|
"""Minimal stand-in carrying just the attribute Daemon._resolve_symbol uses."""
|
|
|
|
def __init__(self, store):
|
|
self.snapshot_store = store
|
|
|
|
|
|
def _store(*items):
|
|
s = SnapshotStore()
|
|
s.swap(list(items))
|
|
return s
|
|
|
|
|
|
def test_resolve_symbol_reads_item_symbol():
|
|
store = _store(PortfolioItem(portfolio_id="pf-005930", symbol="005930",
|
|
cash_balance=0.0, asset_quantity=0.0))
|
|
holder = _StoreHolder(store)
|
|
# call the real method with a stub self (no full daemon construction needed)
|
|
assert Daemon._resolve_symbol(holder, "pf-005930") == "005930"
|
|
|
|
|
|
def test_resolve_symbol_unknown_returns_none():
|
|
holder = _StoreHolder(_store())
|
|
assert Daemon._resolve_symbol(holder, "pf-missing") is None
|
|
|
|
|
|
def test_daemon_wires_shared_alarm_emitter_into_intent_service():
|
|
src = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "trex_core"
|
|
/ "runtime"
|
|
/ "daemon.py"
|
|
).read_text(encoding="utf-8")
|
|
tree = ast.parse(src)
|
|
found = False
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Assign):
|
|
continue
|
|
if not any(isinstance(t, ast.Attribute) and t.attr == "intent_service" for t in node.targets):
|
|
continue
|
|
call = node.value
|
|
if not (
|
|
isinstance(call, ast.Call)
|
|
and isinstance(call.func, ast.Name)
|
|
and call.func.id == "IntentService"
|
|
):
|
|
continue
|
|
for kw in call.keywords:
|
|
if kw.arg != "alarm_emitter":
|
|
continue
|
|
if (
|
|
isinstance(kw.value, ast.Attribute)
|
|
and kw.value.attr == "alarm_emitter"
|
|
and isinstance(kw.value.value, ast.Name)
|
|
and kw.value.value.id == "self"
|
|
):
|
|
found = True
|
|
break
|
|
assert found, "Daemon must pass self.alarm_emitter into IntentService(alarm_emitter=...)"
|
|
|
|
|
|
def test_daemon_wires_reconciliation_broker_into_snapshot_scheduler():
|
|
src = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "trex_core"
|
|
/ "runtime"
|
|
/ "daemon.py"
|
|
).read_text(encoding="utf-8")
|
|
tree = ast.parse(src)
|
|
found = False
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Assign):
|
|
continue
|
|
if not any(
|
|
isinstance(t, ast.Attribute) and t.attr == "snapshot_scheduler"
|
|
for t in node.targets
|
|
):
|
|
continue
|
|
call = node.value
|
|
if not (
|
|
isinstance(call, ast.Call)
|
|
and isinstance(call.func, ast.Name)
|
|
and call.func.id == "SnapshotScheduler"
|
|
):
|
|
continue
|
|
for kw in call.keywords:
|
|
if kw.arg != "core_holdings_broker":
|
|
continue
|
|
if (
|
|
isinstance(kw.value, ast.Attribute)
|
|
and kw.value.attr == "reconciliation_broker"
|
|
and isinstance(kw.value.value, ast.Name)
|
|
and kw.value.value.id == "self"
|
|
):
|
|
found = True
|
|
break
|
|
assert found, (
|
|
"Daemon must pass self.reconciliation_broker into "
|
|
"SnapshotScheduler(core_holdings_broker=...)"
|
|
)
|