wuyang
32 lines
1.3 KiB
Python
32 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import cast
|
|
|
|
import httpx
|
|
|
|
from agent_platform.auth import UserIdentity, decode_openwebui_identity
|
|
from agent_platform.runtime.tools import ToolContext, ToolRegistry
|
|
from agent_platform.store import RuntimeStore
|
|
|
|
|
|
async def test_gateway_call_uses_a_fresh_identity_token(settings) -> None:
|
|
captured: dict[str, str] = {}
|
|
|
|
async def handle(request: httpx.Request) -> httpx.Response:
|
|
captured["authorization"] = request.headers["Authorization"]
|
|
captured["identity"] = request.headers["X-OpenWebUI-User-Jwt"]
|
|
return httpx.Response(200, json={"ok": True, "output": "ready"})
|
|
|
|
identity = UserIdentity("user-123", "user@example.test", "Test User", "user")
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(handle)) as client:
|
|
registry = ToolRegistry(settings, cast(RuntimeStore, None), client=client)
|
|
result = await registry.execute(
|
|
"workspace_status",
|
|
{},
|
|
ToolContext(identity=identity, chat_id="chat-123"),
|
|
)
|
|
|
|
assert result["ok"] is True
|
|
assert captured["authorization"] == f"Bearer {settings.internal_gateway_key}"
|
|
assert decode_openwebui_identity(captured["identity"], settings.openwebui_forward_jwt_secret) == identity
|