momentum devandClaude Fable 5
e7c34f298f
Adds OperatorCommandEvent + a closed command vocabulary (flatten_all, flatten_symbol, pause_entries, resume_entries). POST /api/command emits it (token-required: 503 when DASHBOARD_TOKEN unset — commands never run unauthenticated); the live loop's new 5s "command" step consumes them via OperatorCommandConsumer with three replay-safety layers (in-session nonce set, persisted operator_command_done reload, 10-min staleness window). flatten_all routes through close_session DIRECTLY (never step_eod_flatten) so it cannot trip plan 003's daily-restart exit. Pause is checked at the top of BOTH request_approvals_for_ready_signals AND submit_breakout_now (missing either leaks pause on that path). qabot gains an authorized-chat-only /flatten slash command (prose "flatten" still refuses via _ACTION_WORDS). Snapshot exposes broker truth + command_bus + a defensive experiment ledger. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
"""POST /api/command — token gate, enum validation, event emission (plan 007).
|
|
|
|
Drives DashboardHandler.do_POST directly (no loopback socket), against a real
|
|
DashboardState pinned to an isolated Postgres schema so emit_command actually
|
|
lands an event."""
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import uuid
|
|
|
|
import psycopg2
|
|
import pytest
|
|
|
|
from api.main import DashboardHandler, DashboardState
|
|
from storage.event_schema import EventType
|
|
from storage.event_store import EventStore
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolated_pg_schema(monkeypatch):
|
|
schema = f"mem_{uuid.uuid4().hex[:12]}"
|
|
monkeypatch.setenv("MOMENTUM_PG_SCHEMA", schema)
|
|
yield
|
|
url = os.environ.get("DATABASE_URL")
|
|
if url:
|
|
conn = psycopg2.connect(url)
|
|
conn.autocommit = True
|
|
conn.cursor().execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
|
|
conn.close()
|
|
|
|
|
|
class CaptureHandler(DashboardHandler):
|
|
def __init__(self, state, path, body=b"", headers=None):
|
|
self.state = state
|
|
self.path = path
|
|
self.rfile = io.BytesIO(body)
|
|
self.headers = headers or {}
|
|
self.captured = []
|
|
|
|
def _send_json(self, payload, status=200):
|
|
self.captured.append((status, payload))
|
|
|
|
|
|
def _post(state, body, headers=None):
|
|
raw = json.dumps(body).encode()
|
|
hdrs = {"Content-Length": str(len(raw))}
|
|
if headers:
|
|
hdrs.update(headers)
|
|
h = CaptureHandler(state, "/api/command", raw, hdrs)
|
|
h.do_POST()
|
|
return h.captured[-1]
|
|
|
|
|
|
def test_command_503_when_token_unset(monkeypatch):
|
|
monkeypatch.delenv("DASHBOARD_TOKEN", raising=False)
|
|
status, payload = _post(DashboardState("momentum"), {"command": "flatten_all"})
|
|
assert status == 503
|
|
assert payload["ok"] is False
|
|
|
|
|
|
def test_command_unknown_rejected(monkeypatch):
|
|
monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret")
|
|
status, payload = _post(DashboardState("momentum"), {"command": "sell_everything"},
|
|
headers={"X-Dashboard-Token": "s3cret"})
|
|
assert status == 400
|
|
assert "unknown command" in payload["error"]
|
|
|
|
|
|
def test_command_flatten_symbol_requires_symbol(monkeypatch):
|
|
monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret")
|
|
status, payload = _post(DashboardState("momentum"), {"command": "flatten_symbol"},
|
|
headers={"X-Dashboard-Token": "s3cret"})
|
|
assert status == 400
|
|
|
|
|
|
def test_command_bad_token_401(monkeypatch):
|
|
monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret")
|
|
status, _ = _post(DashboardState("momentum"), {"command": "flatten_all"},
|
|
headers={"X-Dashboard-Token": "wrong"})
|
|
assert status == 401
|
|
|
|
|
|
def test_command_valid_emits_event(monkeypatch):
|
|
monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret")
|
|
state = DashboardState("momentum")
|
|
status, payload = _post(state, {"command": "flatten_all"},
|
|
headers={"X-Dashboard-Token": "s3cret"})
|
|
assert status == 200 and payload["ok"] is True and payload["nonce"]
|
|
store = EventStore("momentum")
|
|
try:
|
|
events = store.query_events(event_type=EventType.OPERATOR_COMMAND.value, limit=None)
|
|
assert len(events) == 1
|
|
assert json.loads(events[0]["payload_json"])["command"] == "flatten_all"
|
|
finally:
|
|
store.close()
|
|
|
|
|
|
def test_command_emits_even_without_execution_service(monkeypatch):
|
|
# the standalone dashboard (execution_service=None) is a thin emitter
|
|
monkeypatch.setenv("DASHBOARD_TOKEN", "s3cret")
|
|
state = DashboardState("momentum", execution_service=None)
|
|
status, payload = _post(state, {"command": "pause_entries"},
|
|
headers={"X-Dashboard-Token": "s3cret"})
|
|
assert status == 200 and payload["ok"] is True
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
pytest.main([__file__, "-q"])
|