259 lines
9.0 KiB
Python
259 lines
9.0 KiB
Python
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",
|
|
)
|