86 lines
3.5 KiB
Python
86 lines
3.5 KiB
Python
"""Local-only transport and failure tests; no real model or credentials used."""
|
|
|
|
import io
|
|
import json
|
|
from urllib.error import HTTPError
|
|
from urllib.request import ProxyHandler
|
|
|
|
import pytest
|
|
|
|
from services.hermes.scripts import lan_generate as api
|
|
|
|
|
|
def body(**changes):
|
|
"""Build a synthetic native request, allowing invalid-field variants."""
|
|
return json.dumps({"model": api.MODEL, "stream": False, "prompt": "Synthetic only", **changes}).encode()
|
|
|
|
|
|
def test_transport_ignores_proxies_and_rejects_redirects(monkeypatch):
|
|
"""Poisoned proxy variables and redirects cannot choose another backend."""
|
|
monkeypatch.setenv("HTTP_PROXY", "http://unapproved.invalid:8080")
|
|
assert not any(isinstance(h, ProxyHandler) and h.proxies for h in api._http.handlers)
|
|
redirect = next(h for h in api._http.handlers if isinstance(h, api.NoRedirect))
|
|
assert redirect.redirect_request(None, None, 307, "redirect", {}, "https://unapproved.invalid") is None
|
|
captured = []
|
|
|
|
class Opener:
|
|
"""Capture the single actual HTTP boundary and emulate an outage."""
|
|
|
|
def open(self, request, timeout):
|
|
captured.append((request, timeout))
|
|
raise HTTPError(request.full_url, 307, "PRIVATE_ERROR_TEXT", {}, io.BytesIO(b"private"))
|
|
|
|
monkeypatch.setattr(api, "_http", Opener())
|
|
status, result = api.generate(body())
|
|
assert status == 502 and "PRIVATE_ERROR_TEXT" not in json.dumps(result)
|
|
assert len(captured) == 1
|
|
assert captured[0][0].full_url == api.UPSTREAM + "/api/version"
|
|
assert "Authorization" not in captured[0][0].headers
|
|
|
|
|
|
@pytest.mark.parametrize("changes", [
|
|
{"options": {"num_ctx": 32768}}, {"options": {"seed": True}},
|
|
{"options": {"num_predict": True}}, {"format": {"type": "object", "$ref": "https://unapproved.invalid"}},
|
|
{"format": {"type": "array"}}, {"format": "xml"}, {"prompt": ""},
|
|
{"system": "new system prompt"}, {"options": {"temperature": float("inf")}},
|
|
])
|
|
def test_invalid_input_fails_before_metadata_or_inference(monkeypatch, changes):
|
|
monkeypatch.setattr(api, "_request", lambda *a, **k: pytest.fail("backend must not be called"))
|
|
assert api.generate(body(**changes))[0] == 400
|
|
|
|
|
|
@pytest.mark.parametrize("mutation,expected", [
|
|
({"model": "different"}, 503), ({"done": False}, 503),
|
|
({"done_reason": "length"}, 422), ({"response": "not JSON"}, 422),
|
|
])
|
|
def test_incomplete_or_substituted_output_is_not_success(monkeypatch, mutation, expected):
|
|
monkeypatch.setattr(api, "verify_model", lambda: {"model_digest": api.DIGEST})
|
|
monkeypatch.setattr(api, "_request", lambda *a, **k: {
|
|
"model": api.MODEL, "done": True, "response": '{"status":"ok"}', **mutation,
|
|
})
|
|
assert api.generate(body(format="json"))[0] == expected
|
|
|
|
|
|
def test_changed_digest_prevents_prompt_submission(monkeypatch):
|
|
calls = []
|
|
|
|
def upstream(path, *args, **kwargs):
|
|
calls.append(path)
|
|
return ({"version": api.RUNTIME} if path == "/api/version" else
|
|
{"models": [{"name": api.MODEL, "digest": "changed"}]})
|
|
|
|
monkeypatch.setattr(api, "_request", upstream)
|
|
assert api.generate(body())[0] == 503
|
|
assert calls == ["/api/version", "/api/tags"]
|
|
|
|
|
|
def test_timeout_releases_capacity_and_redacts_error(monkeypatch):
|
|
def timeout(*args, **kwargs):
|
|
raise TimeoutError("PRIVATE_PROMPT")
|
|
|
|
monkeypatch.setattr(api, "_request", timeout)
|
|
status, result = api.generate(body())
|
|
assert status == 504 and "PRIVATE_PROMPT" not in json.dumps(result)
|
|
assert api._inference.acquire(blocking=False)
|
|
api._inference.release()
|