# -*- coding: utf-8 -*- """Tests for the extracted runtime service app surface.""" import json 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/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()