From f2566767f8a72446bc6329258c3b09ac63df173f Mon Sep 17 00:00:00 2001 From: Allen Porter Date: Thu, 25 Jun 2026 08:04:10 -0700 Subject: [PATCH] Fix voluptuous-openapi LazySchema unhashable type bug --- tests/test_lib.py | 130 +++++++++++++++++++++++++++++++++ voluptuous_openapi/__init__.py | 48 ++++++++++-- 2 files changed, 173 insertions(+), 5 deletions(-) diff --git a/tests/test_lib.py b/tests/test_lib.py index a230dba..fd4fdb4 100644 --- a/tests/test_lib.py +++ b/tests/test_lib.py @@ -1315,3 +1315,133 @@ def test_anonymized_ghp_home_automation_invalid() -> None: } with pytest.raises(vol.Invalid): validator(invalid_automation) + + +def test_convert_schema_with_reference() -> None: + """Test that converting a voluptuous schema containing a reference back to OpenAPI works.""" + schema = { + "$defs": { + "PositiveInteger": { + "type": "integer", + "minimum": 0, + } + }, + "type": "object", + "properties": { + "value": {"$ref": "#/$defs/PositiveInteger"}, + }, + } + vol_schema = convert_to_voluptuous(schema) + + # Verify that the returned vol_schema validates correctly + assert vol_schema({"value": 42}) == {"value": 42} + with pytest.raises(vol.Invalid): + vol_schema({"value": -5}) + + # Verify that convert() successfully serializes the voluptuous schema + # and denormalizes the reference. + res = convert(vol_schema) + assert res == { + "type": "object", + "properties": { + "value": {"type": "integer", "minimum": 0}, + }, + "required": [], + } + + +def test_convert_schema_with_nested_reference() -> None: + """Test that converting a voluptuous schema containing a nested reference back to OpenAPI works.""" + schema = { + "$defs": { + "PositiveInteger": { + "type": "integer", + "minimum": 0, + } + }, + "type": "object", + "properties": { + "nested": { + "type": "object", + "properties": { + "value": {"$ref": "#/$defs/PositiveInteger"}, + }, + } + }, + } + vol_schema = convert_to_voluptuous(schema) + + # Verify that the returned vol_schema validates correctly + assert vol_schema({"nested": {"value": 42}}) == {"nested": {"value": 42}} + with pytest.raises(vol.Invalid): + vol_schema({"nested": {"value": -5}}) + + # Verify that convert() successfully serializes the voluptuous schema and denormalizes the nested reference. + res = convert(vol_schema) + assert res == { + "type": "object", + "properties": { + "nested": { + "type": "object", + "properties": { + "value": {"type": "integer", "minimum": 0}, + }, + "required": [], + } + }, + "required": [], + } + + +def test_convert_recursive_schema() -> None: + """Test that converting a voluptuous schema containing a recursive reference back to OpenAPI works.""" + schema = { + "$defs": { + "Node": { + "type": "object", + "properties": { + "value": {"type": "string"}, + "child": {"$ref": "#/$defs/Node"}, + }, + } + }, + "$ref": "#/$defs/Node", + } + vol_schema = convert_to_voluptuous(schema) + + # Verify that the returned vol_schema validates correctly + valid_data = {"value": "root", "child": {"value": "child"}} + assert vol_schema(valid_data) == valid_data + + # Verify that convert() successfully serializes the voluptuous schema, + # denormalizes the reference, and breaks the cycle at "child". + res = convert(vol_schema) + assert res == { + "type": "object", + "properties": { + "value": {"type": "string"}, + "child": {"type": "string"}, + }, + "required": [], + } + + +def test_convert_custom_callable_class_instance() -> None: + """Test that convert() handles custom callable validator instances.""" + + class CustomValidator: + def __call__(self, value: int) -> int: + if value < 0: + raise vol.Invalid("Must be positive") + return value + + vol_schema = vol.Schema(CustomValidator()) + + # Verify native validation works + assert vol_schema(42) == 42 + with pytest.raises(vol.Invalid): + vol_schema(-5) + + # Verify serialization extracts the type hint from __call__ parameter + res = convert(vol_schema) + assert res == {"type": "integer"} diff --git a/voluptuous_openapi/__init__.py b/voluptuous_openapi/__init__.py index 52929c4..d506055 100644 --- a/voluptuous_openapi/__init__.py +++ b/voluptuous_openapi/__init__.py @@ -1,7 +1,7 @@ """Module to convert voluptuous schemas to dictionaries.""" from collections.abc import Callable, Mapping, Sequence -from inspect import signature +from inspect import isroutine, signature from enum import Enum, StrEnum import itertools import re @@ -47,6 +47,7 @@ def convert( *, custom_serializer: Callable | None = None, openapi_version: OpenApiVersion = OpenApiVersion.V3, + _seen_refs: set[str] | None = None, ) -> dict: """Convert a voluptuous schema to a OpenAPI Schema object.""" # pylint: disable=too-many-return-statements,too-many-branches @@ -54,9 +55,28 @@ def convert( def convert_with_args(schema: Any) -> dict: """Convert schema for recusing and propagating arguments.""" return convert( - schema, custom_serializer=custom_serializer, openapi_version=openapi_version + schema, + custom_serializer=custom_serializer, + openapi_version=openapi_version, + _seen_refs=_seen_refs, ) + if isinstance(schema, LazySchema): + # Recursively resolve and denormalize references. We track visited references + # in _seen_refs to detect cycle/recursion loops. If a cycle is detected, we + # break it by returning {} (representing Any). + if _seen_refs is None: + _seen_refs = set() + if schema.ref in _seen_refs: + return {} + _seen_refs.add(schema.ref) + try: + target_schema = resolve_ref(schema.ref, schema.root_schema) + vol_target = convert_to_voluptuous(target_schema, schema.root_schema) + return convert_with_args(vol_target) + finally: + _seen_refs.remove(schema.ref) + def ensure_default(value: dict[str:Any]): """Make sure that type is set.""" if all(x not in value for x in ("type", "anyOf", "oneOf", "allOf", "not")): @@ -433,9 +453,16 @@ def ensure_default(value: dict[str:Any]): return {"type": "object", "additionalProperties": True} if callable(schema): - schema = get_type_hints(schema).get( - list(signature(schema).parameters.keys())[0], Any - ) + if isinstance(schema, type) or isroutine(schema): + hints = get_type_hints(schema) + else: + # For custom callable class instances (such as LazySchema), we inspect + # the __call__ method instead. We cannot fully serialize arbitrary Python + # callables back to OpenAPI, but we can try to extract the type annotation + # of their first parameter to preserve type information in the generated schema. + hints = get_type_hints(schema.__call__) + params = list(signature(schema).parameters.keys()) + schema = hints.get(params[0], Any) if params else Any if schema is Any or isinstance(schema, TypeVar): return {} if isinstance(schema, UnionType) or get_origin(schema) is Union: @@ -499,6 +526,14 @@ class LazySchema: By returning a LazySchema instance during compilation, we avoid infinite recursion/loops for circular or self-referential schemas. The schema is resolved and compiled only on its first invocation, and cached for future runs. + + Serialization Round-Trip Caveats: + When converting a voluptuous schema containing a LazySchema back to an OpenAPI + schema: + 1. Non-recursive references are fully denormalized (resolved and expanded). + 2. Recursive/circular references are broken at the recursion boundary and return + an empty schema {} (representing Any, which is defaulted to a string type by + the library's ensure_default helper) to prevent infinite recursion/loops. """ def __init__(self, ref: str, root_schema: dict): @@ -525,6 +560,9 @@ def __eq__(self, other: Any) -> bool: return False return self.ref == other.ref and self.root_schema == other.root_schema + def __hash__(self) -> int: + return hash(self.ref) + def __repr__(self) -> str: return f"LazySchema({self.ref!r})"