# -*- coding: utf-8 -*- """Tests for the extracted runtime service app surface.""" import json from pathlib import Path from fastapi.testclient import TestClient from backend.api import runtime as runtime_module from backend.apps.runtime_service import create_app def test_runtime_service_routes_are_exposed(): app = create_app() paths = {route.path for route in app.routes} assert "/health" in paths assert "/api/status" in paths assert "/api/runtime/start" in paths assert "/api/runtime/stop" in paths assert "/api/runtime/cleanup" in paths assert "/api/runtime/history" in paths assert "/api/runtime/current" in paths assert "/api/runtime/gateway/port" in paths def test_runtime_service_health_and_status(monkeypatch): runtime_state = runtime_module.get_runtime_state() runtime_state.gateway_process = None runtime_state.gateway_port = 9876 runtime_state.runtime_manager = object() with TestClient(create_app()) as client: health_response = client.get("/health") status_response = client.get("/api/status") assert health_response.status_code == 200 assert health_response.json() == { "status": "healthy", "service": "runtime-service", "gateway_running": False, "gateway_port": 9876, } assert status_response.status_code == 200 assert status_response.json() == { "status": "operational", "service": "runtime-service", "runtime": { "gateway_running": False, "gateway_port": 9876, "has_runtime_manager": True, }, } def test_runtime_service_gateway_port_endpoint_uses_runtime_router(monkeypatch): runtime_module.get_runtime_state().gateway_port = 9345 monkeypatch.setattr(runtime_module, "_is_gateway_running", lambda: True) with TestClient(create_app()) as client: response = client.get( "/api/runtime/gateway/port", headers={"host": "runtime.example:8003", "x-forwarded-proto": "https"}, ) assert response.status_code == 200 assert response.json() == { "port": 9345, "is_running": True, "ws_url": "wss://runtime.example:9345", } def test_runtime_service_get_runtime_config(monkeypatch, tmp_path): run_dir = tmp_path / "runs" / "demo" state_dir = run_dir / "state" state_dir.mkdir(parents=True) (run_dir / "BOOTSTRAP.md").write_text( "---\n" "tickers:\n" " - AAPL\n" "schedule_mode: intraday\n" "interval_minutes: 30\n" "trigger_time: '10:00'\n" "max_comm_cycles: 3\n" "enable_memory: true\n" "---\n", encoding="utf-8", ) (state_dir / "runtime_state.json").write_text( json.dumps( { "context": { "config_name": "demo", "run_dir": str(run_dir), "bootstrap_values": { "tickers": ["AAPL"], "schedule_mode": "intraday", "interval_minutes": 30, "trigger_time": "10:00", "max_comm_cycles": 3, "enable_memory": True, }, } } ), encoding="utf-8", ) monkeypatch.setattr(runtime_module, "PROJECT_ROOT", tmp_path) monkeypatch.setattr(runtime_module, "_is_gateway_running", lambda: True) runtime_module.get_runtime_state().gateway_port = 8765 with TestClient(create_app()) as client: response = client.get("/api/runtime/config") assert response.status_code == 200 payload = response.json() assert payload["run_id"] == "demo" assert payload["bootstrap"]["schedule_mode"] == "intraday" assert payload["resolved"]["interval_minutes"] == 30 assert payload["resolved"]["enable_memory"] is True def test_runtime_service_update_runtime_config_persists_bootstrap(monkeypatch, tmp_path): run_dir = tmp_path / "runs" / "demo" state_dir = run_dir / "state" state_dir.mkdir(parents=True) (run_dir / "BOOTSTRAP.md").write_text( "---\n" "tickers:\n" " - AAPL\n" "schedule_mode: daily\n" "interval_minutes: 60\n" "trigger_time: '09:30'\n" "max_comm_cycles: 2\n" "---\n", encoding="utf-8", ) (state_dir / "runtime_state.json").write_text( json.dumps( { "context": { "config_name": "demo", "run_dir": str(run_dir), "bootstrap_values": { "tickers": ["AAPL"], "schedule_mode": "daily", "interval_minutes": 60, "trigger_time": "09:30", "max_comm_cycles": 2, }, } } ), encoding="utf-8", ) class _DummyContext: def __init__(self): self.bootstrap_values = { "tickers": ["AAPL"], "schedule_mode": "daily", "interval_minutes": 60, "trigger_time": "09:30", "max_comm_cycles": 2, } class _DummyManager: def __init__(self): self.config_name = "demo" self.bootstrap = dict(_DummyContext().bootstrap_values) self.context = _DummyContext() def _persist_snapshot(self): return None monkeypatch.setattr(runtime_module, "PROJECT_ROOT", tmp_path) monkeypatch.setattr(runtime_module, "_is_gateway_running", lambda: True) runtime_module.get_runtime_state().runtime_manager = _DummyManager() runtime_module.get_runtime_state().gateway_port = 8765 with TestClient(create_app()) as client: response = client.put( "/api/runtime/config", json={ "schedule_mode": "intraday", "interval_minutes": 15, "trigger_time": "10:15", "max_comm_cycles": 4, }, ) assert response.status_code == 200 payload = response.json() assert payload["bootstrap"]["schedule_mode"] == "intraday" assert payload["resolved"]["interval_minutes"] == 15 assert "interval_minutes: 15" in (run_dir / "BOOTSTRAP.md").read_text(encoding="utf-8") def test_prune_old_timestamped_runs_keeps_named_runs(monkeypatch, tmp_path): runs_dir = tmp_path / "runs" runs_dir.mkdir() keep_dirs = ["20260324_110000", "20260324_120000"] prune_dir = "20260324_100000" named_dir = "smoke_fullstack" for name in [*keep_dirs, prune_dir, named_dir]: (runs_dir / name).mkdir(parents=True) monkeypatch.setattr(runtime_module, "PROJECT_ROOT", tmp_path) pruned = runtime_module._prune_old_timestamped_runs(keep=1, exclude_run_ids={"20260324_120000"}) assert prune_dir in pruned assert (runs_dir / named_dir).exists() assert (runs_dir / "20260324_120000").exists() assert (runs_dir / "20260324_110000").exists() def test_runtime_cleanup_endpoint_prunes_old_runs(monkeypatch, tmp_path): runs_dir = tmp_path / "runs" runs_dir.mkdir() for name in ["20260324_090000", "20260324_100000", "20260324_110000", "smoke_fullstack"]: (runs_dir / name).mkdir(parents=True) monkeypatch.setattr(runtime_module, "PROJECT_ROOT", tmp_path) monkeypatch.setattr(runtime_module, "_is_gateway_running", lambda: False) with TestClient(create_app()) as client: response = client.post("/api/runtime/cleanup?keep=1") assert response.status_code == 200 payload = response.json() assert payload["status"] == "ok" assert sorted(payload["pruned_run_ids"]) == ["20260324_090000", "20260324_100000"] assert (runs_dir / "20260324_110000").exists() assert (runs_dir / "smoke_fullstack").exists() def test_runtime_history_lists_recent_runs(monkeypatch, tmp_path): run_dir = tmp_path / "runs" / "20260324_120000" (run_dir / "state").mkdir(parents=True) (run_dir / "team_dashboard").mkdir(parents=True) (run_dir / "state" / "runtime_state.json").write_text( json.dumps( { "context": { "config_name": "20260324_120000", "run_dir": str(run_dir), "bootstrap_values": {"tickers": ["AAPL"]}, }, "events": [], } ), encoding="utf-8", ) (run_dir / "team_dashboard" / "summary.json").write_text( json.dumps({"totalTrades": 3, "totalAssetValue": 123456.0}), encoding="utf-8", ) monkeypatch.setattr(runtime_module, "PROJECT_ROOT", tmp_path) with TestClient(create_app()) as client: response = client.get("/api/runtime/history?limit=5") assert response.status_code == 200 payload = response.json() assert payload["runs"][0]["run_id"] == "20260324_120000" assert payload["runs"][0]["total_trades"] == 3 def test_restore_run_assets_copies_state(monkeypatch, tmp_path): source_run = tmp_path / "runs" / "20260324_100000" (source_run / "team_dashboard").mkdir(parents=True) (source_run / "state").mkdir(parents=True) (source_run / "agents").mkdir(parents=True) (source_run / "team_dashboard" / "_internal_state.json").write_text("{}", encoding="utf-8") (source_run / "state" / "server_state.json").write_text("{}", encoding="utf-8") target_run = tmp_path / "runs" / "20260324_130000" monkeypatch.setattr(runtime_module, "PROJECT_ROOT", tmp_path) runtime_module._restore_run_assets("20260324_100000", target_run) assert (target_run / "team_dashboard" / "_internal_state.json").exists() assert (target_run / "state" / "server_state.json").exists() def test_start_runtime_restore_reuses_historical_run_id(monkeypatch, tmp_path): run_dir = tmp_path / "runs" / "20260324_100000" (run_dir / "state").mkdir(parents=True) (run_dir / "state" / "runtime_state.json").write_text( json.dumps( { "context": { "config_name": "20260324_100000", "run_dir": str(run_dir), "bootstrap_values": { "tickers": ["AAPL"], "schedule_mode": "intraday", "interval_minutes": 30, "trigger_time": "now", "max_comm_cycles": 2, "initial_cash": 100000.0, "margin_requirement": 0.0, "enable_memory": False, "mode": "live", "poll_interval": 10, "enable_mock": False, }, } } ), encoding="utf-8", ) class _DummyManager: def __init__(self, config_name, run_dir, bootstrap): self.config_name = config_name self.run_dir = Path(run_dir) self.bootstrap = bootstrap self.context = None def prepare_run(self): self.context = type( "Ctx", (), { "config_name": self.config_name, "run_dir": self.run_dir, "bootstrap_values": self.bootstrap, }, )() return self.context class _DummyProcess: def poll(self): return None monkeypatch.setattr(runtime_module, "PROJECT_ROOT", tmp_path) monkeypatch.setattr(runtime_module, "_find_available_port", lambda start_port=8765, max_port=9000: 8765) monkeypatch.setattr(runtime_module, "_start_gateway_process", lambda **kwargs: _DummyProcess()) monkeypatch.setattr(runtime_module, "_stop_gateway", lambda: True) monkeypatch.setattr("backend.runtime.manager.TradingRuntimeManager", _DummyManager) runtime_state = runtime_module.get_runtime_state() runtime_state.gateway_process = None with TestClient(create_app()) as client: response = client.post( "/api/runtime/start", json={ "launch_mode": "restore", "restore_run_id": "20260324_100000", "tickers": [], }, ) assert response.status_code == 200 payload = response.json() assert payload["run_id"] == "20260324_100000" assert payload["run_dir"] == str(run_dir)