trader-mcp/tests/test_engine.py

from __future__ import annotations

from pathlib import Path
from tempfile import TemporaryDirectory

from fastapi.testclient import TestClient

from src.trader_mcp import strategy_registry, strategy_store, strategy_engine
from src.trader_mcp.server import app


STRATEGY_CODE = '''
from src.trader_mcp.strategy_sdk import Strategy

class Strategy(Strategy):
    CONFIG_SCHEMA = {"label": {"type": "string", "default": "hello world"}}

    def init(self):
        return {"started": True, "config_copy": dict(self.config)}

    def on_tick(self, tick):
        self.state["ticks"] = self.state.get("ticks", 0) + 1
        return self.state["ticks"]

    def render(self):
        return {"widgets": [{"type": "metric", "label": "ticks", "value": self.state.get("ticks", 0)}]}
'''


def test_mode_off_does_not_instantiate_and_active_does(tmp_path):
    original_db = strategy_store.DB_PATH
    original_dir = strategy_registry.STRATEGIES_DIR
    try:
        strategy_store.DB_PATH = tmp_path / "trader_mcp.sqlite3"
        strategy_registry.STRATEGIES_DIR = tmp_path / "strategies"
        strategy_registry.STRATEGIES_DIR.mkdir()
        (strategy_registry.STRATEGIES_DIR / "demo.py").write_text(STRATEGY_CODE)

        strategy_store.add_strategy_instance(id="s1", strategy_type="demo", account_id="acct-1", client_id="cid-1", mode="off", config={"x": 1})
        result = strategy_engine.reconcile_all()
        assert result["running"] == []
        assert strategy_engine.get_running_strategy("s1") is None

        strategy_store.update_strategy_mode("s1", "active")
        result = strategy_engine.reconcile_all()
        assert "s1" in result["running"]
        runtime = strategy_engine.get_running_strategy("s1")
        assert runtime is not None
        assert runtime.instance.state["started"] is True
        assert runtime.instance.context.account_id == "acct-1"
        assert runtime.instance.context.client_id == "cid-1"
        tick_result = strategy_engine.tick_strategy("s1", {"price": 1})
        assert tick_result["ok"] is True
        assert tick_result["result"] == 1
        render_result = strategy_engine.render_strategy("s1")
        assert render_result["ok"] is True
        assert render_result["render"]["widgets"][0]["value"] == 1
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_mode_change_route_reconciles(tmp_path):
    original_db = strategy_store.DB_PATH
    original_dir = strategy_registry.STRATEGIES_DIR
    try:
        strategy_store.DB_PATH = tmp_path / "trader_mcp.sqlite3"
        strategy_registry.STRATEGIES_DIR = tmp_path / "strategies"
        strategy_registry.STRATEGIES_DIR.mkdir()
        (strategy_registry.STRATEGIES_DIR / "demo.py").write_text(STRATEGY_CODE)

        client = TestClient(app)
        client.post(
            "/strategies",
            json={"id": "s2", "strategy_type": "demo", "account_id": "acct-1", "client_id": "cid-1", "mode": "off", "config": {"x": 1}},
        )
        r = client.post("/strategies/s2/mode", json={"mode": "active"})
        assert r.status_code == 200
        assert "s2" in r.json()["running"]
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_dashboard_add_strategy_synthesizes_client_and_defaults(tmp_path):
    original_db = strategy_store.DB_PATH
    original_dir = strategy_registry.STRATEGIES_DIR
    try:
        strategy_store.DB_PATH = tmp_path / "trader_mcp.sqlite3"
        strategy_registry.STRATEGIES_DIR = tmp_path / "strategies"
        strategy_registry.STRATEGIES_DIR.mkdir()
        (strategy_registry.STRATEGIES_DIR / "hello_world.py").write_text(STRATEGY_CODE)

        client = TestClient(app)
        response = client.post(
            "/dashboard/strategies/add",
            data={"name": "My super Grid 0.5", "strategy_type": "hello_world", "account_id": "acct-1"},
            follow_redirects=False,
        )
        assert response.status_code in {302, 303}

        record = strategy_store.list_strategy_instances()[0]
        assert record.name == "My super Grid 0.5"
        assert record.client_id.startswith("hello_world:")
        assert record.config == {"label": "hello world"}
        assert record.mode == "off"
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_runtime_pause_suppresses_tick_and_render(tmp_path):
    original_db = strategy_store.DB_PATH
    original_dir = strategy_registry.STRATEGIES_DIR
    try:
        strategy_store.DB_PATH = tmp_path / "trader_mcp.sqlite3"
        strategy_registry.STRATEGIES_DIR = tmp_path / "strategies"
        strategy_registry.STRATEGIES_DIR.mkdir()
        (strategy_registry.STRATEGIES_DIR / "hello_world.py").write_text(STRATEGY_CODE)

        strategy_store.add_strategy_instance(id="s3", strategy_type="hello_world", account_id="acct-1", client_id="cid-1", mode="active", config={})
        strategy_engine.reconcile_all()

        assert strategy_engine.pause_strategy("s3")["ok"] is True
        tick_result = strategy_engine.tick_strategy("s3", {"price": 1})
        assert tick_result["paused"] is True
        assert tick_result["skipped"] is True
        render_result = strategy_engine.render_strategy("s3")
        assert render_result["paused"] is True
        assert render_result["render"] is None

        assert strategy_engine.resume_strategy("s3")["ok"] is True
        tick_result = strategy_engine.tick_strategy("s3", {"price": 1})
        assert tick_result["result"] == 1
        render_result = strategy_engine.render_strategy("s3")
        assert render_result["render"]["widgets"][0]["value"] == 1
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()


def test_dashboard_pause_toggle(tmp_path):
    original_db = strategy_store.DB_PATH
    original_dir = strategy_registry.STRATEGIES_DIR
    try:
        strategy_store.DB_PATH = tmp_path / "trader_mcp.sqlite3"
        strategy_registry.STRATEGIES_DIR = tmp_path / "strategies"
        strategy_registry.STRATEGIES_DIR.mkdir()
        (strategy_registry.STRATEGIES_DIR / "hello_world.py").write_text(STRATEGY_CODE)

        client = TestClient(app)
        client.post(
            "/dashboard/strategies/add",
            data={"name": "Pause test", "strategy_type": "hello_world", "account_id": "acct-1"},
            follow_redirects=False,
        )
        strategy_id = strategy_store.list_strategy_instances()[0].id

        strategy_store.update_strategy_mode(strategy_id, "active")
        strategy_engine.reconcile_instance(strategy_id)
        assert strategy_engine.get_running_strategy(strategy_id).paused is False

        client.post(f"/dashboard/strategies/{strategy_id}/pause", follow_redirects=False)
        assert strategy_engine.get_running_strategy(strategy_id).paused is True

        client.post(f"/dashboard/strategies/{strategy_id}/pause", follow_redirects=False)
        assert strategy_engine.get_running_strategy(strategy_id).paused is False
    finally:
        strategy_store.DB_PATH = original_db
        strategy_registry.STRATEGIES_DIR = original_dir
        strategy_engine._running.clear()