feat(mock-provider): deterministic OpenAI-compatible endpoint for local and CI use
Adds src/mock_provider/, a standalone ASGI app that lets Honcho run with no model provider, no API key, and no spend. It answers /v1/chat/completions and /v1/embeddings with obviously-synthetic content derived from the request, so the same request always produces the same response. It runs as its own service from the standard Honcho image with a different entrypoint, the way api and deriver already differ, so there is no second image to build or keep in digest-sync. The app imports nothing from src.config or src.db, so it boots even when the rest of the stack is misconfigured. The chat endpoint generates from the JSON Schema it is sent rather than answering with prose. That matters because a prose answer does not fail loudly: repair_response_model_json swallows the parse error and returns an empty PromptRepresentation, which reads as "the deriver found nothing" rather than "the mock is wrong". Generation resolves $ref/$defs indirection, caps recursion for reasoning-tree schemas, and covers json_object mode by recovering the schema Honcho injects into the prompt. Embeddings are hash-derived, so identical input yields an identical vector. Tests drive the production OpenAIBackend and _EmbeddingClient against the app over ASGI, including the strict json_schema transform that chat.completions.parse() applies. Verified end to end against a real stack: messages in, conclusions and 1536-dim embeddings written to pgvector, with no calls to any real provider. Mock embeddings carry no semantic similarity, so recall against this provider must use lexical search. CONTRIBUTING notes that, and the load_dotenv(override= True) behaviour that lets a stale repo .env win over exported environment variables. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Eugene Eisenstein
GitHub
parent
2ad56a4d71
commit
b1594089aa
|
|
@ -219,6 +219,40 @@ uv run python -m src.deriver # background worker
|
|||
Everything Python goes through `uv run`. Redis is optional for local development; without it
|
||||
caching is simply disabled.
|
||||
|
||||
### Running without a model provider
|
||||
|
||||
`src/mock_provider/` is a deterministic, OpenAI-compatible endpoint, so you can run the full
|
||||
stack with no provider account, no API key, and no spend. It answers `/v1/chat/completions`
|
||||
and `/v1/embeddings` with obviously-synthetic content derived from the request, and the same
|
||||
request always produces the same response. Run it from the standard image or the repo:
|
||||
|
||||
```bash
|
||||
uv run fastapi run --host 0.0.0.0 --port 8106 src/mock_provider/main.py
|
||||
```
|
||||
|
||||
Then point Honcho at it. All three variables are required:
|
||||
|
||||
```bash
|
||||
export LLM_OPENAI_API_KEY=sandbox # gates client construction, not just auth
|
||||
export LLM_OPENAI_BASE_URL=http://localhost:8106/v1
|
||||
export EMBEDDING_MODEL_CONFIG__OVERRIDES__BASE_URL=http://localhost:8106/v1
|
||||
```
|
||||
|
||||
Embeddings resolve through a separate client that reads the base URL only from the per-module
|
||||
override, so without the third variable your embedding calls go to `api.openai.com` for real.
|
||||
Do not set any per-module credential override (`..._OVERRIDES__API_KEY` / `API_KEY_ENV`) —
|
||||
that makes the module ignore the global base URL.
|
||||
|
||||
Two things to know:
|
||||
|
||||
- **A repo `.env` beats your exported environment.** `src/config.py` calls
|
||||
`load_dotenv(override=True)` at import, so a stale `.env` silently wins over the variables
|
||||
above. Set `PYTHON_DOTENV_DISABLED=1` (and `HONCHO_CONFIG_TOML_DISABLED=1` for a local
|
||||
`config.toml`) when you need the environment to be the only input.
|
||||
- **Mock embeddings are hash-derived and carry no semantic similarity.** Two paraphrases are as
|
||||
far apart as two unrelated strings. Recall against this provider must use lexical/full-text
|
||||
search; anything asserting on vector ranking needs a real embedding provider.
|
||||
|
||||
## Making the change
|
||||
|
||||
### Branches and commits
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""Deterministic OpenAI-compatible provider for local and CI use."""
|
||||
|
|
@ -0,0 +1,212 @@
|
|||
"""OpenAI-compatible ``/chat/completions``, answered without inference."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Request
|
||||
from fastapi.responses import StreamingResponse
|
||||
|
||||
from src.mock_provider.coerce import as_dict, as_list, as_str
|
||||
from src.mock_provider.schema_gen import generate
|
||||
|
||||
router = APIRouter(tags=["mock-provider"])
|
||||
|
||||
# Honcho's json_object mode injects the schema into the prompt text rather than
|
||||
# into response_format (see _apply_json_object_mode in the OpenAI backend), so
|
||||
# the only machine-readable copy of the schema is inside a message.
|
||||
_SCHEMA_HINT = re.compile(r"schema:\s*(\{)", re.IGNORECASE)
|
||||
|
||||
|
||||
def _completion_id(payload: dict[str, Any]) -> str:
|
||||
"""Stable id, so a replayed request is byte-identical."""
|
||||
digest = hashlib.sha256(
|
||||
json.dumps(payload, sort_keys=True, default=str).encode()
|
||||
).hexdigest()
|
||||
return f"chatcmpl-mock-{digest[:24]}"
|
||||
|
||||
|
||||
def _extract_balanced_json(text: str, start: int) -> dict[str, Any] | None:
|
||||
"""Read one balanced ``{...}`` beginning at ``start`` and parse it.
|
||||
|
||||
A plain regex cannot do this — a JSON Schema contains nested objects, and
|
||||
braces inside string literals must not count toward the depth.
|
||||
"""
|
||||
depth = 0
|
||||
in_string = False
|
||||
escaped = False
|
||||
for index in range(start, len(text)):
|
||||
char = text[index]
|
||||
if in_string:
|
||||
if escaped:
|
||||
escaped = False
|
||||
elif char == "\\":
|
||||
escaped = True
|
||||
elif char == '"':
|
||||
in_string = False
|
||||
continue
|
||||
if char == '"':
|
||||
in_string = True
|
||||
elif char == "{":
|
||||
depth += 1
|
||||
elif char == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
try:
|
||||
parsed = json.loads(text[start : index + 1])
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
return as_dict(parsed)
|
||||
return None
|
||||
|
||||
|
||||
def _schema_from_messages(messages: list[Any]) -> dict[str, Any] | None:
|
||||
"""Recover an injected schema from the prompt, for json_object mode."""
|
||||
for message in reversed(messages):
|
||||
message_dict = as_dict(message)
|
||||
if message_dict is None:
|
||||
continue
|
||||
content = as_str(message_dict.get("content"))
|
||||
if content is None:
|
||||
continue
|
||||
for match in _SCHEMA_HINT.finditer(content):
|
||||
candidate = _extract_balanced_json(content, match.start(1))
|
||||
if candidate and ("properties" in candidate or "$defs" in candidate):
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def _response_content(payload: dict[str, Any]) -> str:
|
||||
"""The assistant message body: schema-conforming JSON, or prose."""
|
||||
response_format = as_dict(payload.get("response_format"))
|
||||
messages = as_list(payload.get("messages")) or []
|
||||
|
||||
if response_format is not None:
|
||||
kind = as_str(response_format.get("type"))
|
||||
if kind == "json_schema":
|
||||
wrapper = as_dict(response_format.get("json_schema"))
|
||||
if wrapper is not None:
|
||||
schema = as_dict(wrapper.get("schema"))
|
||||
if schema is not None:
|
||||
return json.dumps(generate(schema))
|
||||
# A json_schema request whose schema we cannot read must not fall
|
||||
# through to prose — that is the silent-empty failure this mock
|
||||
# exists to avoid. An empty object at least parses.
|
||||
return "{}"
|
||||
if kind == "json_object":
|
||||
schema = _schema_from_messages(messages)
|
||||
return json.dumps(generate(schema)) if schema else "{}"
|
||||
|
||||
return (
|
||||
"[mock] This is a synthetic response from Honcho's mock provider. "
|
||||
"No model was called."
|
||||
)
|
||||
|
||||
|
||||
def _usage(payload: dict[str, Any], content: str) -> dict[str, int]:
|
||||
"""Rough token accounting, so cost telemetry has plausible numbers."""
|
||||
messages = as_list(payload.get("messages")) or []
|
||||
prompt_chars = 0
|
||||
for message in messages:
|
||||
message_dict = as_dict(message)
|
||||
text = as_str(message_dict.get("content")) if message_dict else None
|
||||
if text is not None:
|
||||
prompt_chars += len(text)
|
||||
prompt_tokens = max(1, prompt_chars // 4)
|
||||
completion_tokens = max(1, len(content) // 4)
|
||||
return {
|
||||
"prompt_tokens": prompt_tokens,
|
||||
"completion_tokens": completion_tokens,
|
||||
"total_tokens": prompt_tokens + completion_tokens,
|
||||
}
|
||||
|
||||
|
||||
def _created() -> int:
|
||||
# Fixed rather than time-based: a mock that changes its output between
|
||||
# identical calls defeats the point.
|
||||
return 1577836800 # 2020-01-01T00:00:00Z
|
||||
|
||||
|
||||
async def _stream(
|
||||
completion_id: str, model: str, content: str, usage: dict[str, int]
|
||||
) -> AsyncIterator[bytes]:
|
||||
def chunk(payload: dict[str, Any]) -> bytes:
|
||||
return f"data: {json.dumps(payload)}\n\n".encode()
|
||||
|
||||
base = {
|
||||
"id": completion_id,
|
||||
"object": "chat.completion.chunk",
|
||||
"created": _created(),
|
||||
"model": model,
|
||||
}
|
||||
yield chunk(
|
||||
{
|
||||
**base,
|
||||
"choices": [
|
||||
{
|
||||
"index": 0,
|
||||
"delta": {"role": "assistant", "content": ""},
|
||||
"finish_reason": None,
|
||||
}
|
||||
],
|
||||
}
|
||||
)
|
||||
yield chunk(
|
||||
{
|
||||
**base,
|
||||
"choices": [
|
||||
{"index": 0, "delta": {"content": content}, "finish_reason": None}
|
||||
],
|
||||
}
|
||||
)
|
||||
yield chunk(
|
||||
{
|
||||
**base,
|
||||
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
|
||||
}
|
||||
)
|
||||
# The backend requests stream_options.include_usage and ends the stream on
|
||||
# the usage chunk, so it must come last and must carry choices: [].
|
||||
yield chunk({**base, "choices": [], "usage": usage})
|
||||
yield b"data: [DONE]\n\n"
|
||||
|
||||
|
||||
@router.post("/chat/completions")
|
||||
async def chat_completions(request: Request) -> Any:
|
||||
payload = as_dict(await request.json()) or {}
|
||||
|
||||
model = as_str(payload.get("model")) or "mock-model"
|
||||
content = _response_content(payload)
|
||||
usage = _usage(payload, content)
|
||||
completion_id = _completion_id(payload)
|
||||
|
||||
if payload.get("stream"):
|
||||
return StreamingResponse(
|
||||
_stream(completion_id, model, content, usage),
|
||||
media_type="text/event-stream",
|
||||
)
|
||||
|
||||
return {
|
||||
"id": completion_id,
|
||||
"object": "chat.completion",
|
||||
"created": _created(),
|
||||
"model": model,
|
||||
"choices": [
|
||||
{
|
||||
"index": 0,
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": content,
|
||||
"refusal": None,
|
||||
"tool_calls": None,
|
||||
},
|
||||
"logprobs": None,
|
||||
"finish_reason": "stop",
|
||||
}
|
||||
],
|
||||
"usage": usage,
|
||||
}
|
||||
|
|
@ -0,0 +1,34 @@
|
|||
"""Typed narrowing for values decoded from JSON.
|
||||
|
||||
``isinstance(value, dict)`` on an ``Any`` narrows to ``dict[Unknown, Unknown]``,
|
||||
which spreads unknown types through everything downstream. These helpers narrow
|
||||
and pin the element types in one step.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, cast
|
||||
|
||||
|
||||
def as_dict(value: object) -> dict[str, Any] | None:
|
||||
"""The value as a JSON object, or None if it is not one."""
|
||||
return cast("dict[str, Any]", value) if isinstance(value, dict) else None
|
||||
|
||||
|
||||
def as_list(value: object) -> list[Any] | None:
|
||||
"""The value as a JSON array, or None if it is not one."""
|
||||
return cast("list[Any]", value) if isinstance(value, list) else None
|
||||
|
||||
|
||||
def as_str(value: object) -> str | None:
|
||||
"""The value as a JSON string, or None if it is not one."""
|
||||
return value if isinstance(value, str) else None
|
||||
|
||||
|
||||
def as_int(value: object) -> int | None:
|
||||
"""The value as a JSON integer, or None if it is not one.
|
||||
|
||||
``bool`` is excluded: it is an ``int`` subclass, and a JSON ``true`` reaching
|
||||
a size or dimension field is a malformed request, not the number one.
|
||||
"""
|
||||
return value if isinstance(value, int) and not isinstance(value, bool) else None
|
||||
|
|
@ -0,0 +1,103 @@
|
|||
"""OpenAI-compatible ``/embeddings``, answered from a content hash."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import struct
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Request
|
||||
|
||||
from src.mock_provider.coerce import as_dict, as_int, as_list, as_str
|
||||
|
||||
router = APIRouter(tags=["mock-provider"])
|
||||
|
||||
# Honcho's default. EmbeddingClient._validate_embedding_dimensions raises when a
|
||||
# vector comes back at the wrong width, and validate_embedding_schema refuses to
|
||||
# boot when the width disagrees with the pgvector column, so the request's own
|
||||
# `dimensions` is honoured whenever it is present.
|
||||
DEFAULT_DIMENSIONS = 1536
|
||||
|
||||
|
||||
def content_to_embedding(content: str, dimensions: int) -> list[float]:
|
||||
"""A deterministic vector for ``content``.
|
||||
|
||||
Identical input yields an identical vector, and different inputs differ —
|
||||
which is what deduplication logic needs. It carries no semantic similarity:
|
||||
two paraphrases are as far apart as two unrelated strings. Anything
|
||||
asserting on ranking quality must not use this provider.
|
||||
|
||||
Mirrors ``_content_to_embedding`` in tests/conftest.py.
|
||||
"""
|
||||
digest = hashlib.sha256(content.encode()).digest()
|
||||
return [(digest[i % len(digest)] / 255.0) * 2 - 1 for i in range(dimensions)]
|
||||
|
||||
|
||||
def _encode_base64(vector: list[float]) -> str:
|
||||
"""Little-endian float32, which is what the OpenAI SDK decodes."""
|
||||
return base64.b64encode(struct.pack(f"<{len(vector)}f", *vector)).decode()
|
||||
|
||||
|
||||
def _normalize_input(raw: object) -> list[str]:
|
||||
"""Accept every input shape the OpenAI embeddings API allows.
|
||||
|
||||
Token-array inputs (list[int] / list[list[int]]) are rendered back to a
|
||||
stable string rather than rejected — the vector only has to be deterministic.
|
||||
"""
|
||||
if raw is None:
|
||||
return []
|
||||
if (text := as_str(raw)) is not None:
|
||||
return [text]
|
||||
|
||||
items = as_list(raw)
|
||||
if items is None:
|
||||
return [str(raw)]
|
||||
|
||||
# A flat list of ints is one tokenized input, not many single-token ones.
|
||||
if items and all(as_int(item) is not None for item in items):
|
||||
return [",".join(str(item) for item in items)]
|
||||
|
||||
texts: list[str] = []
|
||||
for item in items:
|
||||
if (item_text := as_str(item)) is not None:
|
||||
texts.append(item_text)
|
||||
elif (parts := as_list(item)) is not None:
|
||||
texts.append(",".join(str(part) for part in parts))
|
||||
else:
|
||||
texts.append(str(item))
|
||||
return texts
|
||||
|
||||
|
||||
@router.post("/embeddings")
|
||||
async def embeddings(request: Request) -> Any:
|
||||
payload = as_dict(await request.json()) or {}
|
||||
|
||||
texts = _normalize_input(payload.get("input"))
|
||||
requested = as_int(payload.get("dimensions"))
|
||||
dimensions = requested if requested and requested > 0 else DEFAULT_DIMENSIONS
|
||||
|
||||
# The SDK omits encoding_format only when it wants base64 (it sets the
|
||||
# parameter explicitly for float), so absent means base64.
|
||||
encoding_format = as_str(payload.get("encoding_format")) or "base64"
|
||||
|
||||
data: list[dict[str, Any]] = []
|
||||
for index, text in enumerate(texts):
|
||||
vector = content_to_embedding(text, dimensions)
|
||||
data.append(
|
||||
{
|
||||
"object": "embedding",
|
||||
"index": index,
|
||||
"embedding": (
|
||||
vector if encoding_format == "float" else _encode_base64(vector)
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
prompt_tokens = max(1, sum(len(text) for text in texts) // 4)
|
||||
return {
|
||||
"object": "list",
|
||||
"data": data,
|
||||
"model": as_str(payload.get("model")) or "mock-embedding",
|
||||
"usage": {"prompt_tokens": prompt_tokens, "total_tokens": prompt_tokens},
|
||||
}
|
||||
|
|
@ -0,0 +1,59 @@
|
|||
"""A deterministic, OpenAI-compatible provider for local and CI use.
|
||||
|
||||
Lets Honcho run with no model provider, no API key, and no spend. It answers
|
||||
``/v1/chat/completions`` and ``/v1/embeddings`` with obviously-synthetic content
|
||||
derived from the request, so the same request always produces the same response.
|
||||
|
||||
Runs as its own service from the standard Honcho image:
|
||||
|
||||
fastapi run --host 0.0.0.0 src/mock_provider/main.py
|
||||
|
||||
Point Honcho at it with three variables — all three are required:
|
||||
|
||||
LLM_OPENAI_API_KEY=sandbox # gates client construction, not just auth
|
||||
LLM_OPENAI_BASE_URL=http://mock-provider:8000/v1
|
||||
EMBEDDING_MODEL_CONFIG__OVERRIDES__BASE_URL=http://mock-provider:8000/v1
|
||||
|
||||
Embeddings resolve through a separate client that reads the base URL only from
|
||||
the per-module override, so without the third variable embedding calls go to
|
||||
api.openai.com for real. Do not set any per-module credential override
|
||||
(``..._OVERRIDES__API_KEY`` / ``API_KEY_ENV``) — that makes the module ignore the
|
||||
global base URL.
|
||||
|
||||
Embeddings are hash-derived and carry no semantic similarity. Recall assertions
|
||||
against this provider must use lexical/full-text search, not vector ranking.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from fastapi import FastAPI
|
||||
|
||||
from src.mock_provider import chat, embeddings
|
||||
|
||||
app = FastAPI(
|
||||
title="Honcho Mock Provider",
|
||||
description="Deterministic OpenAI-compatible endpoint for local and CI use.",
|
||||
version="1.0.0",
|
||||
)
|
||||
|
||||
# Mounted at both prefixes so the base URL works with or without /v1.
|
||||
for _router in (chat.router, embeddings.router):
|
||||
app.include_router(_router, prefix="/v1")
|
||||
app.include_router(_router)
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
async def health() -> dict[str, str]:
|
||||
return {"status": "ok", "provider": "mock"}
|
||||
|
||||
|
||||
@app.get("/{path:path}")
|
||||
async def catch_all(path: str) -> dict[str, Any]:
|
||||
"""Answer any other GET, so a bare ``/`` works as a container healthcheck.
|
||||
|
||||
Deliberately GET-only: an unimplemented POST returns 405 rather than a
|
||||
plausible-looking 200, so a missing endpoint fails loudly.
|
||||
"""
|
||||
return {"object": "mock", "path": path, "detail": "mock provider placeholder"}
|
||||
|
|
@ -0,0 +1,285 @@
|
|||
"""Generate a conforming instance from a JSON Schema.
|
||||
|
||||
The deriver is a structured-output caller: it sends a schema and parses the
|
||||
reply back into a Pydantic model. A mock that answers with prose does not fail
|
||||
loudly — ``repair_response_model_json`` swallows the error and hands back an
|
||||
empty ``PromptRepresentation``, which reads as "the deriver found nothing"
|
||||
rather than "the mock is wrong". So generation is driven by the schema that was
|
||||
actually sent, ``$ref`` indirection and all.
|
||||
|
||||
Values are derived from a hash of the property path, so the same schema always
|
||||
produces the same instance and two different fields never collide.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from typing import Any
|
||||
|
||||
from src.mock_provider.coerce import as_dict, as_int, as_list, as_str
|
||||
|
||||
# Depth cap for self-referential schemas. Reasoning-tree models nest premises
|
||||
# inside conclusions, so a $ref cycle is normal input, not a malformed schema.
|
||||
MAX_DEPTH = 6
|
||||
|
||||
_WORDS = (
|
||||
"synthetic",
|
||||
"placeholder",
|
||||
"mock",
|
||||
"sample",
|
||||
"fixture",
|
||||
"stub",
|
||||
"generated",
|
||||
"example",
|
||||
"inert",
|
||||
"dummy",
|
||||
)
|
||||
|
||||
|
||||
def _seed(path: str) -> int:
|
||||
return int.from_bytes(hashlib.sha256(path.encode()).digest()[:8], "big")
|
||||
|
||||
|
||||
def _phrase(path: str, words: int = 6) -> str:
|
||||
"""An obviously-synthetic sentence, stable for a given path."""
|
||||
seed = _seed(path)
|
||||
picked = [_WORDS[(seed >> (i * 5)) % len(_WORDS)] for i in range(words)]
|
||||
return f"[mock] {' '.join(picked)}"
|
||||
|
||||
|
||||
def _resolve(schema: dict[str, Any], root: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Follow a local ``$ref`` chain to the schema it points at.
|
||||
|
||||
Only local refs are supported: the mock never fetches over the network, and
|
||||
Pydantic's ``model_json_schema()`` only ever emits ``#/$defs/...``.
|
||||
"""
|
||||
seen: set[str] = set()
|
||||
current = schema
|
||||
while "$ref" in current:
|
||||
ref = as_str(current["$ref"])
|
||||
if ref is None or not ref.startswith("#/") or ref in seen:
|
||||
return {}
|
||||
seen.add(ref)
|
||||
|
||||
target: dict[str, Any] | None = root
|
||||
for part in ref[2:].split("/"):
|
||||
if target is None or part not in target:
|
||||
return {}
|
||||
target = as_dict(target[part])
|
||||
if target is None:
|
||||
return {}
|
||||
current = target
|
||||
return current
|
||||
|
||||
|
||||
def _merge_all_of(schema: dict[str, Any], root: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Flatten ``allOf`` into the parent so one pass can read properties off it."""
|
||||
branches = as_list(schema.get("allOf"))
|
||||
if branches is None:
|
||||
return schema
|
||||
|
||||
merged: dict[str, Any] = {k: v for k, v in schema.items() if k != "allOf"}
|
||||
for branch in branches:
|
||||
resolved_branch = as_dict(branch)
|
||||
if resolved_branch is None:
|
||||
continue
|
||||
resolved = _resolve(resolved_branch, root)
|
||||
for key, value in resolved.items():
|
||||
if key == "properties":
|
||||
properties = as_dict(value)
|
||||
if properties is not None:
|
||||
existing = as_dict(merged.get("properties")) or {}
|
||||
merged["properties"] = {**properties, **existing}
|
||||
continue
|
||||
if key == "required":
|
||||
required = as_list(value)
|
||||
if required is not None:
|
||||
previous = as_list(merged.get("required")) or []
|
||||
merged["required"] = list({*previous, *required})
|
||||
continue
|
||||
merged.setdefault(key, value)
|
||||
return merged
|
||||
|
||||
|
||||
def _infer_type(schema: dict[str, Any]) -> str:
|
||||
"""Best-effort type when the schema omits an explicit ``type``."""
|
||||
declared = schema.get("type")
|
||||
if (name := as_str(declared)) is not None:
|
||||
return name
|
||||
if (names := as_list(declared)) is not None:
|
||||
# Nullable unions arrive as ["string", "null"]; prefer the real type.
|
||||
for candidate in names:
|
||||
if (candidate_name := as_str(candidate)) and candidate_name != "null":
|
||||
return candidate_name
|
||||
return "null"
|
||||
if "properties" in schema:
|
||||
return "object"
|
||||
if "items" in schema:
|
||||
return "array"
|
||||
return "string"
|
||||
|
||||
|
||||
def generate(schema: dict[str, Any], root: dict[str, Any] | None = None) -> Any:
|
||||
"""Build a value satisfying ``schema``.
|
||||
|
||||
``root`` carries the document that ``$ref`` resolves against; it defaults to
|
||||
``schema`` itself, which is the shape Pydantic emits.
|
||||
"""
|
||||
return _generate(schema, root if root is not None else schema, "$", 0)
|
||||
|
||||
|
||||
def _generate(
|
||||
schema: dict[str, Any], root: dict[str, Any], path: str, depth: int
|
||||
) -> Any:
|
||||
resolved = _merge_all_of(_resolve(schema, root), root)
|
||||
|
||||
if "const" in resolved:
|
||||
return resolved["const"]
|
||||
|
||||
enum = as_list(resolved.get("enum"))
|
||||
if enum:
|
||||
return enum[_seed(path) % len(enum)]
|
||||
|
||||
if depth >= MAX_DEPTH and "default" in resolved:
|
||||
return resolved["default"]
|
||||
|
||||
for key in ("anyOf", "oneOf"):
|
||||
branches = as_list(resolved.get(key))
|
||||
if branches:
|
||||
return _generate(_pick_branch(branches, root, depth), root, path, depth)
|
||||
|
||||
kind = _infer_type(resolved)
|
||||
if kind == "object":
|
||||
return _generate_object(resolved, root, path, depth)
|
||||
if kind == "array":
|
||||
return _generate_array(resolved, root, path, depth)
|
||||
if kind == "integer":
|
||||
return _bounded_int(resolved, path)
|
||||
if kind == "number":
|
||||
return float(_bounded_int(resolved, path))
|
||||
if kind == "boolean":
|
||||
return _seed(path) % 2 == 0
|
||||
if kind == "null":
|
||||
return None
|
||||
return _generate_string(resolved, path)
|
||||
|
||||
|
||||
def _pick_branch(
|
||||
branches: list[Any], root: dict[str, Any], depth: int
|
||||
) -> dict[str, Any]:
|
||||
"""Choose a union member, preferring a non-null one.
|
||||
|
||||
Past the depth cap the order flips: a nullable recursive field terminates on
|
||||
``null`` instead of nesting another level.
|
||||
"""
|
||||
resolved: list[dict[str, Any]] = []
|
||||
for branch in branches:
|
||||
branch_dict = as_dict(branch)
|
||||
if branch_dict is not None:
|
||||
resolved.append(_resolve(branch_dict, root))
|
||||
if not resolved:
|
||||
return {}
|
||||
|
||||
if depth >= MAX_DEPTH:
|
||||
nulls = [b for b in resolved if _infer_type(b) == "null"]
|
||||
if nulls:
|
||||
return nulls[0]
|
||||
non_null = [b for b in resolved if _infer_type(b) != "null"]
|
||||
return non_null[0] if non_null else resolved[0]
|
||||
|
||||
|
||||
def _generate_object(
|
||||
schema: dict[str, Any], root: dict[str, Any], path: str, depth: int
|
||||
) -> dict[str, Any]:
|
||||
properties = as_dict(schema.get("properties"))
|
||||
if properties is None:
|
||||
return {}
|
||||
|
||||
# OpenAI structured outputs run in strict mode, where every property is
|
||||
# required. Emitting the full property set satisfies both strict and loose
|
||||
# schemas, so `required` is only consulted to decide what to drop once the
|
||||
# depth cap has been hit.
|
||||
declared_required = as_list(schema.get("required"))
|
||||
required: set[str] = (
|
||||
{name for name in (as_str(item) for item in declared_required) if name}
|
||||
if declared_required is not None
|
||||
else set(properties)
|
||||
)
|
||||
|
||||
result: dict[str, Any] = {}
|
||||
for name, subschema in properties.items():
|
||||
if depth >= MAX_DEPTH and name not in required:
|
||||
continue
|
||||
child = as_dict(subschema)
|
||||
if child is None:
|
||||
continue
|
||||
result[name] = _generate(child, root, f"{path}.{name}", depth + 1)
|
||||
return result
|
||||
|
||||
|
||||
def _generate_array(
|
||||
schema: dict[str, Any], root: dict[str, Any], path: str, depth: int
|
||||
) -> list[Any]:
|
||||
items = as_dict(schema.get("items"))
|
||||
min_items = as_int(schema.get("minItems"))
|
||||
max_items = as_int(schema.get("maxItems"))
|
||||
|
||||
count = 2
|
||||
if min_items is not None:
|
||||
count = max(count, min_items)
|
||||
if max_items is not None:
|
||||
count = min(count, max_items)
|
||||
if depth >= MAX_DEPTH:
|
||||
count = min_items or 0
|
||||
if items is None or count <= 0:
|
||||
return []
|
||||
|
||||
return [_generate(items, root, f"{path}[{i}]", depth + 1) for i in range(count)]
|
||||
|
||||
|
||||
def _generate_string(schema: dict[str, Any], path: str) -> str:
|
||||
fmt = as_str(schema.get("format"))
|
||||
if fmt == "date-time":
|
||||
return "2020-01-01T00:00:00Z"
|
||||
if fmt == "date":
|
||||
return "2020-01-01"
|
||||
if fmt == "uuid":
|
||||
stem = hashlib.sha256(path.encode()).hexdigest()[:8]
|
||||
return f"{stem}-0000-4000-8000-000000000000"
|
||||
if fmt in ("uri", "url"):
|
||||
return "https://mock.invalid/placeholder"
|
||||
if fmt == "email":
|
||||
return "placeholder@mock.invalid"
|
||||
|
||||
value = _phrase(path)
|
||||
min_length = as_int(schema.get("minLength"))
|
||||
max_length = as_int(schema.get("maxLength"))
|
||||
if min_length is not None and len(value) < min_length:
|
||||
value = value.ljust(min_length, "x")
|
||||
if max_length is not None and len(value) > max_length:
|
||||
value = value[:max_length]
|
||||
return value
|
||||
|
||||
|
||||
def _bounded_int(schema: dict[str, Any], path: str) -> int:
|
||||
low = as_int(schema.get("minimum"))
|
||||
if (
|
||||
low is None
|
||||
and (exclusive := as_int(schema.get("exclusiveMinimum"))) is not None
|
||||
):
|
||||
low = exclusive + 1
|
||||
high = as_int(schema.get("maximum"))
|
||||
if (
|
||||
high is None
|
||||
and (exclusive := as_int(schema.get("exclusiveMaximum"))) is not None
|
||||
):
|
||||
high = exclusive - 1
|
||||
|
||||
if low is not None and high is not None:
|
||||
span = high - low
|
||||
return low + (_seed(path) % (span + 1) if span > 0 else 0)
|
||||
if low is not None:
|
||||
return low + (_seed(path) % 8)
|
||||
if high is not None:
|
||||
return high - (_seed(path) % 8)
|
||||
return _seed(path) % 100
|
||||
|
|
@ -91,6 +91,9 @@ _RUNTIME_MOCK_TEST_BLOCKLIST_PREFIXES = (
|
|||
# Pure JWT scope tests — operate on src.security directly, no DB needed.
|
||||
"tests/test_security.py",
|
||||
"tests/test_generate_jwt_script.py",
|
||||
# The mock provider is a standalone ASGI app with no database or LLM of its
|
||||
# own; the runtime mocks would patch the very seams it exists to replace.
|
||||
"tests/mock_provider/",
|
||||
)
|
||||
|
||||
_LIVE_LLM_MARKER = "live_llm"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,221 @@
|
|||
"""Drive Honcho's real provider clients against the mock over ASGI.
|
||||
|
||||
The unit tests assert the mock's own output. These assert the hop that actually
|
||||
matters: ``OpenAIBackend`` and ``EmbeddingClient`` — the production classes,
|
||||
unpatched — talking to the mock through the genuine OpenAI SDK, including the
|
||||
``strict: true`` json_schema transform that ``chat.completions.parse()`` applies
|
||||
on the way out and the Pydantic validation it applies on the way back.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from openai import AsyncOpenAI
|
||||
|
||||
from src.config import EmbeddingModelConfig
|
||||
from src.embedding_client import _EmbeddingClient # pyright: ignore[reportPrivateUsage]
|
||||
from src.llm.backends.openai import OpenAIBackend
|
||||
from src.mock_provider.embeddings import content_to_embedding
|
||||
from src.mock_provider.main import app
|
||||
from src.utils.representation import PromptRepresentation
|
||||
|
||||
MESSAGES: list[dict[str, Any]] = [
|
||||
{"role": "user", "content": "I switched the service from pip to uv last week."}
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def openai_client() -> AsyncOpenAI:
|
||||
return AsyncOpenAI(
|
||||
api_key="sandbox",
|
||||
base_url="http://mock-provider.invalid/v1",
|
||||
http_client=httpx.AsyncClient(transport=httpx.ASGITransport(app=app)),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backend_parses_the_deriver_response_model(
|
||||
openai_client: AsyncOpenAI,
|
||||
) -> None:
|
||||
"""The production path: parse() with a Pydantic response_format."""
|
||||
backend = OpenAIBackend(openai_client)
|
||||
|
||||
result = await backend.complete(
|
||||
model="mock-model",
|
||||
messages=MESSAGES,
|
||||
max_tokens=512,
|
||||
response_format=PromptRepresentation,
|
||||
)
|
||||
|
||||
assert isinstance(result.content, PromptRepresentation)
|
||||
# An empty explicit list is what a prose-answering mock silently produces,
|
||||
# so it is the specific thing worth asserting against.
|
||||
assert result.content.explicit
|
||||
assert result.output_tokens > 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backend_json_object_mode_recovers_the_schema(
|
||||
openai_client: AsyncOpenAI,
|
||||
) -> None:
|
||||
"""json_object mode carries the schema in the prompt, not response_format."""
|
||||
backend = OpenAIBackend(openai_client)
|
||||
|
||||
result = await backend.complete(
|
||||
model="mock-model",
|
||||
messages=MESSAGES,
|
||||
max_tokens=512,
|
||||
response_format=PromptRepresentation,
|
||||
extra_params={"structured_output_mode": "json_object"},
|
||||
)
|
||||
|
||||
assert isinstance(result.content, PromptRepresentation)
|
||||
assert result.content.explicit
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backend_with_tools_uses_json_schema_and_still_parses(
|
||||
openai_client: AsyncOpenAI,
|
||||
) -> None:
|
||||
"""Non-strict tools force create() + explicit json_schema instead of parse()."""
|
||||
backend = OpenAIBackend(openai_client)
|
||||
|
||||
result = await backend.complete(
|
||||
model="mock-model",
|
||||
messages=MESSAGES,
|
||||
max_tokens=512,
|
||||
response_format=PromptRepresentation,
|
||||
tools=[
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "search_memory",
|
||||
"description": "Search memory",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {"query": {"type": "string"}},
|
||||
},
|
||||
},
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
assert isinstance(result.content, PromptRepresentation)
|
||||
assert result.content.explicit
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backend_plain_completion(openai_client: AsyncOpenAI) -> None:
|
||||
backend = OpenAIBackend(openai_client)
|
||||
|
||||
result = await backend.complete(
|
||||
model="mock-model", messages=MESSAGES, max_tokens=128
|
||||
)
|
||||
|
||||
assert isinstance(result.content, str)
|
||||
assert "[mock]" in result.content
|
||||
assert result.finish_reason == "stop"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backend_stream_yields_content_then_a_usage_terminator(
|
||||
openai_client: AsyncOpenAI,
|
||||
) -> None:
|
||||
backend = OpenAIBackend(openai_client)
|
||||
|
||||
chunks = [
|
||||
chunk
|
||||
async for chunk in backend.stream(
|
||||
model="mock-model", messages=MESSAGES, max_tokens=128
|
||||
)
|
||||
]
|
||||
|
||||
assert "[mock]" in "".join(chunk.content or "" for chunk in chunks)
|
||||
|
||||
terminator = chunks[-1]
|
||||
assert terminator.is_done
|
||||
assert terminator.finish_reason == "stop"
|
||||
# None here means the stream ended without a usage chunk, which is the
|
||||
# failure mode when stream_options.include_usage goes unanswered.
|
||||
assert terminator.output_tokens is not None
|
||||
assert terminator.output_tokens > 0
|
||||
|
||||
|
||||
def _embedding_client(dimensions: int, encoding_format: str) -> _EmbeddingClient:
|
||||
# The public EmbeddingClient is a settings-driven singleton wrapper; the
|
||||
# transport behaviour under test lives on the implementation it wraps.
|
||||
return _EmbeddingClient(
|
||||
EmbeddingModelConfig(
|
||||
model="text-embedding-3-small",
|
||||
transport="openai",
|
||||
api_key="sandbox",
|
||||
base_url="http://mock-provider.invalid/v1",
|
||||
),
|
||||
vector_dimensions=dimensions,
|
||||
max_input_tokens=8192,
|
||||
max_tokens_per_request=300000,
|
||||
send_dimensions=True,
|
||||
encoding_format=encoding_format, # pyright: ignore[reportArgumentType]
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _route_embedding_client_over_asgi( # pyright: ignore[reportUnusedFunction]
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Give the embedding client's AsyncOpenAI an ASGI transport.
|
||||
|
||||
EmbeddingClient builds its own client internally, so the transport has to be
|
||||
injected at construction rather than passed in.
|
||||
"""
|
||||
original = AsyncOpenAI.__init__
|
||||
|
||||
def patched(self: AsyncOpenAI, *args: Any, **kwargs: Any) -> None:
|
||||
kwargs.setdefault(
|
||||
"http_client",
|
||||
httpx.AsyncClient(transport=httpx.ASGITransport(app=app)),
|
||||
)
|
||||
original(self, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(AsyncOpenAI, "__init__", patched)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("encoding_format", ["float", "base64"])
|
||||
async def test_embedding_client_round_trip(encoding_format: str) -> None:
|
||||
"""Covers both wire encodings; base64 is what the SDK uses by default."""
|
||||
client = _embedding_client(1536, encoding_format)
|
||||
|
||||
vector = await client.embed("I switched the service from pip to uv.")
|
||||
|
||||
# _validate_embedding_dimensions raises on a width mismatch, so reaching
|
||||
# here already proves the width is right; assert the values too.
|
||||
assert len(vector) == 1536
|
||||
assert vector == pytest.approx( # pyright: ignore[reportUnknownMemberType]
|
||||
content_to_embedding("I switched the service from pip to uv.", 1536),
|
||||
abs=1e-6,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_embedding_client_honours_a_non_default_dimension() -> None:
|
||||
"""send_dimensions=True forwards `dimensions`; the mock must obey it."""
|
||||
client = _embedding_client(256, "float")
|
||||
|
||||
assert len(await client.embed("hello")) == 256
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_embedding_client_batches() -> None:
|
||||
"""_validate_embedding_count rejects a mismatched count."""
|
||||
client = _embedding_client(1536, "float")
|
||||
texts = [f"observation number {index}" for index in range(12)]
|
||||
|
||||
vectors = await client.simple_batch_embed(texts)
|
||||
|
||||
assert len(vectors) == len(texts)
|
||||
assert all(len(vector) == 1536 for vector in vectors)
|
||||
assert len({tuple(vector) for vector in vectors}) == len(texts)
|
||||
|
|
@ -0,0 +1,439 @@
|
|||
"""Contract tests for the mock provider.
|
||||
|
||||
The failure this guards against is silent: when the mock answers a structured
|
||||
request with something the deriver cannot parse, ``repair_response_model_json``
|
||||
falls back to an empty ``PromptRepresentation`` and the run looks like "the
|
||||
deriver found nothing" rather than "the mock is broken". So the assertions here
|
||||
are about parseability against real Honcho models, not about response shape.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
import struct
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from src.mock_provider.embeddings import content_to_embedding
|
||||
from src.mock_provider.main import app
|
||||
from src.mock_provider.schema_gen import generate
|
||||
from src.utils.representation import PromptRepresentation
|
||||
|
||||
# A $ref/$defs schema, which is what Pydantic emits for any nested model and the
|
||||
# indirection a naive generator silently drops.
|
||||
PROBE_SCHEMA: dict[str, Any] = {
|
||||
"$defs": {
|
||||
"Item": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"count": {"type": "integer", "minimum": 1, "maximum": 5},
|
||||
},
|
||||
"required": ["name", "count"],
|
||||
}
|
||||
},
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"label": {"type": "string"},
|
||||
"items": {"type": "array", "items": {"$ref": "#/$defs/Item"}},
|
||||
},
|
||||
"required": ["label", "items"],
|
||||
}
|
||||
|
||||
|
||||
class ProbeItem(BaseModel):
|
||||
name: str
|
||||
count: int = Field(ge=1, le=5)
|
||||
|
||||
|
||||
class Probe(BaseModel):
|
||||
label: str
|
||||
items: list[ProbeItem]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client() -> TestClient:
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def _post_chat(client: TestClient, **payload: Any) -> dict[str, Any]:
|
||||
payload.setdefault("model", "mock-model")
|
||||
payload.setdefault("messages", [{"role": "user", "content": "hello"}])
|
||||
response = client.post("/v1/chat/completions", json=payload)
|
||||
assert response.status_code == 200, response.text
|
||||
return response.json()
|
||||
|
||||
|
||||
def _json_schema_format(schema: dict[str, Any], name: str) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "json_schema",
|
||||
"json_schema": {"name": name, "schema": schema, "strict": True},
|
||||
}
|
||||
|
||||
|
||||
# --- structured output ------------------------------------------------------
|
||||
|
||||
|
||||
def test_json_schema_request_round_trips_into_its_pydantic_model(
|
||||
client: TestClient,
|
||||
) -> None:
|
||||
body = _post_chat(
|
||||
client, response_format=_json_schema_format(PROBE_SCHEMA, "Probe")
|
||||
)
|
||||
content = body["choices"][0]["message"]["content"]
|
||||
|
||||
probe = Probe.model_validate_json(content)
|
||||
assert probe.label
|
||||
assert probe.items, "$ref array must not come back empty"
|
||||
assert all(1 <= item.count <= 5 for item in probe.items)
|
||||
|
||||
|
||||
def test_deriver_response_model_round_trips() -> None:
|
||||
"""The real model the deriver parses, not a stand-in."""
|
||||
schema = PromptRepresentation.model_json_schema()
|
||||
content = json.dumps(generate(schema))
|
||||
|
||||
representation = PromptRepresentation.model_validate_json(content)
|
||||
assert (
|
||||
representation.explicit
|
||||
), "an empty explicit list is exactly the silent failure this mock avoids"
|
||||
|
||||
|
||||
def test_json_schema_response_is_never_prose(client: TestClient) -> None:
|
||||
body = _post_chat(
|
||||
client, response_format=_json_schema_format(PROBE_SCHEMA, "Probe")
|
||||
)
|
||||
json.loads(body["choices"][0]["message"]["content"])
|
||||
|
||||
|
||||
def test_unreadable_json_schema_still_returns_parseable_json(
|
||||
client: TestClient,
|
||||
) -> None:
|
||||
body = _post_chat(
|
||||
client,
|
||||
response_format={"type": "json_schema", "json_schema": {"name": "Broken"}},
|
||||
)
|
||||
assert json.loads(body["choices"][0]["message"]["content"]) == {}
|
||||
|
||||
|
||||
def test_json_object_mode_recovers_the_schema_from_the_prompt(
|
||||
client: TestClient,
|
||||
) -> None:
|
||||
"""json_object mode puts the schema in the prompt, not in response_format."""
|
||||
body = _post_chat(
|
||||
client,
|
||||
messages=[
|
||||
{"role": "user", "content": "Extract facts."},
|
||||
{
|
||||
"role": "user",
|
||||
"content": "Respond with valid JSON matching this schema:\n"
|
||||
+ json.dumps(PROBE_SCHEMA),
|
||||
},
|
||||
],
|
||||
response_format={"type": "json_object"},
|
||||
)
|
||||
Probe.model_validate_json(body["choices"][0]["message"]["content"])
|
||||
|
||||
|
||||
def test_json_object_mode_without_a_schema_returns_an_empty_object(
|
||||
client: TestClient,
|
||||
) -> None:
|
||||
body = _post_chat(client, response_format={"type": "json_object"})
|
||||
assert json.loads(body["choices"][0]["message"]["content"]) == {}
|
||||
|
||||
|
||||
def test_plain_request_returns_prose(client: TestClient) -> None:
|
||||
body = _post_chat(client)
|
||||
content = body["choices"][0]["message"]["content"]
|
||||
assert "[mock]" in content
|
||||
with pytest.raises(json.JSONDecodeError):
|
||||
json.loads(content)
|
||||
|
||||
|
||||
def test_tools_request_does_not_emit_tool_calls(client: TestClient) -> None:
|
||||
"""The tool loop must terminate; a mock that calls tools would spin."""
|
||||
body = _post_chat(
|
||||
client,
|
||||
tools=[
|
||||
{
|
||||
"type": "function",
|
||||
"function": {"name": "search_memory", "parameters": {}},
|
||||
}
|
||||
],
|
||||
)
|
||||
assert body["choices"][0]["message"]["tool_calls"] is None
|
||||
assert body["choices"][0]["finish_reason"] == "stop"
|
||||
|
||||
|
||||
def test_identical_requests_are_byte_identical(client: TestClient) -> None:
|
||||
payload: dict[str, Any] = {
|
||||
"model": "mock-model",
|
||||
"messages": [{"role": "user", "content": "determinism"}],
|
||||
"response_format": _json_schema_format(PROBE_SCHEMA, "Probe"),
|
||||
}
|
||||
first = client.post("/v1/chat/completions", json=payload).json()
|
||||
second = client.post("/v1/chat/completions", json=payload).json()
|
||||
assert first == second
|
||||
|
||||
|
||||
def test_usage_is_reported(client: TestClient) -> None:
|
||||
body = _post_chat(client)
|
||||
usage = body["usage"]
|
||||
assert usage["total_tokens"] == usage["prompt_tokens"] + usage["completion_tokens"]
|
||||
assert usage["completion_tokens"] > 0
|
||||
|
||||
|
||||
# --- schema generation edge cases -------------------------------------------
|
||||
|
||||
|
||||
def test_recursive_schema_terminates() -> None:
|
||||
"""Reasoning trees nest premises inside conclusions, so cycles are normal."""
|
||||
schema: dict[str, Any] = {
|
||||
"$defs": {
|
||||
"Node": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"value": {"type": "string"},
|
||||
"child": {"anyOf": [{"$ref": "#/$defs/Node"}, {"type": "null"}]},
|
||||
},
|
||||
"required": ["value", "child"],
|
||||
}
|
||||
},
|
||||
"$ref": "#/$defs/Node",
|
||||
}
|
||||
node: dict[str, Any] | None = generate(schema)
|
||||
depth = 0
|
||||
while node is not None and node.get("child") is not None:
|
||||
node = node["child"]
|
||||
depth += 1
|
||||
assert depth < 50, "recursive schema did not terminate"
|
||||
|
||||
|
||||
_SCALAR_CASES: list[tuple[str, dict[str, Any], Callable[[Any], bool]]] = [
|
||||
("enum", {"type": "string", "enum": ["a", "b"]}, lambda v: v in ("a", "b")),
|
||||
("const", {"const": 7}, lambda v: v == 7),
|
||||
("boolean", {"type": "boolean"}, lambda v: isinstance(v, bool)),
|
||||
("null", {"type": "null"}, lambda v: v is None),
|
||||
("number", {"type": "number"}, lambda v: isinstance(v, float)),
|
||||
("nullable-union", {"type": ["string", "null"]}, lambda v: isinstance(v, str)),
|
||||
("pinned-int", {"type": "integer", "minimum": 3, "maximum": 3}, lambda v: v == 3),
|
||||
(
|
||||
"exclusive-bounds",
|
||||
{"type": "integer", "exclusiveMinimum": 1, "exclusiveMaximum": 3},
|
||||
lambda v: v == 2,
|
||||
),
|
||||
(
|
||||
"date-time",
|
||||
{"type": "string", "format": "date-time"},
|
||||
lambda v: str(v).endswith("Z"),
|
||||
),
|
||||
("min-length", {"type": "string", "minLength": 400}, lambda v: len(v) >= 400),
|
||||
("max-length", {"type": "string", "maxLength": 4}, lambda v: len(v) == 4),
|
||||
(
|
||||
"min-items",
|
||||
{"type": "array", "items": {"type": "string"}, "minItems": 3},
|
||||
lambda v: len(v) >= 3,
|
||||
),
|
||||
(
|
||||
"max-items",
|
||||
{"type": "array", "items": {"type": "string"}, "maxItems": 1},
|
||||
lambda v: len(v) == 1,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("schema", "check"),
|
||||
[(schema, check) for _, schema, check in _SCALAR_CASES],
|
||||
ids=[name for name, _, _ in _SCALAR_CASES],
|
||||
)
|
||||
def test_scalar_schema_forms(
|
||||
schema: dict[str, Any], check: Callable[[Any], bool]
|
||||
) -> None:
|
||||
assert check(generate(schema))
|
||||
|
||||
|
||||
def test_all_of_is_flattened() -> None:
|
||||
schema: dict[str, Any] = {
|
||||
"allOf": [
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"a": {"type": "string"}},
|
||||
"required": ["a"],
|
||||
},
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"b": {"type": "integer"}},
|
||||
"required": ["b"],
|
||||
},
|
||||
]
|
||||
}
|
||||
result = generate(schema)
|
||||
assert isinstance(result["a"], str)
|
||||
assert isinstance(result["b"], int)
|
||||
|
||||
|
||||
def test_generation_is_stable_across_calls() -> None:
|
||||
assert generate(PROBE_SCHEMA) == generate(PROBE_SCHEMA)
|
||||
|
||||
|
||||
def test_sibling_fields_of_the_same_type_differ() -> None:
|
||||
"""Path-seeded, so a schema of identical fields is not all one value."""
|
||||
schema: dict[str, Any] = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"first": {"type": "string"},
|
||||
"second": {"type": "string"},
|
||||
},
|
||||
"required": ["first", "second"],
|
||||
}
|
||||
result = generate(schema)
|
||||
assert result["first"] != result["second"]
|
||||
|
||||
|
||||
# --- embeddings -------------------------------------------------------------
|
||||
|
||||
|
||||
def test_embeddings_default_to_1536_and_are_stable(client: TestClient) -> None:
|
||||
payload = {
|
||||
"model": "text-embedding-3-small",
|
||||
"input": "hello",
|
||||
"encoding_format": "float",
|
||||
}
|
||||
first = client.post("/v1/embeddings", json=payload)
|
||||
assert first.status_code == 200, first.text
|
||||
vector = first.json()["data"][0]["embedding"]
|
||||
|
||||
assert len(vector) == 1536
|
||||
assert all(-1.0 <= value <= 1.0 for value in vector)
|
||||
assert client.post("/v1/embeddings", json=payload).json() == first.json()
|
||||
|
||||
|
||||
def test_embeddings_honour_the_requested_dimension(client: TestClient) -> None:
|
||||
"""A width mismatch raises in EmbeddingClient and blocks startup."""
|
||||
response = client.post(
|
||||
"/v1/embeddings",
|
||||
json={"input": "hello", "dimensions": 256, "encoding_format": "float"},
|
||||
)
|
||||
assert len(response.json()["data"][0]["embedding"]) == 256
|
||||
|
||||
|
||||
def test_different_inputs_give_different_vectors(client: TestClient) -> None:
|
||||
response = client.post(
|
||||
"/v1/embeddings",
|
||||
json={"input": ["alpha", "beta"], "encoding_format": "float"},
|
||||
)
|
||||
data = response.json()["data"]
|
||||
assert len(data) == 2
|
||||
assert [item["index"] for item in data] == [0, 1]
|
||||
assert data[0]["embedding"] != data[1]["embedding"]
|
||||
|
||||
|
||||
def test_batch_returns_one_embedding_per_input(client: TestClient) -> None:
|
||||
"""EmbeddingClient._validate_embedding_count rejects any other count."""
|
||||
texts = [f"text-{index}" for index in range(17)]
|
||||
response = client.post(
|
||||
"/v1/embeddings", json={"input": texts, "encoding_format": "float"}
|
||||
)
|
||||
assert len(response.json()["data"]) == len(texts)
|
||||
|
||||
|
||||
def test_base64_is_the_default_encoding_and_decodes_to_the_float_vector(
|
||||
client: TestClient,
|
||||
) -> None:
|
||||
"""The SDK omits encoding_format precisely when it wants base64."""
|
||||
response = client.post("/v1/embeddings", json={"input": "hello"})
|
||||
encoded = response.json()["data"][0]["embedding"]
|
||||
assert isinstance(encoded, str)
|
||||
|
||||
raw = base64.b64decode(encoded)
|
||||
decoded = list(struct.unpack(f"<{len(raw) // 4}f", raw))
|
||||
assert len(decoded) == 1536
|
||||
expected = content_to_embedding("hello", 1536)
|
||||
assert decoded == pytest.approx(expected, abs=1e-6) # pyright: ignore[reportUnknownMemberType]
|
||||
|
||||
|
||||
def test_embedding_matches_the_test_suite_helper() -> None:
|
||||
"""Kept in step with _content_to_embedding in tests/conftest.py.
|
||||
|
||||
Both must derive the same vector from the same text, so a suite that mocks
|
||||
the embedding client in-process and one that talks to this provider over
|
||||
HTTP agree on what a given string embeds to.
|
||||
"""
|
||||
digest = hashlib.sha256(b"hello").digest()
|
||||
expected = [(digest[i % len(digest)] / 255.0) * 2 - 1 for i in range(8)]
|
||||
|
||||
assert content_to_embedding("hello", 8) == pytest.approx(expected) # pyright: ignore[reportUnknownMemberType]
|
||||
|
||||
|
||||
# --- routing ----------------------------------------------------------------
|
||||
|
||||
|
||||
def test_routes_are_mounted_with_and_without_the_v1_prefix(
|
||||
client: TestClient,
|
||||
) -> None:
|
||||
for path in ("/v1/chat/completions", "/chat/completions"):
|
||||
response = client.post(
|
||||
path, json={"model": "m", "messages": [{"role": "user", "content": "x"}]}
|
||||
)
|
||||
assert response.status_code == 200, path
|
||||
|
||||
|
||||
def test_unimplemented_post_returns_405_not_a_plausible_200(
|
||||
client: TestClient,
|
||||
) -> None:
|
||||
"""A catch-all POST would make a missing endpoint look like it worked."""
|
||||
assert client.post("/v1/completions", json={}).status_code == 405
|
||||
|
||||
|
||||
def test_health_and_catch_all_get(client: TestClient) -> None:
|
||||
assert client.get("/health").json()["status"] == "ok"
|
||||
assert client.get("/").status_code == 200
|
||||
|
||||
|
||||
# --- streaming --------------------------------------------------------------
|
||||
|
||||
|
||||
def test_stream_emits_content_then_a_final_usage_chunk(client: TestClient) -> None:
|
||||
"""The backend ends the stream on the usage chunk, so it must come last."""
|
||||
with client.stream(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
json={
|
||||
"model": "mock-model",
|
||||
"messages": [{"role": "user", "content": "stream please"}],
|
||||
"stream": True,
|
||||
"stream_options": {"include_usage": True},
|
||||
},
|
||||
) as response:
|
||||
assert response.status_code == 200
|
||||
lines = [
|
||||
line[len("data: ") :]
|
||||
for line in response.iter_lines()
|
||||
if line.startswith("data: ")
|
||||
]
|
||||
|
||||
assert lines[-1] == "[DONE]"
|
||||
chunks = [json.loads(line) for line in lines[:-1]]
|
||||
|
||||
content = "".join(
|
||||
chunk["choices"][0]["delta"].get("content", "")
|
||||
for chunk in chunks
|
||||
if chunk["choices"]
|
||||
)
|
||||
assert "[mock]" in content
|
||||
|
||||
assert any(
|
||||
chunk["choices"] and chunk["choices"][0]["finish_reason"] == "stop"
|
||||
for chunk in chunks
|
||||
)
|
||||
|
||||
usage_chunk = chunks[-1]
|
||||
assert usage_chunk["usage"]["completion_tokens"] > 0
|
||||
assert usage_chunk["choices"] == []
|
||||
Loading…
Reference in New Issue