wuyang
155 lines
5.3 KiB
Python
155 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import httpx
|
|
from fastapi.testclient import TestClient
|
|
|
|
from agent_platform.runtime.app import create_app
|
|
from agent_platform.store import RuntimeStore
|
|
|
|
|
|
class FakeModelProvider:
|
|
def __init__(self) -> None:
|
|
self.completed = []
|
|
self.forwarded = []
|
|
self.forward_options = []
|
|
|
|
async def complete(self, **kwargs):
|
|
self.completed.append(kwargs)
|
|
return {
|
|
"choices": [{"message": {"role": "assistant", "content": "agent complete"}, "finish_reason": "stop"}],
|
|
"usage": {"total_tokens": 10},
|
|
}
|
|
|
|
async def forward(self, payload, **kwargs):
|
|
self.forwarded.append(payload)
|
|
self.forward_options.append(kwargs)
|
|
request = httpx.Request("POST", "https://model.example.test/v1/chat/completions")
|
|
return httpx.Response(
|
|
200,
|
|
request=request,
|
|
headers={"content-type": "application/json"},
|
|
json={
|
|
"model": payload["model"],
|
|
"choices": [{"message": {"role": "assistant", "content": "background reply"}}],
|
|
},
|
|
)
|
|
|
|
async def close(self):
|
|
return None
|
|
|
|
|
|
class FakeTools:
|
|
async def close(self):
|
|
return None
|
|
|
|
def specs(self, *, read_only=False, allow_delegate=True):
|
|
return []
|
|
|
|
async def execute(self, name, arguments, context):
|
|
raise AssertionError("No tool should be called")
|
|
|
|
|
|
def headers(settings, identity_jwt, chat_id="chat-1"):
|
|
return {
|
|
"Authorization": f"Bearer {settings.internal_provider_key}",
|
|
"X-OpenWebUI-User-Jwt": identity_jwt,
|
|
"X-OpenWebUI-Chat-Id": chat_id,
|
|
}
|
|
|
|
|
|
def _app(settings, provider=None):
|
|
return create_app(
|
|
settings,
|
|
store=RuntimeStore(settings.database_url),
|
|
provider=provider or FakeModelProvider(),
|
|
tools=FakeTools(),
|
|
)
|
|
|
|
|
|
def test_model_list_requires_internal_key_and_exposes_four_models(settings) -> None:
|
|
with TestClient(_app(settings)) as client:
|
|
assert client.get("/v1/models").status_code == 401
|
|
response = client.get(
|
|
"/v1/models",
|
|
headers={"Authorization": f"Bearer {settings.internal_provider_key}"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert {item["id"] for item in response.json()["data"]} == {
|
|
"luna",
|
|
"terra",
|
|
"sol",
|
|
"deepseek-v4-pro",
|
|
}
|
|
|
|
|
|
def test_every_user_model_runs_the_custom_agent_loop(settings, identity_jwt) -> None:
|
|
provider = FakeModelProvider()
|
|
with TestClient(_app(settings, provider)) as client:
|
|
luna = client.post(
|
|
"/v1/chat/completions",
|
|
headers=headers(settings, identity_jwt, "luna-run"),
|
|
json={"model": "luna", "messages": [{"role": "user", "content": "hello"}]},
|
|
)
|
|
assert luna.status_code == 200
|
|
assert luna.json()["model"] == "luna"
|
|
assert provider.completed[-1]["model"] == "ChatGPT-5.6:Luna"
|
|
assert provider.completed[-1]["thinking_enabled"] is True
|
|
assert provider.forwarded == []
|
|
|
|
deepseek = client.post(
|
|
"/v1/chat/completions",
|
|
headers=headers(settings, identity_jwt, "deepseek-run"),
|
|
json={"model": "deepseek-v4-pro", "messages": [{"role": "user", "content": "do it"}]},
|
|
)
|
|
assert deepseek.status_code == 200
|
|
assert provider.completed[-1]["model"] == "deepseek-v4-pro"
|
|
assert provider.completed[-1]["provider"] == "deepseek"
|
|
assert provider.completed[-1]["reasoning_effort"] == "max"
|
|
|
|
legacy = client.post(
|
|
"/v1/chat/completions",
|
|
headers=headers(settings, identity_jwt, "legacy-run"),
|
|
json={"model": "work-medium", "messages": [{"role": "user", "content": "old"}]},
|
|
)
|
|
assert legacy.status_code == 404
|
|
|
|
|
|
def test_openwebui_background_task_bypasses_agent_loop(settings, identity_jwt) -> None:
|
|
provider = FakeModelProvider()
|
|
with TestClient(_app(settings, provider)) as client:
|
|
response = client.post(
|
|
"/v1/chat/completions",
|
|
headers=headers(settings, identity_jwt, "background-task"),
|
|
json={
|
|
"model": "terra",
|
|
"messages": [{"role": "user", "content": "Generate tags"}],
|
|
"metadata": {"task": "tags_generation"},
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["model"] == "terra"
|
|
assert provider.completed == []
|
|
assert provider.forwarded[-1]["model"] == "ChatGPT-5.6:Terra"
|
|
|
|
|
|
def test_agent_stream_never_emits_delta_tool_calls(settings, identity_jwt) -> None:
|
|
with TestClient(_app(settings)) as client:
|
|
response = client.post(
|
|
"/v1/chat/completions",
|
|
headers=headers(settings, identity_jwt, "stream-chat"),
|
|
json={
|
|
"model": "terra",
|
|
"stream": True,
|
|
"messages": [{"role": "user", "content": "do it"}],
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "agent complete" in response.text
|
|
for line in response.text.splitlines():
|
|
if line.startswith("data: {"):
|
|
chunk = json.loads(line.removeprefix("data: "))
|
|
assert "tool_calls" not in chunk["choices"][0]["delta"]
|
|
assert '"finish_reason": "stop"' in response.text
|