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()