- ping the board every 2 s so an unplugged or silent board shows as gone within seconds - never nudge the pointer right after the board connects or while the phone is not mirroring - read and log the board's hello (last reset reason, uptime) - serve the web UI with Cache-Control: no-cache so updates show on refresh - LLM_TOKEN_PARAM (auto by default) for models that want max_completion_tokens Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import httpx
|
|
|
|
from hidagent.llm import LLM
|
|
|
|
|
|
def fake_server(llm: LLM, wants: str) -> list[dict]:
|
|
"""Point the client at an in-memory endpoint that only accepts the `wants` token parameter."""
|
|
seen = []
|
|
|
|
def handle(request: httpx.Request) -> httpx.Response:
|
|
body = json.loads(request.content)
|
|
seen.append(body)
|
|
if wants not in body:
|
|
return httpx.Response(400, json={"error": {"message": f"Unsupported parameter. Use '{wants}' instead."}})
|
|
return httpx.Response(200, json={"choices": [{"message": {"content": "DO: wait"}}]})
|
|
|
|
llm._http = httpx.AsyncClient(base_url="http://test/v1", transport=httpx.MockTransport(handle))
|
|
return seen
|
|
|
|
|
|
def test_classic_servers_get_max_tokens_once():
|
|
llm = LLM("http://test/v1", "qwen")
|
|
seen = fake_server(llm, "max_tokens")
|
|
assert asyncio.run(llm.chat("sys", "hi")) == "DO: wait"
|
|
assert len(seen) == 1 and seen[0]["max_tokens"] == 600
|
|
|
|
|
|
def test_auto_switches_to_max_completion_tokens_and_remembers():
|
|
llm = LLM("http://test/v1", "gpt")
|
|
seen = fake_server(llm, "max_completion_tokens")
|
|
asyncio.run(llm.chat("sys", "hi"))
|
|
asyncio.run(llm.chat("sys", "again"))
|
|
assert [("max_tokens" in b, "max_completion_tokens" in b) for b in seen] == [(True, False), (False, True),
|
|
(False, True)]
|
|
|
|
|
|
def test_explicit_token_param_is_used_directly():
|
|
llm = LLM("http://test/v1", "gpt", token_param="max_completion_tokens")
|
|
seen = fake_server(llm, "max_completion_tokens")
|
|
asyncio.run(llm.chat("sys", "hi"))
|
|
assert len(seen) == 1
|
|
|
|
|
|
def test_extra_body_and_image_are_sent():
|
|
llm = LLM("http://test/v1", "gpt", extra={"reasoning_effort": "none"})
|
|
seen = fake_server(llm, "max_tokens")
|
|
asyncio.run(llm.chat("sys", "hi", image=b"\xff\xd8jpeg"))
|
|
body = seen[0]
|
|
assert body["reasoning_effort"] == "none"
|
|
assert body["messages"][1]["content"][0]["image_url"]["url"].startswith("data:image/jpeg;base64,")
|