"""Operator CLI tests (dcy-016 / REQ-032) — RPC injected, no broker/DB needed.""" from __future__ import annotations import json import pytest from trex_core.runtime import cli @pytest.fixture(autouse=True) def _env(monkeypatch): monkeypatch.setenv("TREX_POSTGRESQL_DSN", "postgresql://x/y") monkeypatch.setenv("TREX_BROKER", "paper") monkeypatch.delenv("TREX_KIS_ALLOW_LIVE", raising=False) class _RpcRecorder: def __init__(self, response): self.response = response self.calls = [] def __call__(self, queue, payload, *, correlation_id, timeout): self.calls.append({"queue": queue, "payload": payload, "cid": correlation_id}) return self.response _SNAPSHOT = { "assets": [{"asset_id": "PF-1", "symbol": "005930", "asset_quantity": 0, "active_strategy_id": "orb"}], "view": "all", "available_slots": 9, "degraded": False, "reason": None, } _ADMITTED = {"intent_id": "i", "status": "admitted", "slot_id": "S-1", "reason": None} _REJECTED = {"intent_id": "i", "status": "rejected", "slot_id": None, "reason": "unknown_portfolio"} # --- universe / slots (snapshot RPC + watchdog suppression) ------------------ def test_universe_uses_snapshot_queue_with_operator_suppression(capsys): rpc = _RpcRecorder(_SNAPSHOT) rc = cli.main(["universe"], rpc=rpc) assert rc == 0 call = rpc.calls[0] assert call["queue"] == "trex.snapshot.req" # default instance assert call["payload"]["actor"] == "operator" assert call["payload"]["touch_watchdog"] is False # FR-2a / R-4 assert "PF-1" in capsys.readouterr().out def test_slots_status_holding_maps_to_held_view(): rpc = _RpcRecorder(_SNAPSHOT) cli.main(["slots", "--status", "holding"], rpc=rpc) assert rpc.calls[0]["payload"]["view"] == "held" def test_slots_marks_watchdog_owned(capsys): """dcy-017: slots flags strategy-absent (watchdog-owned) positions.""" resp = { "assets": [{"asset_id": "PF-1", "symbol": "005930", "asset_quantity": 10, "active_strategy_id": "orb"}], "view": "held", "available_slots": 9, "degraded": False, "reason": None, "watchdog": [{"slot_id": "S-1", "portfolio_id": "PF-1", "symbol": "005930", "expired": True, "last_seen": None}], } cli.main(["slots", "--status", "held"], rpc=_RpcRecorder(resp)) out = capsys.readouterr().out assert "watchdog_owned=1" in out and "WATCHDOG-OWNED" in out def test_instance_shortcut_routes_to_instance_queue(): rpc = _RpcRecorder(_SNAPSHOT) cli.main(["universe", "--instance", "inst-2"], rpc=rpc) assert rpc.calls[0]["queue"] == "trex.snapshot.req.inst-2" def test_json_output_is_valid_json(capsys): rpc = _RpcRecorder(_SNAPSHOT) cli.main(["universe", "--json"], rpc=rpc) parsed = json.loads(capsys.readouterr().out) assert parsed["assets"][0]["asset_id"] == "PF-1" # --- enter (buy intent RPC) + safeguards ------------------------------------ def test_enter_aborts_without_confirmation_no_rpc(): rpc = _RpcRecorder(_ADMITTED) rc = cli.main(["enter", "PF-1", "--capital", "100000"], rpc=rpc, confirm=lambda _p: False) assert rc == 1 and rpc.calls == [] # aborted -> no order def test_enter_with_yes_submits_buy_intent(): rpc = _RpcRecorder(_ADMITTED) rc = cli.main(["enter", "PF-1", "--capital", "100000", "--yes"], rpc=rpc) assert rc == 0 p = rpc.calls[0]["payload"] assert rpc.calls[0]["queue"] == "trex.intent.req" assert p["intent_type"] == "buy" and p["portfolio_id"] == "PF-1" assert p["allocated_capital"] == 100000 assert "symbol" not in p # server-resolved (dcy-011) def test_enter_rejected_returns_nonzero(): rpc = _RpcRecorder(_REJECTED) rc = cli.main(["enter", "PF-x", "--yes"], rpc=rpc) assert rc == 1 def test_enter_dry_run_does_not_submit(capsys): rpc = _RpcRecorder(_ADMITTED) rc = cli.main(["enter", "PF-1", "--capital", "5", "--dry-run"], rpc=rpc) assert rc == 0 and rpc.calls == [] assert "dry-run" in capsys.readouterr().out def test_enter_json_requires_yes(): rpc = _RpcRecorder(_ADMITTED) rc = cli.main(["enter", "PF-1", "--json"], rpc=rpc) assert rc == 2 and rpc.calls == [] # no auto-submit (FR-7) # --- exit (sell intent RPC) ------------------------------------------------- def test_exit_with_yes_submits_sell_with_limit_price(): rpc = _RpcRecorder(_ADMITTED) cli.main(["exit", "S-1", "--price", "70000", "--yes"], rpc=rpc) p = rpc.calls[0]["payload"] assert p["intent_type"] == "sell" and p["slot_id"] == "S-1" assert p["limit_price"] == 70000 def test_exit_market_when_no_price(): rpc = _RpcRecorder(_ADMITTED) cli.main(["exit", "S-1", "--yes"], rpc=rpc) assert rpc.calls[0]["payload"]["limit_price"] is None # --- live guard (OQ-2 double ack) ------------------------------------------- def test_live_requires_live_flag(monkeypatch): monkeypatch.setattr(cli, "_broker_mode", lambda cfg: ("kis", True)) rpc = _RpcRecorder(_ADMITTED) rc = cli.main(["enter", "PF-1", "--yes"], rpc=rpc) # --yes but no --live assert rc == 2 and rpc.calls == [] def test_live_proceeds_with_yes_live(monkeypatch): monkeypatch.setattr(cli, "_broker_mode", lambda cfg: ("kis", True)) rpc = _RpcRecorder(_ADMITTED) rc = cli.main(["enter", "PF-1", "--yes", "--live"], rpc=rpc) assert rc == 0 and len(rpc.calls) == 1 # --- transport failure surfaces as a clear error ---------------------------- def test_rpc_failure_surfaces_error(capsys): def boom(*a, **k): raise RuntimeError("daemon unreachable") rc = cli.main(["universe"], rpc=boom) assert rc == 3 assert "daemon unreachable" in capsys.readouterr().err