mirror of
https://github.com/home-assistant/core.git
synced 2026-09-26 09:23:17 -04:00
Fix strict schema for structured output in openai_conversation (#182723)
This commit is contained in:
Denis Shulyaka
@@ -106,6 +106,7 @@ from .const import (
|
||||
RECOMMENDED_WEB_SEARCH_INLINE_CITATIONS,
|
||||
UNSUPPORTED_EXTENDED_CACHE_RETENTION_MODELS,
|
||||
)
|
||||
from .schema import adjust_schema
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from . import OpenAIConfigEntry
|
||||
@@ -115,31 +116,6 @@ if TYPE_CHECKING:
|
||||
MAX_TOOL_ITERATIONS = 10
|
||||
|
||||
|
||||
def _adjust_schema(schema: dict[str, Any]) -> None:
|
||||
"""Adjust the output schema to be compatible with OpenAI API."""
|
||||
if schema["type"] == "object":
|
||||
schema.setdefault("strict", True)
|
||||
schema.setdefault("additionalProperties", False)
|
||||
if "properties" not in schema:
|
||||
return
|
||||
|
||||
if "required" not in schema:
|
||||
schema["required"] = []
|
||||
|
||||
# Ensure all properties are required
|
||||
for prop, prop_info in schema["properties"].items():
|
||||
_adjust_schema(prop_info)
|
||||
if prop not in schema["required"]:
|
||||
prop_info["type"] = [prop_info["type"], "null"]
|
||||
schema["required"].append(prop)
|
||||
|
||||
elif schema["type"] == "array":
|
||||
if "items" not in schema:
|
||||
return
|
||||
|
||||
_adjust_schema(schema["items"])
|
||||
|
||||
|
||||
def _format_structured_output(
|
||||
schema: probatio.Schema, llm_api: llm.APIInstance | None
|
||||
) -> dict[str, Any]:
|
||||
@@ -152,7 +128,7 @@ def _format_structured_output(
|
||||
openapi_version="3.1.0",
|
||||
)
|
||||
|
||||
_adjust_schema(result)
|
||||
adjust_schema(result)
|
||||
|
||||
return result
|
||||
|
||||
@@ -674,12 +650,11 @@ class OpenAIBaseLLMEntity(Entity):
|
||||
]
|
||||
|
||||
if structure and structure_name:
|
||||
model_args["text"] = {
|
||||
"format": {
|
||||
"type": "json_schema",
|
||||
"name": slugify(structure_name),
|
||||
"schema": _format_structured_output(structure, chat_log.llm_api),
|
||||
},
|
||||
model_args.setdefault("text", {})["format"] = {
|
||||
"type": "json_schema",
|
||||
"name": slugify(structure_name),
|
||||
"schema": _format_structured_output(structure, chat_log.llm_api),
|
||||
"strict": True,
|
||||
}
|
||||
|
||||
client = self.entry.runtime_data
|
||||
|
||||
@@ -0,0 +1,263 @@
|
||||
"""Convert output schemas to OpenAI's supported JSON Schema subset for strict structured output format."""
|
||||
# Documentation: https://developers.openai.com/api/docs/guides/structured-outputs?api-mode=responses#supported-schemas
|
||||
|
||||
from collections.abc import Iterator
|
||||
from copy import deepcopy
|
||||
import logging
|
||||
from typing import Any
|
||||
from urllib.parse import unquote
|
||||
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
_ANNOTATIONS = {
|
||||
"default",
|
||||
"examples",
|
||||
"$comment",
|
||||
"deprecated",
|
||||
"readOnly",
|
||||
"writeOnly",
|
||||
}
|
||||
_UNSUPPORTED_KEYWORDS = {
|
||||
"not",
|
||||
"dependentRequired",
|
||||
"dependentSchemas",
|
||||
"if",
|
||||
"then",
|
||||
"else",
|
||||
}
|
||||
_SELECTOR_FORMATS = {"entity_id", "jinja2", "RFC 5646", "ISO 3166-1 alpha-2", "RGB"}
|
||||
_SCHEMA_MAPS = (
|
||||
"properties",
|
||||
"$defs",
|
||||
"definitions",
|
||||
"patternProperties",
|
||||
"dependentSchemas",
|
||||
)
|
||||
_SCHEMA_LISTS = ("anyOf", "oneOf", "allOf", "prefixItems")
|
||||
_SCHEMA_VALUES = (
|
||||
"items",
|
||||
"contains",
|
||||
"additionalProperties",
|
||||
"propertyNames",
|
||||
"not",
|
||||
"if",
|
||||
"then",
|
||||
"else",
|
||||
)
|
||||
|
||||
|
||||
def adjust_schema(schema: dict[str, Any]) -> None:
|
||||
"""Normalize known incompatibilities, preserving unfamiliar API features."""
|
||||
_stabilize_references(schema)
|
||||
_adjust_schema(schema, "$")
|
||||
if schema.get("type") != "object" or "anyOf" in schema:
|
||||
raise HomeAssistantError("OpenAI structured output requires an object root")
|
||||
|
||||
|
||||
def _walk_schemas(schema: dict[str, Any]) -> Iterator[dict[str, Any]]:
|
||||
"""Walk schema locations without interpreting examples or literal data as schemas."""
|
||||
yield schema
|
||||
for keyword in _SCHEMA_MAPS:
|
||||
for child in schema.get(keyword, {}).values():
|
||||
if isinstance(child, dict):
|
||||
yield from _walk_schemas(child)
|
||||
for keyword in _SCHEMA_LISTS:
|
||||
for child in schema.get(keyword, []):
|
||||
if isinstance(child, dict):
|
||||
yield from _walk_schemas(child)
|
||||
for keyword in _SCHEMA_VALUES:
|
||||
if isinstance(child := schema.get(keyword), dict):
|
||||
yield from _walk_schemas(child)
|
||||
|
||||
|
||||
def _stabilize_references(schema: dict[str, Any]) -> None:
|
||||
"""Move reference targets to root definitions before their paths can change."""
|
||||
references = {node["$ref"] for node in _walk_schemas(schema) if "$ref" in node}
|
||||
replacements: dict[str, str] = {}
|
||||
definitions: dict[str, Any] = {}
|
||||
existing_names = set(schema.get("$defs", {}))
|
||||
for reference in sorted(references):
|
||||
target = _resolve_reference(reference, schema)
|
||||
parts = _reference_parts(reference)
|
||||
if not parts or (len(parts) == 2 and parts[0] == "$defs"):
|
||||
replacements[reference] = reference
|
||||
continue
|
||||
name = f"_ha_ref_{len(definitions)}"
|
||||
while name in existing_names:
|
||||
name = f"_{name}"
|
||||
existing_names.add(name)
|
||||
definitions[name] = deepcopy(target)
|
||||
replacements[reference] = f"#/$defs/{name}"
|
||||
|
||||
if definitions:
|
||||
schema.setdefault("$defs", {}).update(definitions)
|
||||
# Legacy definitions have been copied to $defs wherever they are referenced.
|
||||
for node in _walk_schemas(schema):
|
||||
node.pop("definitions", None)
|
||||
if "$ref" in node:
|
||||
node["$ref"] = replacements[node["$ref"]]
|
||||
|
||||
|
||||
def _reference_parts(reference: str) -> list[str]:
|
||||
"""Decode a local JSON Pointer, including URI fragment escaping."""
|
||||
if reference == "#":
|
||||
return []
|
||||
if not reference.startswith("#/"):
|
||||
raise HomeAssistantError(
|
||||
f"Unsupported OpenAI output schema reference: {reference}"
|
||||
)
|
||||
return [
|
||||
part.replace("~1", "/").replace("~0", "~")
|
||||
for part in unquote(reference[2:]).split("/")
|
||||
]
|
||||
|
||||
|
||||
def _resolve_reference(reference: str, root: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Resolve local references through both objects and arrays."""
|
||||
target: Any = root
|
||||
try:
|
||||
for part in _reference_parts(reference):
|
||||
if isinstance(target, list):
|
||||
target = target[_array_index(part)]
|
||||
else:
|
||||
target = target[part]
|
||||
except (KeyError, IndexError, TypeError, ValueError) as err:
|
||||
raise HomeAssistantError(
|
||||
f"Invalid OpenAI output schema reference: {reference}"
|
||||
) from err
|
||||
if not isinstance(target, dict):
|
||||
raise HomeAssistantError(
|
||||
f"Unsupported OpenAI output schema reference: {reference}"
|
||||
)
|
||||
return target
|
||||
|
||||
|
||||
def _array_index(value: str) -> int:
|
||||
"""Parse the array-index form of a JSON Pointer token."""
|
||||
if (
|
||||
not value.isascii()
|
||||
or not value.isdecimal()
|
||||
or (len(value) > 1 and value[0] == "0")
|
||||
):
|
||||
raise ValueError("Invalid array index")
|
||||
return int(value)
|
||||
|
||||
|
||||
def _flatten_all_of(schema: dict[str, Any], path: str) -> None:
|
||||
"""Unwrap intersections only when sibling constraints can all be retained."""
|
||||
while "allOf" in schema:
|
||||
branches = schema["allOf"]
|
||||
if len(branches) != 1 or not isinstance(branches[0], dict):
|
||||
raise HomeAssistantError(
|
||||
f"Unsupported OpenAI output schema allOf at {path}"
|
||||
)
|
||||
branch = branches[0]
|
||||
siblings = schema.keys() - {"allOf", "description", "title"} - _ANNOTATIONS
|
||||
conflicts = {
|
||||
key for key in siblings & branch.keys() if schema[key] != branch[key]
|
||||
}
|
||||
if conflicts:
|
||||
raise HomeAssistantError(
|
||||
f"Conflicting OpenAI output schema allOf at {path}: {', '.join(sorted(conflicts))}"
|
||||
)
|
||||
del schema["allOf"]
|
||||
for key, value in branch.items():
|
||||
schema.setdefault(key, value)
|
||||
|
||||
|
||||
def _adjust_reference(schema: dict[str, Any], path: str, *, nullable: bool) -> None:
|
||||
"""Keep references bare and preserve annotations on nullable wrappers."""
|
||||
if siblings := schema.keys() - {"$ref", "title", "description"}:
|
||||
raise HomeAssistantError(
|
||||
f"Unsupported OpenAI output schema reference siblings at {path}: {', '.join(sorted(siblings))}"
|
||||
)
|
||||
annotations: dict[str, Any] = {
|
||||
keyword: schema.pop(keyword)
|
||||
for keyword in ("title", "description")
|
||||
if keyword in schema
|
||||
}
|
||||
if nullable:
|
||||
_make_nullable(schema)
|
||||
schema.update(annotations)
|
||||
elif annotations:
|
||||
_LOGGER.debug(
|
||||
"Removed reference annotations %s from OpenAI output schema at %s",
|
||||
", ".join(annotations),
|
||||
path,
|
||||
)
|
||||
|
||||
|
||||
def _adjust_schema(
|
||||
schema: dict[str, Any] | bool, path: str, *, nullable: bool = False
|
||||
) -> None:
|
||||
"""Normalize nested schemas and keep unsupported enforcement out of requests."""
|
||||
if not isinstance(schema, dict):
|
||||
raise HomeAssistantError(f"Unsupported OpenAI output schema at {path}")
|
||||
_flatten_all_of(schema, path)
|
||||
for keyword in _ANNOTATIONS:
|
||||
schema.pop(keyword, None)
|
||||
if unsupported := schema.keys() & _UNSUPPORTED_KEYWORDS:
|
||||
raise HomeAssistantError(
|
||||
f"Unsupported OpenAI output schema keywords at {path}: {', '.join(sorted(unsupported))}"
|
||||
)
|
||||
if not schema:
|
||||
raise HomeAssistantError(f"Unsupported OpenAI output schema at {path}")
|
||||
if schema.get("format") in _SELECTOR_FORMATS:
|
||||
del schema["format"]
|
||||
if schema.pop("uniqueItems", None) is True:
|
||||
_LOGGER.debug(
|
||||
"Removed unsupported uniqueItems: true from OpenAI output schema at %s",
|
||||
path,
|
||||
)
|
||||
if "$ref" in schema:
|
||||
_adjust_reference(schema, path, nullable=nullable)
|
||||
return
|
||||
|
||||
for name, definition in schema.get("$defs", {}).items():
|
||||
_adjust_schema(definition, f"{path}.$defs.{name}")
|
||||
for keyword in ("anyOf", "oneOf"):
|
||||
for index, variant in enumerate(schema.get(keyword, [])):
|
||||
_adjust_schema(variant, f"{path}.{keyword}[{index}]")
|
||||
|
||||
schema_type = schema.get("type", [])
|
||||
types = [schema_type] if isinstance(schema_type, str) else schema_type
|
||||
if "object" in types:
|
||||
if schema.get("additionalProperties", False) is not False:
|
||||
raise HomeAssistantError(
|
||||
f"OpenAI output schema requires explicitly defined object fields at {path}"
|
||||
)
|
||||
schema["additionalProperties"] = False
|
||||
properties = schema.setdefault("properties", {})
|
||||
required = schema.setdefault("required", [])
|
||||
for name, prop in properties.items():
|
||||
_adjust_schema(
|
||||
prop, f"{path}.properties.{name}", nullable=name not in required
|
||||
)
|
||||
if name not in required:
|
||||
required.append(name)
|
||||
if "array" in types:
|
||||
if "items" not in schema:
|
||||
raise HomeAssistantError(
|
||||
f"OpenAI output schema requires array items at {path}"
|
||||
)
|
||||
_adjust_schema(schema["items"], f"{path}.items")
|
||||
if nullable:
|
||||
_make_nullable(schema)
|
||||
|
||||
|
||||
def _make_nullable(schema: dict[str, Any]) -> None:
|
||||
"""Allow null without weakening the non-null schema's constraints."""
|
||||
if "type" not in schema or schema.keys() & {"$ref", "const", "anyOf", "oneOf"}:
|
||||
original = schema.copy()
|
||||
schema.clear()
|
||||
schema["anyOf"] = [original, {"type": "null"}]
|
||||
return
|
||||
schema_type = schema["type"]
|
||||
types = [schema_type] if isinstance(schema_type, str) else schema_type
|
||||
if "null" not in types:
|
||||
types.append("null")
|
||||
schema["type"] = types
|
||||
if "enum" in schema and None not in schema["enum"]:
|
||||
schema["enum"].append(None)
|
||||
@@ -0,0 +1,718 @@
|
||||
# serializer version: 1
|
||||
# name: test_nested_references
|
||||
dict({
|
||||
'$defs': dict({
|
||||
'node': dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'children': dict({
|
||||
'items': dict({
|
||||
'$ref': '#/$defs/node',
|
||||
}),
|
||||
'type': list([
|
||||
'array',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
'parent': dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'$ref': '#',
|
||||
}),
|
||||
dict({
|
||||
'type': 'null',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
'value': dict({
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
'variant': dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'name': dict({
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'name',
|
||||
]),
|
||||
'type': 'object',
|
||||
}),
|
||||
dict({
|
||||
'type': 'integer',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
dict({
|
||||
'type': 'null',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
'children',
|
||||
'parent',
|
||||
'variant',
|
||||
]),
|
||||
'type': list([
|
||||
'object',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'node': dict({
|
||||
'$ref': '#/$defs/node',
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'node',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_optional_fields[already-nullable]
|
||||
dict({
|
||||
'$defs': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_optional_fields[constant]
|
||||
dict({
|
||||
'$defs': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'const': 'a',
|
||||
'type': 'string',
|
||||
}),
|
||||
dict({
|
||||
'type': 'null',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_optional_fields[constrained-union]
|
||||
dict({
|
||||
'$defs': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
}),
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
dict({
|
||||
'type': 'null',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_optional_fields[enum]
|
||||
dict({
|
||||
'$defs': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
None,
|
||||
]),
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_optional_fields[nullable-enum]
|
||||
dict({
|
||||
'$defs': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
None,
|
||||
]),
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_optional_fields[reference]
|
||||
dict({
|
||||
'$defs': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'$ref': '#/$defs/value',
|
||||
}),
|
||||
dict({
|
||||
'type': 'null',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_optional_fields[string]
|
||||
dict({
|
||||
'$defs': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_optional_fields[union]
|
||||
dict({
|
||||
'$defs': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'type': 'string',
|
||||
}),
|
||||
dict({
|
||||
'type': 'integer',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
dict({
|
||||
'type': 'null',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_recoverable_schemas[all-of-siblings]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'maximum': 5,
|
||||
'minimum': 0,
|
||||
'type': 'number',
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_recoverable_schemas[annotations]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_recoverable_schemas[future-features]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'format': 'future-format',
|
||||
'futureConstraint': 'new',
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_recoverable_schemas[no-uniqueness]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'items': dict({
|
||||
'type': 'string',
|
||||
}),
|
||||
'type': 'array',
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_recoverable_schemas[single-all-of]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_recursive_reference_annotations_removed[array-items]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'children': dict({
|
||||
'items': dict({
|
||||
'$ref': '#',
|
||||
}),
|
||||
'type': 'array',
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'children',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_recursive_reference_annotations_removed[array-items][logs]
|
||||
list([
|
||||
'Removed reference annotations title, description from OpenAI output schema at $.properties.children.items',
|
||||
])
|
||||
# ---
|
||||
# name: test_recursive_reference_annotations_removed[required]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'children': dict({
|
||||
'$ref': '#',
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'children',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_recursive_reference_annotations_removed[required][logs]
|
||||
list([
|
||||
'Removed reference annotations title, description from OpenAI output schema at $.properties.children',
|
||||
])
|
||||
# ---
|
||||
# name: test_reference_annotations[optional]
|
||||
dict({
|
||||
'$defs': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'$ref': '#/$defs/value',
|
||||
}),
|
||||
dict({
|
||||
'type': 'null',
|
||||
}),
|
||||
]),
|
||||
'description': 'A value',
|
||||
'title': 'Value',
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_reference_annotations[optional][logs]
|
||||
list([
|
||||
])
|
||||
# ---
|
||||
# name: test_reference_annotations[required]
|
||||
dict({
|
||||
'$defs': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
}),
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'$ref': '#/$defs/value',
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_reference_annotations[required][logs]
|
||||
list([
|
||||
'Removed reference annotations title, description from OpenAI output schema at $.properties.value',
|
||||
])
|
||||
# ---
|
||||
# name: test_selector_schemas[country]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_selector_schemas[date]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'format': 'date',
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_selector_schemas[entity]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_selector_schemas[enum]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
None,
|
||||
]),
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_selector_schemas[language]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_selector_schemas[multi-select]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'items': dict({
|
||||
'enum': list([
|
||||
'a',
|
||||
'b',
|
||||
]),
|
||||
'type': 'string',
|
||||
}),
|
||||
'type': list([
|
||||
'array',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_selector_schemas[number]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'maximum': 120.0,
|
||||
'minimum': 0.0,
|
||||
'type': list([
|
||||
'number',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_selector_schemas[rgb]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'items': dict({
|
||||
'type': 'number',
|
||||
}),
|
||||
'maxItems': 3,
|
||||
'minItems': 3,
|
||||
'type': list([
|
||||
'array',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_selector_schemas[template]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'type': list([
|
||||
'string',
|
||||
'null',
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
# name: test_selector_schemas[union]
|
||||
dict({
|
||||
'additionalProperties': False,
|
||||
'properties': dict({
|
||||
'value': dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'anyOf': list([
|
||||
dict({
|
||||
'type': 'string',
|
||||
}),
|
||||
dict({
|
||||
'type': 'integer',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
dict({
|
||||
'type': 'null',
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
}),
|
||||
'required': list([
|
||||
'value',
|
||||
]),
|
||||
'type': 'object',
|
||||
})
|
||||
# ---
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Test AI Task platform of OpenAI Conversation integration."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
@@ -11,8 +12,10 @@ import pytest
|
||||
from homeassistant.components import ai_task, media_source
|
||||
from homeassistant.components.openai_conversation import DOMAIN
|
||||
from homeassistant.components.openai_conversation.const import (
|
||||
CONF_CHAT_MODEL,
|
||||
CONF_IMAGE_MODEL,
|
||||
CONF_STORE_RESPONSES,
|
||||
CONF_VERBOSITY,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
@@ -76,13 +79,39 @@ async def test_generate_data(
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_init_component")
|
||||
@pytest.mark.parametrize(
|
||||
("model", "verbosity", "expected_verbosity"),
|
||||
[
|
||||
pytest.param("gpt-4o-mini", "low", None, id="without-verbosity"),
|
||||
pytest.param("gpt-5-mini", "low", "low", id="low-verbosity"),
|
||||
pytest.param("gpt-5-mini", "high", "high", id="high-verbosity"),
|
||||
],
|
||||
)
|
||||
async def test_generate_structured_data(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_create_stream: AsyncMock,
|
||||
entity_registry: er.EntityRegistry,
|
||||
model: str,
|
||||
verbosity: str,
|
||||
expected_verbosity: str | None,
|
||||
) -> None:
|
||||
"""Test AI Task structured data generation."""
|
||||
ai_task_entry = next(
|
||||
entry
|
||||
for entry in mock_config_entry.subentries.values()
|
||||
if entry.subentry_type == "ai_task_data"
|
||||
)
|
||||
hass.config_entries.async_update_subentry(
|
||||
mock_config_entry,
|
||||
ai_task_entry,
|
||||
data={
|
||||
**ai_task_entry.data,
|
||||
CONF_CHAT_MODEL: model,
|
||||
CONF_VERBOSITY: verbosity,
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Mock the OpenAI response stream with JSON data
|
||||
mock_create_stream.return_value = [
|
||||
create_message_item(
|
||||
@@ -109,6 +138,9 @@ async def test_generate_structured_data(
|
||||
)
|
||||
|
||||
assert result.data == {"characters": ["Mario", "Luigi"]}
|
||||
text = mock_create_stream.call_args.kwargs["text"]
|
||||
assert text["format"]["strict"] is True
|
||||
assert text.get("verbosity") == expected_verbosity
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_init_component")
|
||||
@@ -146,6 +178,56 @@ async def test_generate_invalid_structured_data(
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def selection_structure() -> probatio.Schema:
|
||||
"""A multi-select with optional fields represented as null on the wire."""
|
||||
return probatio.Schema(
|
||||
{
|
||||
probatio.Optional("names"): selector.SelectSelector(
|
||||
{"options": ["a", "b"], "multiple": True}
|
||||
),
|
||||
probatio.Optional("label"): selector.TextSelector(),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_init_component")
|
||||
@pytest.mark.parametrize(
|
||||
"names",
|
||||
[
|
||||
pytest.param(["a", "b"], id="selection"),
|
||||
pytest.param(["a", "a"], id="duplicates"),
|
||||
pytest.param(None, id="omitted"),
|
||||
],
|
||||
)
|
||||
async def test_generate_selection(
|
||||
hass: HomeAssistant,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
mock_create_stream: AsyncMock,
|
||||
selection_structure: probatio.Schema,
|
||||
names: list[str] | None,
|
||||
) -> None:
|
||||
"""Generate multi-select results while accepting duplicates and optional nulls."""
|
||||
data = {"names": names, "label": None}
|
||||
mock_create_stream.return_value = [
|
||||
create_message_item(id="msg_A", text=json.dumps(data), output_index=0)
|
||||
]
|
||||
result = await ai_task.async_generate_data(
|
||||
hass,
|
||||
task_name="Selection",
|
||||
entity_id="ai_task.openai_ai_task",
|
||||
instructions="Select names",
|
||||
structure=selection_structure,
|
||||
)
|
||||
assert result.data == data
|
||||
schema = mock_create_stream.call_args.kwargs["text"]["format"]["schema"]
|
||||
assert "uniqueItems" not in schema["properties"]["names"]
|
||||
assert (
|
||||
"Removed unsupported uniqueItems: true from OpenAI output schema at $.properties.names"
|
||||
in caplog.text
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_init_component")
|
||||
async def test_generate_data_with_attachments(
|
||||
hass: HomeAssistant,
|
||||
|
||||
@@ -71,7 +71,6 @@ async def test_format_structured_output() -> None:
|
||||
],
|
||||
"type": "object",
|
||||
"additionalProperties": False,
|
||||
"strict": True,
|
||||
},
|
||||
"type": "array",
|
||||
},
|
||||
@@ -81,7 +80,6 @@ async def test_format_structured_output() -> None:
|
||||
"stuff",
|
||||
"age",
|
||||
],
|
||||
"strict": True,
|
||||
"type": "object",
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,398 @@
|
||||
"""Test conversion of structured output schemas."""
|
||||
|
||||
from copy import deepcopy
|
||||
from typing import Any
|
||||
|
||||
import probatio
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.openai_conversation.entity import (
|
||||
_format_structured_output,
|
||||
)
|
||||
from homeassistant.components.openai_conversation.schema import adjust_schema
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import selector
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"field",
|
||||
[
|
||||
pytest.param({"type": "string"}, id="string"),
|
||||
pytest.param({"type": ["string", "null"]}, id="already-nullable"),
|
||||
pytest.param({"type": "string", "enum": ["a", "b"]}, id="enum"),
|
||||
pytest.param(
|
||||
{"type": ["string", "null"], "enum": ["a", None]}, id="nullable-enum"
|
||||
),
|
||||
pytest.param({"type": "string", "const": "a"}, id="constant"),
|
||||
pytest.param({"anyOf": [{"type": "string"}, {"type": "integer"}]}, id="union"),
|
||||
pytest.param(
|
||||
{"type": "string", "anyOf": [{"enum": ["a", "b"]}]},
|
||||
id="constrained-union",
|
||||
),
|
||||
pytest.param({"$ref": "#/$defs/value"}, id="reference"),
|
||||
],
|
||||
)
|
||||
def test_optional_fields(field: dict[str, Any], snapshot: SnapshotAssertion) -> None:
|
||||
"""Optional fields accept null while preserving their non-null constraints."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {"value": field},
|
||||
"$defs": {"value": {"type": "string", "enum": ["a", "b"]}},
|
||||
}
|
||||
adjust_schema(schema)
|
||||
|
||||
validator = probatio.from_json_schema(schema)
|
||||
validator({"value": None})
|
||||
validator({"value": "a"})
|
||||
with pytest.raises(probatio.Invalid):
|
||||
validator({})
|
||||
with pytest.raises(probatio.Invalid):
|
||||
validator({"value": []})
|
||||
assert schema == snapshot
|
||||
|
||||
|
||||
def test_nested_references(snapshot: SnapshotAssertion) -> None:
|
||||
"""Normalize referenced objects and unions without expanding recursive refs."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {"node": {"$ref": "#/$defs/node"}},
|
||||
"required": ["node"],
|
||||
"$defs": {
|
||||
"node": {
|
||||
"type": ["object", "null"],
|
||||
"properties": {
|
||||
"value": {"type": "string"},
|
||||
"children": {"type": "array", "items": {"$ref": "#/$defs/node"}},
|
||||
"parent": {"$ref": "#"},
|
||||
"variant": {
|
||||
"anyOf": [
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"name": {"type": "string"}},
|
||||
},
|
||||
{"type": "integer"},
|
||||
]
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
adjust_schema(schema)
|
||||
|
||||
validator = probatio.from_json_schema(schema)
|
||||
validator(
|
||||
{
|
||||
"node": {
|
||||
"value": "a",
|
||||
"children": [],
|
||||
"parent": None,
|
||||
"variant": {"name": "b"},
|
||||
}
|
||||
}
|
||||
)
|
||||
with pytest.raises(probatio.Invalid):
|
||||
validator({"node": {"value": "a"}})
|
||||
assert schema == snapshot
|
||||
|
||||
|
||||
def test_recursive_reference_description(caplog: pytest.LogCaptureFixture) -> None:
|
||||
"""Preserve a recursive field's description without expanding its reference."""
|
||||
schema = _format_structured_output(
|
||||
probatio.Schema(
|
||||
{
|
||||
probatio.Optional("child", description="The next node"): probatio.Self,
|
||||
}
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
assert schema["properties"]["child"] == {
|
||||
"anyOf": [{"$ref": "#"}, {"type": "null"}],
|
||||
"description": "The next node",
|
||||
}
|
||||
validator = probatio.from_json_schema(schema)
|
||||
validator({"child": {"child": None}})
|
||||
with pytest.raises(probatio.Invalid):
|
||||
validator({"child": {"child": "invalid"}})
|
||||
assert "Removed reference annotations" not in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"required",
|
||||
[pytest.param([], id="optional"), pytest.param(["value"], id="required")],
|
||||
)
|
||||
def test_reference_annotations(
|
||||
required: list[str], snapshot: SnapshotAssertion, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Keep annotations on nullable wrappers and log removals elsewhere."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"value": {
|
||||
"$ref": "#/$defs/value",
|
||||
"title": "Value",
|
||||
"description": "A value",
|
||||
}
|
||||
},
|
||||
"required": required.copy(),
|
||||
"$defs": {"value": {"type": "string", "enum": ["a", "b"]}},
|
||||
}
|
||||
adjust_schema(schema)
|
||||
|
||||
assert schema == snapshot
|
||||
assert caplog.messages == snapshot(name="logs")
|
||||
validator = probatio.from_json_schema(schema)
|
||||
validator({"value": "a"})
|
||||
with pytest.raises(probatio.Invalid):
|
||||
validator({"value": "c"})
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"field",
|
||||
[
|
||||
pytest.param({"$ref": "#"}, id="required"),
|
||||
pytest.param({"type": "array", "items": {"$ref": "#"}}, id="array-items"),
|
||||
],
|
||||
)
|
||||
def test_recursive_reference_annotations_removed(
|
||||
field: dict[str, Any], snapshot: SnapshotAssertion, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Recursive references stay bare when no nullable wrapper is needed."""
|
||||
field = deepcopy(field)
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {"children": field},
|
||||
"required": ["children"],
|
||||
}
|
||||
target = field.get("items", field)
|
||||
target.update({"title": "Children", "description": "Child nodes"})
|
||||
adjust_schema(schema)
|
||||
|
||||
assert schema == snapshot
|
||||
assert caplog.messages == snapshot(name="logs")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"required",
|
||||
[pytest.param([], id="optional"), pytest.param(["value"], id="required")],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("field", "keyword"),
|
||||
[
|
||||
pytest.param(
|
||||
{"$ref": "#/$defs/value", "maxLength": 10}, "maxLength", id="length"
|
||||
),
|
||||
pytest.param(
|
||||
{"$ref": "#/$defs/value", "allOf": [{"maxLength": 10}]},
|
||||
"maxLength",
|
||||
id="all-of",
|
||||
),
|
||||
pytest.param(
|
||||
{"$ref": "#", "maxProperties": 1}, "maxProperties", id="recursive"
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_reference_constraint_siblings(
|
||||
field: dict[str, Any], keyword: str, required: list[str]
|
||||
) -> None:
|
||||
"""Reject reference constraints instead of dropping them or expanding cycles."""
|
||||
with pytest.raises(HomeAssistantError, match="reference siblings") as err:
|
||||
adjust_schema(
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"value": deepcopy(field)},
|
||||
"required": required.copy(),
|
||||
"$defs": {"value": {"type": "string"}},
|
||||
}
|
||||
)
|
||||
assert "$.properties.value" in str(err.value)
|
||||
assert keyword in str(err.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"field",
|
||||
[
|
||||
pytest.param(selector.EntitySelector(), id="entity"),
|
||||
pytest.param(selector.TemplateSelector(), id="template"),
|
||||
pytest.param(selector.CountrySelector(), id="country"),
|
||||
pytest.param(selector.LanguageSelector(), id="language"),
|
||||
pytest.param(selector.ColorRGBSelector(), id="rgb"),
|
||||
pytest.param(selector.DateSelector(), id="date"),
|
||||
pytest.param(selector.NumberSelector({"min": 0, "max": 120}), id="number"),
|
||||
pytest.param(selector.SelectSelector({"options": ["a", "b"]}), id="enum"),
|
||||
pytest.param(
|
||||
selector.SelectSelector({"options": ["a", "b"], "multiple": True}),
|
||||
id="multi-select",
|
||||
),
|
||||
pytest.param(probatio.Any(str, int), id="union"),
|
||||
],
|
||||
)
|
||||
def test_selector_schemas(
|
||||
field: selector.Selector | probatio.Any, snapshot: SnapshotAssertion
|
||||
) -> None:
|
||||
"""Convert actual selectors while preserving supported constraints."""
|
||||
schema = _format_structured_output(
|
||||
probatio.Schema({probatio.Optional("value"): field}), None
|
||||
)
|
||||
assert schema == snapshot
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("field", "message"),
|
||||
[
|
||||
pytest.param(
|
||||
selector.ObjectSelector(),
|
||||
"explicitly defined object fields",
|
||||
id="free-object",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_unsupported_selectors(field: selector.Selector, message: str) -> None:
|
||||
"""Reject constraints that cannot be preserved in strict output."""
|
||||
with pytest.raises(HomeAssistantError, match=message):
|
||||
_format_structured_output(
|
||||
probatio.Schema({probatio.Required("value"): field}), None
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("field", "message"),
|
||||
[
|
||||
pytest.param(
|
||||
{"allOf": [{"type": "string"}, {"type": "integer"}]}, "allOf", id="all-of"
|
||||
),
|
||||
pytest.param(
|
||||
{"type": "string", "allOf": [{"type": "integer"}]},
|
||||
"Conflicting",
|
||||
id="conflicting-all-of",
|
||||
),
|
||||
pytest.param({"type": "string", "not": {"enum": ["a"]}}, "not", id="not"),
|
||||
pytest.param({"type": "array"}, "array items", id="array-without-items"),
|
||||
pytest.param({}, "schema", id="unconstrained"),
|
||||
pytest.param(True, "schema", id="boolean-schema"),
|
||||
],
|
||||
)
|
||||
def test_unsupported_schema(field: dict[str, Any] | bool, message: str) -> None:
|
||||
"""Unsupported schemas fail locally with their field location."""
|
||||
with pytest.raises(HomeAssistantError, match=message) as err:
|
||||
adjust_schema({"type": "object", "properties": {"value": field}})
|
||||
assert "$.properties.value" in str(err.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"schema",
|
||||
[
|
||||
pytest.param({"type": "array", "items": {"type": "string"}}, id="array"),
|
||||
pytest.param({"anyOf": [{"type": "object"}]}, id="union"),
|
||||
],
|
||||
)
|
||||
def test_invalid_root(schema: dict[str, Any]) -> None:
|
||||
"""Require an object at the root of a structured output schema."""
|
||||
with pytest.raises(HomeAssistantError, match="object root"):
|
||||
adjust_schema(schema)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"field",
|
||||
[
|
||||
pytest.param(
|
||||
{
|
||||
"type": "string",
|
||||
"examples": ["a"],
|
||||
"$comment": "hint",
|
||||
"deprecated": False,
|
||||
},
|
||||
id="annotations",
|
||||
),
|
||||
pytest.param(
|
||||
{"type": "array", "items": {"type": "string"}, "uniqueItems": False},
|
||||
id="no-uniqueness",
|
||||
),
|
||||
pytest.param({"allOf": [{"type": "string"}]}, id="single-all-of"),
|
||||
pytest.param(
|
||||
{"minimum": 0, "allOf": [{"type": "number", "maximum": 5}]},
|
||||
id="all-of-siblings",
|
||||
),
|
||||
pytest.param(
|
||||
{"type": "string", "format": "future-format", "futureConstraint": "new"},
|
||||
id="future-features",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_recoverable_schemas(
|
||||
field: dict[str, Any], snapshot: SnapshotAssertion
|
||||
) -> None:
|
||||
"""Recover safely and let the API decide whether it supports new features."""
|
||||
schema = {"type": "object", "properties": {"value": field}, "required": ["value"]}
|
||||
adjust_schema(schema)
|
||||
assert schema == snapshot
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"reference",
|
||||
[
|
||||
pytest.param("#/$defs/missing", id="missing"),
|
||||
pytest.param("https://example.com/schema", id="remote"),
|
||||
pytest.param("#/properties/value/anyOf/01", id="invalid-index"),
|
||||
pytest.param("#/properties/value/anyOf/9", id="out-of-range"),
|
||||
],
|
||||
)
|
||||
def test_invalid_reference(reference: str) -> None:
|
||||
"""Report invalid references before modifying their targets."""
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"value": {"anyOf": [{"type": "string"}]},
|
||||
"alias": {"$ref": reference},
|
||||
},
|
||||
}
|
||||
with pytest.raises(HomeAssistantError, match="reference"):
|
||||
adjust_schema(schema)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("alias_first", [True, False])
|
||||
@pytest.mark.parametrize(
|
||||
("target", "reference"),
|
||||
[
|
||||
pytest.param(
|
||||
{"anyOf": [{"type": "string"}, {"type": "integer"}]},
|
||||
"#/properties/value/anyOf/0",
|
||||
id="array-pointer",
|
||||
),
|
||||
pytest.param(
|
||||
{"const": {}, "type": "object", "properties": {"name": {"type": "string"}}},
|
||||
"#/properties/value/properties/name",
|
||||
id="moved-target",
|
||||
),
|
||||
pytest.param(
|
||||
{"allOf": [{"type": "string"}]},
|
||||
"#/properties/value/allOf/0",
|
||||
id="unwrapped-target",
|
||||
),
|
||||
pytest.param(
|
||||
{"type": "object", "properties": {"a/b~c d": {"type": "string"}}},
|
||||
"#/properties/value/properties/a~1b~0c%20d",
|
||||
id="escaped-pointer",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_reference_targets(
|
||||
target: dict[str, Any], reference: str, alias_first: bool
|
||||
) -> None:
|
||||
"""References keep their constraints regardless of traversal order or wrapping."""
|
||||
fields = [("value", deepcopy(target)), ("alias", {"$ref": "#/$defs/alias"})]
|
||||
fields.sort(key=lambda field: (field[0] == "alias") != alias_first)
|
||||
schema = {
|
||||
"type": "object",
|
||||
"$defs": {"alias": {"$ref": reference}},
|
||||
"properties": dict(fields),
|
||||
"required": ["alias"],
|
||||
}
|
||||
adjust_schema(schema)
|
||||
validator = probatio.from_json_schema(schema)
|
||||
validator({"value": None, "alias": "a"})
|
||||
with pytest.raises(probatio.Invalid):
|
||||
validator({"value": None, "alias": 1})
|
||||
with pytest.raises(probatio.Invalid):
|
||||
validator({"value": None, "alias": None})
|
||||
Reference in New Issue
Block a user