Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 130 additions & 0 deletions tests/test_lib.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"}
48 changes: 43 additions & 5 deletions voluptuous_openapi/__init__.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -47,16 +47,36 @@ 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

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")):
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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):
Expand All @@ -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})"

Expand Down