68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from server.routers.mcp_router import mcp
|
|
from server.utils.auth_middleware import get_admin_user, get_db
|
|
from yuxi.storage.postgres.models_business import User
|
|
|
|
|
|
def _build_app() -> FastAPI:
|
|
app = FastAPI()
|
|
app.include_router(mcp, prefix="/api")
|
|
|
|
async def fake_db():
|
|
return None
|
|
|
|
async def fake_admin_user():
|
|
return User(
|
|
username="admin",
|
|
user_id="admin",
|
|
password_hash="x",
|
|
role="admin",
|
|
)
|
|
|
|
app.dependency_overrides[get_db] = fake_db
|
|
app.dependency_overrides[get_admin_user] = fake_admin_user
|
|
return app
|
|
|
|
|
|
def test_update_mcp_server_status(monkeypatch):
|
|
captured = {}
|
|
|
|
class DummyServer:
|
|
def __init__(self, enabled):
|
|
self.enabled = enabled
|
|
|
|
def to_dict(self):
|
|
return {"name": "sequentialthinking", "enabled": self.enabled}
|
|
|
|
async def fake_set_server_enabled(db, name, enabled, updated_by=None):
|
|
captured["name"] = name
|
|
captured["enabled"] = enabled
|
|
captured["updated_by"] = updated_by
|
|
return enabled, DummyServer(enabled)
|
|
|
|
monkeypatch.setattr("server.routers.mcp_router.set_server_enabled", fake_set_server_enabled)
|
|
|
|
client = TestClient(_build_app())
|
|
resp = client.put("/api/system/mcp-servers/sequentialthinking/status", json={"enabled": False})
|
|
assert resp.status_code == 200, resp.text
|
|
payload = resp.json()
|
|
assert payload["success"] is True
|
|
assert payload["enabled"] is False
|
|
assert payload["data"]["enabled"] is False
|
|
assert captured == {"name": "sequentialthinking", "enabled": False, "updated_by": "admin"}
|
|
|
|
|
|
def test_update_mcp_server_status_not_found(monkeypatch):
|
|
async def fake_set_server_enabled(db, name, enabled, updated_by=None):
|
|
raise ValueError(f"Server '{name}' does not exist")
|
|
|
|
monkeypatch.setattr("server.routers.mcp_router.set_server_enabled", fake_set_server_enabled)
|
|
|
|
client = TestClient(_build_app())
|
|
resp = client.put("/api/system/mcp-servers/missing/status", json={"enabled": True})
|
|
assert resp.status_code == 404, resp.text
|