from __future__ import annotations from types import SimpleNamespace from fastapi.testclient import TestClient import traderai.server as server def test_config_update_rebuilds_runtime_without_restart(monkeypatch, tmp_path): state = {"settings": make_settings(tmp_path, model_provider="ollama", ollama_model="qwen3.5:9b")} class FakeScheduler: def __init__(self, memory): self.memory = memory def bind_agent(self, agent): self.agent = agent def bind_plan_runner(self, plan_runner): self.plan_runner = plan_runner def bind_uex_notifications(self, uex, poll_seconds=60): self.uex = uex self.poll_seconds = poll_seconds def start(self): return None def shutdown(self): return None def list_jobs(self): return [] class FakeUEXClient: def __init__(self, *args, **kwargs): pass async def get_user(self, username=None, authenticated=False): return {} class FakeToolRegistry: def __init__(self, *args, **kwargs): self.pending_actions = {} self.plan_runner = None async def approve(self, action_id): return {"approved": action_id} async def decline(self, action_id): return {"declined": action_id} class FakePlanRunner: def __init__(self, store, tools, memory, agent=None): self.store = store self.tools = tools self.memory = memory self.agent = agent def bind_agent(self, agent): self.agent = agent class FakeClient: def __init__(self, *args, **kwargs): pass async def fake_health(self): return { "online": True, "provider": self.provider, "model": self.model, "model_available": True, "message": f"{self.provider} ready", } async def fake_chat(self, content, thread_id=None, images=None): return {"message": f"{self.provider}:{self.model}", "pending_actions": [], "thread_id": thread_id} def fake_get_settings(): return state["settings"] def fake_save_settings(values): state["settings"] = make_settings( tmp_path, model_provider=values.get("model_provider", state["settings"].model_provider), ollama_model=values.get("ollama_model", state["settings"].ollama_model), codex_model=values.get("codex_model", state["settings"].codex_model), deepseek_model=values.get("deepseek_model", state["settings"].deepseek_model), ) return {"values": values, "fields": {}, "secrets_configured": {}, "app_data_dir": str(tmp_path)} monkeypatch.setattr(server, "WakeScheduler", FakeScheduler) monkeypatch.setattr(server, "UEXClient", FakeUEXClient) monkeypatch.setattr(server, "ToolRegistry", FakeToolRegistry) monkeypatch.setattr(server, "ContinualPlanRunner", FakePlanRunner) monkeypatch.setattr(server, "SCMDBClient", FakeClient) monkeypatch.setattr(server, "CornerstoneClient", FakeClient) monkeypatch.setattr(server, "StarCitizenWikiClient", FakeClient) monkeypatch.setattr(server, "get_settings", fake_get_settings) monkeypatch.setattr(server, "save_settings", fake_save_settings) monkeypatch.setattr( server, "settings_payload", lambda settings=None: {"app_data_dir": str(tmp_path), "values": {}, "fields": {}, "secrets_configured": {}}, ) monkeypatch.setattr(server.OllamaAgent, "health", fake_health) monkeypatch.setattr(server.OllamaAgent, "chat", fake_chat) app = server.create_app() with TestClient(app) as client: before = client.get("/api/health").json() assert before["model_provider"] == "ollama" assert before["inference"]["provider"] == "ollama" updated = client.post( "/api/config", json={"values": {"model_provider": "deepseek", "deepseek_model": "deepseek-v4-flash"}}, ).json() assert updated["restart_required"] is False after = client.get("/api/health").json() assert after["model_provider"] == "deepseek" assert after["inference"]["provider"] == "deepseek" chat = client.post("/api/chat", json={"message": "hi", "thread_id": "thread-1", "images": []}).json() assert chat["message"] == "deepseek:deepseek-v4-flash" def test_plan_draft_endpoint_returns_agent_draft(monkeypatch, tmp_path): state = {"settings": make_settings(tmp_path)} class FakeScheduler: def __init__(self, memory): self.memory = memory def bind_agent(self, agent): self.agent = agent def bind_plan_runner(self, plan_runner): self.plan_runner = plan_runner def bind_uex_notifications(self, uex, poll_seconds=60): self.uex = uex self.poll_seconds = poll_seconds def start(self): return None def shutdown(self): return None def list_jobs(self): return [] class FakeUEXClient: def __init__(self, *args, **kwargs): pass async def get_user(self, username=None, authenticated=False): return {} class FakeToolRegistry: def __init__(self, *args, **kwargs): self.pending_actions = {} self.plan_runner = None async def approve(self, action_id): return {"approved": action_id} async def decline(self, action_id): return {"declined": action_id} class FakePlanRunner: def __init__(self, store, tools, memory, agent=None): self.store = store self.tools = tools self.memory = memory self.agent = agent def bind_agent(self, agent): self.agent = agent class FakeClient: def __init__(self, *args, **kwargs): pass def fake_get_settings(): return state["settings"] monkeypatch.setattr(server, "WakeScheduler", FakeScheduler) monkeypatch.setattr(server, "UEXClient", FakeUEXClient) monkeypatch.setattr(server, "ToolRegistry", FakeToolRegistry) monkeypatch.setattr(server, "ContinualPlanRunner", FakePlanRunner) monkeypatch.setattr(server, "SCMDBClient", FakeClient) monkeypatch.setattr(server, "CornerstoneClient", FakeClient) monkeypatch.setattr(server, "StarCitizenWikiClient", FakeClient) monkeypatch.setattr(server, "get_settings", fake_get_settings) monkeypatch.setattr( server, "settings_payload", lambda settings=None: {"app_data_dir": str(tmp_path), "values": {}, "fields": {}, "secrets_configured": {}}, ) async def fake_generate_plan_draft(self, title="", objective="", kind="buying", constraints=None, items=None): return { "title": title or "Draft title", "objective": objective or "Draft objective", "kind": kind, "cadence": "0 */3 * * *", "constraints": {"message_tone": "friendly and direct", "instructions": "Start with the best listings."}, "items": [{"item_name": "RCMBNT-RGL-1", "desired_quantity": 1, "max_unit_price": None}], } monkeypatch.setattr(server.OllamaAgent, "generate_plan_draft", fake_generate_plan_draft) app = server.create_app() with TestClient(app) as client: response = client.post( "/api/plans/draft", json={"title": "Polaris parts", "objective": "Find the required parts", "kind": "buying", "constraints": {}, "items": []}, ) assert response.status_code == 200 draft = response.json()["draft"] assert draft["cadence"] == "0 */3 * * *" assert draft["constraints"]["instructions"] == "Start with the best listings." assert draft["items"][0]["item_name"] == "RCMBNT-RGL-1" def make_settings(tmp_path, model_provider="ollama", ollama_model="qwen3.5:9b", codex_model="gpt-5.4", deepseek_model="deepseek-v4-flash"): return SimpleNamespace( traderai_memory_path=str(tmp_path / "memory.sqlite3"), model_provider=model_provider, ollama_base_url="http://localhost:11434", ollama_model=ollama_model, ollama_num_ctx=64512, openai_base_url="https://api.openai.com/v1", openai_api_key=None, openai_model="gpt-5.4-mini", deepseek_base_url="https://api.deepseek.com", deepseek_api_key=None, deepseek_model=deepseek_model, model_reasoning_effort="medium", codex_command="codex", codex_model=codex_model, uex_base_url="https://api.uexcorp.space/2.0", uex_secret_key=None, uex_bearer_token=None, uex_negotiation_close_endpoint="marketplace_negotiations_close", traderai_user_name=None, uex_notification_poll_seconds=60, require_write_approval=True, scmdb_base_url="https://scmdb.net", cornerstone_base_url="https://finder.cstone.space", scwiki_base_url="https://starcitizen.tools", scwiki_api_base_url="https://api.star-citizen.wiki", )