momentum-system/tests/unit/test_operator_commands.py

"""Operator command bus: consumer dedup/staleness, pause on BOTH entry paths,
flatten-does-not-self-restart, and the qabot slash-command carve-out (plan 007)."""

import uuid
from datetime import datetime, timedelta

from runtime.operator_commands import OperatorCommandConsumer
from storage.event_schema import EventMode, EventType, OperatorCommandEvent
from storage.event_store import EventStore
from tests.unit.test_trading_execution import FakeExecutor, emit_signal, make_service


class FakeExecution:
    def __init__(self):
        self.closed = []
        self.flattened = []
        self.paused = None

    def close_session(self, reason="eod_flatten"):
        self.closed.append(reason)
        return {"closed_positions": ["AAA"], "cancelled_entries": 0, "errors": []}

    def flatten_symbol(self, symbol, reason="operator_flatten_symbol"):
        self.flattened.append(symbol)
        return {"ok": True, "symbol": symbol}

    def set_operator_paused(self, paused):
        self.paused = bool(paused)


def _emit_command(store, command, symbol=None, when=None, nonce=None):
    store.emit(OperatorCommandEvent(
        timestamp=when or datetime.now(), mode=EventMode.PAPER,
        message=f"cmd {command}", command=command, symbol=symbol,
        requested_by="test", nonce=nonce or uuid.uuid4().hex))


def test_command_consumed_once_nonce_dedup():
    store = EventStore(":memory:")
    ex = FakeExecution()
    _emit_command(store, "flatten_all")
    consumer = OperatorCommandConsumer(store, ex)
    assert len(consumer.poll()) == 1
    assert ex.closed == ["operator_flatten"]
    # second poll: already-consumed nonce -> no re-execution
    assert consumer.poll() == []
    assert ex.closed == ["operator_flatten"]
    store.close()


def test_done_event_prevents_reexecution_after_restart():
    store = EventStore(":memory:")
    _emit_command(store, "flatten_all")
    OperatorCommandConsumer(store, FakeExecution()).poll()  # process A executes + echoes _done
    ex2 = FakeExecution()
    # process B boots fresh, reloads the _done nonce, must NOT re-run
    assert OperatorCommandConsumer(store, ex2).poll() == []
    assert ex2.closed == []
    store.close()


def test_stale_command_ignored_on_boot():
    store = EventStore(":memory:")
    ex = FakeExecution()
    _emit_command(store, "flatten_all", when=datetime.now() - timedelta(minutes=20))
    _emit_command(store, "flatten_symbol", symbol="AAA")  # fresh
    consumer = OperatorCommandConsumer(store, ex, stale_seconds=600)
    consumer.poll()
    assert ex.closed == []          # the 20-min-old flatten_all was NOT run
    assert ex.flattened == ["AAA"]  # the fresh one was
    store.close()


def test_flatten_all_routes_through_close_session_not_self_restart():
    import run_live_paper
    run_live_paper._clean_exit_requested.clear()
    store = EventStore(":memory:")
    ex = FakeExecution()
    _emit_command(store, "flatten_all")
    OperatorCommandConsumer(store, ex).poll()
    assert ex.closed == ["operator_flatten"]
    # the operator flatten must NOT trip plan 003's daily-restart exit
    assert not run_live_paper._clean_exit_requested.is_set()
    store.close()


def test_pause_resume_dispatch():
    store = EventStore(":memory:")
    ex = FakeExecution()
    _emit_command(store, "pause_entries")
    OperatorCommandConsumer(store, ex).poll()
    assert ex.paused is True
    _emit_command(store, "resume_entries")
    OperatorCommandConsumer(store, ex).poll()
    assert ex.paused is False
    store.close()


def test_done_echo_records_result():
    store = EventStore(":memory:")
    _emit_command(store, "flatten_symbol", symbol="BBB")
    OperatorCommandConsumer(store, FakeExecution()).poll()
    done = store.query_events(event_type=EventType.OPERATOR_COMMAND_DONE.value, limit=None)
    assert len(done) == 1


# ---- pause blocks BOTH entry paths (the drift class plan 005 warns about) ----
def test_pause_blocks_both_entry_paths():
    store = EventStore(":memory:")
    emit_signal(store)  # a would-otherwise-pass ready signal
    svc = make_service(store, FakeExecutor())
    svc.set_operator_paused(True)
    assert svc.request_approvals_for_ready_signals() == []      # SLOW path blocked
    fast = svc.submit_breakout_now("GOOD", 14.0, 13.45, last_price=14.1)
    assert fast.get("skipped") == "paused"                       # FAST path blocked
    # resume re-enables the slow path (the same signal now creates a request)
    svc.set_operator_paused(False)
    assert len(svc.request_approvals_for_ready_signals()) == 1
    store.close()


# ---- qabot slash-command carve-out + auth ----
def test_qabot_parse_flatten_command():
    from telegram_bot import parse_flatten_command
    assert parse_flatten_command("/flatten") == ("flatten_all", None)
    assert parse_flatten_command("/flatten nvda") == ("flatten_symbol", "NVDA")
    assert parse_flatten_command("please flatten everything") is None  # prose still refuses
    assert parse_flatten_command("how did we do?") is None


def test_qabot_unauthorized_chat_refused():
    from telegram_bot import is_authorized
    assert is_authorized("999", "123") is False
    assert is_authorized("123", "123") is True
    assert is_authorized("123", None) is False  # fail closed when unconfigured


if __name__ == "__main__":  # pragma: no cover
    import pytest
    pytest.main([__file__, "-q"])