From e5221bee495c78cc0120fe4d7a45c31028627108 Mon Sep 17 00:00:00 2001 From: niccolum Date: Sat, 20 Jun 2026 10:49:32 +0300 Subject: [PATCH 1/6] test_json_schema --- .gitignore | 1 + src/adaptix/_internal/morphing/facade/func.py | 11 +- .../_internal/morphing/json_schema/patch.py | 2 +- .../morphing/json_schema/schema_tools.py | 4 +- .../morphing/test_json_schema_facade.py | 276 ++++++++++++++++++ .../morphing/test_json_schema_overrides.py | 242 +++++++++++++++ .../tests_helpers/structure_tools.py | 104 +++++++ tests/unit/morphing/json_schema/__init__.py | 0 .../morphing/json_schema/test_mangling.py | 141 +++++++++ tests/unit/morphing/json_schema/test_patch.py | 192 ++++++++++++ .../json_schema/test_ref_generator.py | 71 +++++ .../morphing/json_schema/test_resolver.py | 179 ++++++++++++ .../morphing/json_schema/test_schema_tools.py | 221 ++++++++++++++ 13 files changed, 1440 insertions(+), 4 deletions(-) create mode 100644 tests/integration/morphing/test_json_schema_facade.py create mode 100644 tests/integration/morphing/test_json_schema_overrides.py create mode 100644 tests/tests_helpers/tests_helpers/structure_tools.py create mode 100644 tests/unit/morphing/json_schema/__init__.py create mode 100644 tests/unit/morphing/json_schema/test_mangling.py create mode 100644 tests/unit/morphing/json_schema/test_patch.py create mode 100644 tests/unit/morphing/json_schema/test_ref_generator.py create mode 100644 tests/unit/morphing/json_schema/test_resolver.py create mode 100644 tests/unit/morphing/json_schema/test_schema_tools.py diff --git a/.gitignore b/.gitignore index 1a935416..127bf229 100644 --- a/.gitignore +++ b/.gitignore @@ -54,3 +54,4 @@ dmypy.json /docs/build .venv/ +uv.lock \ No newline at end of file diff --git a/src/adaptix/_internal/morphing/facade/func.py b/src/adaptix/_internal/morphing/facade/func.py index 63913e9d..2863399b 100644 --- a/src/adaptix/_internal/morphing/facade/func.py +++ b/src/adaptix/_internal/morphing/facade/func.py @@ -1,4 +1,5 @@ from collections.abc import Container, Iterable, Mapping, Sequence +from contextlib import suppress from dataclasses import fields from typing import Any, TypeVar, overload @@ -12,7 +13,7 @@ from ..json_schema.ref_generator import BuiltinRefGenerator from ..json_schema.request_cls import JSONSchemaContext from ..json_schema.resolver import BuiltinJSONSchemaResolver, JSONSchemaResolver -from ..json_schema.schema_model import JSONObject, _JSONSchemaCore +from ..json_schema.schema_model import JSONObject, JSONSchemaBuiltinFormat, _JSONSchemaCore from ..load_error import TypeLoadError from ..provider_template import ABCProxy from .provider import loader, name_mapping @@ -56,6 +57,13 @@ def _ref_loader(data): raise TypeLoadError(expected_type=str, input_value=data) +def _format_loader(data): + if isinstance(data, str): + with suppress(ValueError): + return JSONSchemaBuiltinFormat(data) + return data + + _global_resolver = BuiltinJSONSchemaResolver( ref_generator=BuiltinRefGenerator(), ref_mangler=CompoundRefMangler(QualnameRefMangler(), IndexRefMangler()), @@ -74,6 +82,7 @@ def _ref_loader(data): extra_out="extra_keywords", ), loader(P[JSONSchema].ref, _ref_loader), + loader(P[JSONSchema].format, _format_loader), ], ) diff --git a/src/adaptix/_internal/morphing/json_schema/patch.py b/src/adaptix/_internal/morphing/json_schema/patch.py index cb9783d1..98a0a2ae 100644 --- a/src/adaptix/_internal/morphing/json_schema/patch.py +++ b/src/adaptix/_internal/morphing/json_schema/patch.py @@ -16,7 +16,7 @@ def __init__(self) -> None: self._patchers: list[Patcher] = [] def _append_with_patcher(self: S, patcher: Callable[[JSONSchema], JSONSchema]) -> S: - self_copy = copy(self) + self_copy = deepcopy(self) self_copy._patchers.append(patcher) return self_copy diff --git a/src/adaptix/_internal/morphing/json_schema/schema_tools.py b/src/adaptix/_internal/morphing/json_schema/schema_tools.py index ae7fdb43..ab84572a 100644 --- a/src/adaptix/_internal/morphing/json_schema/schema_tools.py +++ b/src/adaptix/_internal/morphing/json_schema/schema_tools.py @@ -52,7 +52,7 @@ **_base_json_schema_templates, Omittable[RefT]: dedent( # type: ignore[misc, valid-type] """ - if __value__ != Omitted(): + if __value__ != Omitted() and isinstance(__value__, LocalRefSource): yield from __traverser__(__value__.json_schema) """, ), @@ -88,7 +88,7 @@ def {function_name}(obj, /): """, ) + "\n\n".join(indent(item, " " * 4) for item in result) - namespace: dict[str, Any] = {"Omitted": Omitted} + namespace: dict[str, Any] = {"Omitted": Omitted, "LocalRefSource": LocalRefSource} exec(compile(module_code, file_name, "exec"), namespace, namespace) # noqa: S102 return namespace[function_name] diff --git a/tests/integration/morphing/test_json_schema_facade.py b/tests/integration/morphing/test_json_schema_facade.py new file mode 100644 index 00000000..f7c65bc1 --- /dev/null +++ b/tests/integration/morphing/test_json_schema_facade.py @@ -0,0 +1,276 @@ +from dataclasses import dataclass +from typing import Any, Optional + +import pytest +from tests_helpers.structure_tools import EXISTS, NOT_EXISTS, assert_structure + +from adaptix import Retort +from adaptix._internal.definitions import Direction +from adaptix._internal.morphing.facade.func import ( + generate_json_schema, + generate_json_schemas_namespace, + load_json_schema, +) +from adaptix._internal.morphing.json_schema.definitions import JSONSchema +from adaptix._internal.morphing.json_schema.mangling import IndexRefMangler +from adaptix._internal.morphing.json_schema.ref_generator import BuiltinRefGenerator +from adaptix._internal.morphing.json_schema.resolver import BuiltinJSONSchemaResolver +from adaptix._internal.morphing.json_schema.schema_model import ( + JSONSchemaBuiltinFormat, + JSONSchemaType, +) +from adaptix.load_error import AggregateLoadError + +DIALECT_URI = "https://json-schema.org/draft/2020-12/schema" + + +@pytest.mark.parametrize( + ["raw", "expected"], + [ + ("null", JSONSchemaType.NULL), + ("boolean", JSONSchemaType.BOOLEAN), + ("object", JSONSchemaType.OBJECT), + ("array", JSONSchemaType.ARRAY), + ("number", JSONSchemaType.NUMBER), + ("integer", JSONSchemaType.INTEGER), + ("string", JSONSchemaType.STRING), + ], +) +def test_load_json_schema_type_enum_values(raw: str, expected: JSONSchemaType): + schema = load_json_schema({"type": raw}) + + assert schema.type == expected + + +@pytest.mark.parametrize( + ["schema_data", "expected"], + [ + ( + {"type": "string", "title": "MyStr", "description": "A string"}, + JSONSchema(type=JSONSchemaType.STRING, title="MyStr", description="A string"), + ), + ( + {"type": "string", "format": "date-time"}, + JSONSchema(type=JSONSchemaType.STRING, format=JSONSchemaBuiltinFormat.DATE_TIME), + ), + ( + {"type": "string", "format": "my-custom-format"}, + JSONSchema(type=JSONSchemaType.STRING, format="my-custom-format"), + ), + ( + { + "type": "object", + "properties": { + "name": {"type": "string"}, + "age": {"type": "integer"}, + }, + "required": ["name"], + }, + JSONSchema( + type=JSONSchemaType.OBJECT, + properties={ + "name": JSONSchema(type=JSONSchemaType.STRING), + "age": JSONSchema(type=JSONSchemaType.INTEGER), + }, + required=["name"], + ), + ), + ( + {"anyOf": [{"type": "string"}, {"type": "integer"}]}, + JSONSchema(any_of=[ + JSONSchema(type=JSONSchemaType.STRING), + JSONSchema(type=JSONSchemaType.INTEGER), + ]), + ), + ], +) +def test_load_json_schema_parses_dict_to_object(schema_data, expected): + schema = load_json_schema(schema_data) + + assert schema == expected + + +def test_load_json_schema_strict_unknown_field_raises(): + with pytest.raises(AggregateLoadError): + load_json_schema({"type": "string", "x-custom": "value"}, error_on_extra=True) + + +def test_load_json_schema_lax_unknown_field_goes_to_extra_keywords(): + schema = load_json_schema({"type": "string", "x-custom": "value"}, error_on_extra=False) + + assert schema.extra_keywords == {"x-custom": "value"} + + +def test_load_json_schema_empty_dict(): + schema = load_json_schema({}) + + assert schema == JSONSchema() + + +@dataclass +class SimpleModel: + name: str + value: int + + +@dataclass +class ModelWithOptional: + required_field: str + optional_field: Optional[str] = None + + +@pytest.mark.parametrize( + ["with_dialect_uri", "expected_schema"], + [ + (True, DIALECT_URI), + (False, NOT_EXISTS), + ], + ids=["with_uri", "without_uri"], +) +def test_generate_json_schema_dialect_uri(with_dialect_uri: bool, expected_schema: Any): # noqa: FBT001 + schema = generate_json_schema(Retort(), SimpleModel, Direction.INPUT, with_dialect_uri=with_dialect_uri) + + assert_structure(schema, {"$schema": expected_schema}) + + +def test_generate_json_schema_top_level_keys(): + schema = generate_json_schema(Retort(), SimpleModel, Direction.INPUT) + + assert_structure(schema, { + "$schema": EXISTS, + "$ref": EXISTS, + "$defs": { + "SimpleModel": EXISTS, + }, + }, strict=True) + + +def test_generate_json_schema_custom_ref_prefix(): + schema = generate_json_schema( + Retort(), SimpleModel, Direction.INPUT, + local_ref_prefix="#/components/schemas/", + ) + + assert_structure(schema, {"$ref": "#/components/schemas/SimpleModel"}) + + +def test_generate_json_schema_occupied_refs_triggers_mangling(): + schema = generate_json_schema( + Retort(), SimpleModel, Direction.INPUT, + occupied_refs=("SimpleModel",), + resolver=BuiltinJSONSchemaResolver(BuiltinRefGenerator(), IndexRefMangler()), + ) + + assert_structure(schema, { + "$defs": { + "SimpleModel": NOT_EXISTS, + "SimpleModel-1": EXISTS, + }, + }) + + +def test_generate_json_schema_custom_resolver(): + schema = generate_json_schema( + Retort(), SimpleModel, Direction.INPUT, + resolver=BuiltinJSONSchemaResolver(BuiltinRefGenerator(), IndexRefMangler()), + ) + + assert_structure(schema, {"$defs": {"SimpleModel": EXISTS}}) + + +def test_generate_json_schema_optional_field_not_required_on_input(): + input_schema = generate_json_schema(Retort(), ModelWithOptional, Direction.INPUT) + output_schema = generate_json_schema(Retort(), ModelWithOptional, Direction.OUTPUT) + + # Input schema: only required_field is required (optional_field has default) + assert_structure(input_schema, { + "$defs": { + "ModelWithOptional": { + "required": ["required_field"], + }, + }, + }) + + # Output schema: both fields are required (dumping doesn't care about defaults) + assert_structure(output_schema, { + "$defs": { + "ModelWithOptional": { + "required": ["required_field", "optional_field"], + }, + }, + }) + + +@dataclass +class Tag: + label: str + + +@dataclass +class Article: + title: str + tag: Tag + + +@dataclass +class Post: + body: str + tag: Tag + + +def _namespace_query(): + retort = Retort() + return [(retort, Direction.INPUT, Article), (retort, Direction.INPUT, Post)] + + +@pytest.mark.parametrize( + "resolver", + [ + pytest.param(None, id="default"), + pytest.param( + BuiltinJSONSchemaResolver(BuiltinRefGenerator(), IndexRefMangler()), + id="custom_resolver", + ), + ], +) +def test_generate_schemas_namespace_deduplicates_shared_types(resolver): + query = _namespace_query() + kwargs = {"resolver": resolver} if resolver else {} + + defs, schemas = generate_json_schemas_namespace(query, **kwargs) + + assert len(schemas) == 2 + + assert_structure(defs, { + "Article": EXISTS, + "Post": EXISTS, + "Tag": EXISTS, + }, strict=True) + + +@pytest.mark.parametrize( + "with_dialect_uri", + [ + pytest.param(True, id="with_uri"), + pytest.param(False, id="without_uri"), + ], +) +def test_generate_schemas_namespace_dialect_uri(with_dialect_uri: bool): # noqa: FBT001 + query = _namespace_query() + _, schemas = generate_json_schemas_namespace(query, with_dialect_uri=with_dialect_uri) + + expected_schema = DIALECT_URI if with_dialect_uri else NOT_EXISTS + for schema in schemas: + assert_structure(schema, {"$schema": expected_schema, "$ref": EXISTS}) + + +def test_generate_schemas_namespace_custom_ref_prefix(): + query = _namespace_query() + _, schemas = generate_json_schemas_namespace( + query, local_ref_prefix="#/components/schemas/", + ) + + assert_structure(schemas, [ + {"$ref": "#/components/schemas/Article"}, + {"$ref": "#/components/schemas/Post"}, + ]) diff --git a/tests/integration/morphing/test_json_schema_overrides.py b/tests/integration/morphing/test_json_schema_overrides.py new file mode 100644 index 00000000..a222d4e1 --- /dev/null +++ b/tests/integration/morphing/test_json_schema_overrides.py @@ -0,0 +1,242 @@ +from dataclasses import dataclass + +import pytest +from tests_helpers.morphing import JSONSchemaFork, assert_morphing +from tests_helpers.structure_tools import EXISTS, NOT_EXISTS, assert_structure + +from adaptix import Chain, P, ProviderNotFoundError, Retort, dumper, json_schema, loader +from adaptix._internal.definitions import Direction +from adaptix._internal.morphing.facade.func import generate_json_schema +from adaptix._internal.morphing.json_schema.definitions import JSONSchema +from adaptix._internal.morphing.json_schema.patch import JSONSchemaPatch +from adaptix._internal.morphing.json_schema.providers import EraseJSONSchema, KeepJSONSchema +from adaptix._internal.morphing.json_schema.schema_model import JSONSchemaType + + +@dataclass +class Product: + name: str + price: float + + +_DIALECT_URI = "https://json-schema.org/draft/2020-12/schema" +_PRODUCT_DATA = {"name": "test", "price": 1.5} +_PRODUCT_LOADED = Product(name="test", price=1.5) + + +def _product_schema(name_prop: dict, price_prop: dict) -> JSONSchemaFork: + def _make(product_extra: dict = {}) -> dict: # noqa: B006 + return { + "$ref": "#/$defs/Product", + "$schema": _DIALECT_URI, + "$defs": { + "Product": { + "title": "Product", + "type": "object", + "required": ["name", "price"], + "properties": { + "name": name_prop, + "price": price_prop, + }, + **product_extra, + }, + }, + } + + return JSONSchemaFork( + input=_make({"additionalProperties": True}), + output=_make(), + ) + + +_DEFAULT_SCHEMA = _product_schema({"type": "string"}, {"type": "number"}) + + +def test_json_schema_explicit_replaces_inferred(): + retort = Retort(recipe=[ + json_schema(P[Product].name, JSONSchema(type=JSONSchemaType.INTEGER)), + ]) + schema = generate_json_schema(retort, Product, Direction.INPUT) + + assert_structure(schema, { + "$defs": {"Product": {"properties": {"name": {"type": "integer"}}}}, + }) + + +def test_json_schema_explicit_empty_schema(): + retort = Retort(recipe=[ + json_schema(str, JSONSchema()), + ]) + + assert_morphing( + retort=retort, + tp=Product, + data=_PRODUCT_DATA, + loaded=_PRODUCT_LOADED, + json_schema=_product_schema({}, {"type": "number"}), + ) + + +def test_json_schema_explicit_with_title_and_description(): + retort = Retort(recipe=[ + json_schema(P[Product].price, JSONSchema(title="Price", description="Product price")), + ]) + + assert_morphing( + retort=retort, + tp=Product, + data=_PRODUCT_DATA, + loaded=_PRODUCT_LOADED, + json_schema=_product_schema( + {"type": "string"}, + {"title": "Price", "description": "Product price"}, + ), + ) + + +def test_keep_json_schema_preserves_inferred_schema(): + retort = Retort(recipe=[json_schema(str, KeepJSONSchema())]) + + assert_morphing( + retort=retort, + tp=Product, + data=_PRODUCT_DATA, + loaded=_PRODUCT_LOADED, + json_schema=_DEFAULT_SCHEMA, + ) + + +def test_erase_json_schema_on_loader_raises_error(): + retort = Retort(recipe=[ + loader(str, lambda x: x, json_schema=EraseJSONSchema()), + ]) + + with pytest.raises(ProviderNotFoundError): + generate_json_schema(retort, Product, Direction.INPUT) + + +def test_erase_json_schema_on_dumper_raises_error(): + retort = Retort(recipe=[ + dumper(str, lambda x: x, json_schema=EraseJSONSchema()), + ]) + + with pytest.raises(ProviderNotFoundError): + generate_json_schema(retort, Product, Direction.OUTPUT) + + +def test_patch_merge_with_adds_description(): + retort = Retort(recipe=[ + json_schema( + P[Product].name, + JSONSchemaPatch().merge_with(JSONSchema(description="The product name"), Chain.LAST), + ), + ]) + + assert_morphing( + retort=retort, + tp=Product, + data=_PRODUCT_DATA, + loaded=_PRODUCT_LOADED, + json_schema=_product_schema( + {"description": "The product name", "type": "string"}, + {"type": "number"}, + ), + ) + + +def test_patch_merge_with_chain_first_overrides_inferred(): + retort = Retort(recipe=[ + json_schema( + P[Product].price, + JSONSchemaPatch().merge_with(JSONSchema(type=JSONSchemaType.STRING), Chain.FIRST), + ), + ]) + schema = generate_json_schema(retort, Product, Direction.INPUT) + + assert_structure(schema, { + "$defs": {"Product": {"properties": {"price": {"type": "string"}}}}, + }) + + +def test_patch_replace_title(): + retort = Retort(recipe=[ + json_schema( + P[Product].name, + JSONSchemaPatch().replace("title", lambda _: "CustomTitle"), + ), + ]) + + assert_morphing( + retort=retort, + tp=Product, + data=_PRODUCT_DATA, + loaded=_PRODUCT_LOADED, + json_schema=_product_schema( + {"title": "CustomTitle", "type": "string"}, + {"type": "number"}, + ), + ) + + +@pytest.mark.parametrize( + ["inline", "expected_ref"], + [ + pytest.param(True, NOT_EXISTS, id="inline_true_embeds_directly"), + pytest.param(False, EXISTS, id="inline_false_uses_ref"), + ], +) +def test_json_schema_inline_controls_ref_embedding(inline, expected_ref): + retort = Retort(recipe=[json_schema(str, inline=inline)]) + schema = generate_json_schema(retort, Product, Direction.INPUT) + + assert_structure( + schema["$defs"]["Product"]["properties"]["name"], + {"$ref": expected_ref}, + ) + + +def test_json_schema_pinned_ref(): + retort = Retort(recipe=[json_schema(str, ref="MyString", inline=False)]) + schema = generate_json_schema(retort, Product, Direction.INPUT) + + assert_structure(schema, { + "$defs": { + "MyString": EXISTS, + "Product": {"properties": {"name": {"$ref": "#/$defs/MyString"}}}, + }, + }) + + +def test_loader_custom_json_schema_kwarg_reflected_in_schema(): + retort = Retort(recipe=[ + loader(str, str, json_schema=JSONSchema(type=JSONSchemaType.STRING, title="CustomStr")), + ]) + + assert_morphing( + retort=retort, + tp=Product, + data=_PRODUCT_DATA, + loaded=_PRODUCT_LOADED, + json_schema=_product_schema( + {"title": "CustomStr", "type": "string"}, + {"type": "number"}, + ), + ) + + +def test_multiple_overrides_on_different_fields(): + retort = Retort(recipe=[ + json_schema(P[Product].name, JSONSchema(title="Name")), + json_schema(P[Product].price, JSONSchema(title="Price")), + ]) + + assert_morphing( + retort=retort, + tp=Product, + data=_PRODUCT_DATA, + loaded=_PRODUCT_LOADED, + json_schema=_product_schema( + {"title": "Name"}, + {"title": "Price"}, + ), + ) diff --git a/tests/tests_helpers/tests_helpers/structure_tools.py b/tests/tests_helpers/tests_helpers/structure_tools.py new file mode 100644 index 00000000..f1911b78 --- /dev/null +++ b/tests/tests_helpers/tests_helpers/structure_tools.py @@ -0,0 +1,104 @@ +from dataclasses import dataclass +from typing import Any, Mapping, Union + + +@dataclass(slots=True) +class Exists: + """Marker to check that a key exists in the dict (value type not checked).""" + + +@dataclass(slots=True) +class NotExists: + """Marker to check that a key does NOT exist in the dict.""" + + +EXISTS = Exists() +NOT_EXISTS = NotExists() + + +def assert_structure( + actual: Union[Mapping[str, Any], list], + expected: Union[Mapping[str, Any], list], + *, + path: str = "", + strict: bool = False, +) -> None: + """ + Assert that a dict or list matches the expected structure. + + Supports nested dicts and lists. + For dict values, supports: + - EXISTS: check that key exists (any value) + - NOT_EXISTS: check that key does NOT exist + - Other values: direct equality check or recursive structure check + + Args: + actual: The actual dict or list to check + expected: Dict or list describing expected structure + path: Current path for error messages (used internally for recursion) + strict: If True, fail if actual has keys/elements not in expected + + Example: + assert_structure(schema, { + "$ref": "#/$defs/SimpleModel", + "$defs": { + "SimpleModel": {}, + }, + }) + assert_structure([{"a": 1}, {"b": 2}], [{"a": 1}, {"b": EXISTS}]) + """ + if isinstance(expected, list): + assert isinstance(actual, list), ( + f"At {path!r}: expected list, got {type(actual).__name__}" + ) + if strict: + assert len(actual) == len(expected), ( + f"At {path!r}: expected list of length {len(expected)}, got {len(actual)}" + ) + else: + assert len(actual) >= len(expected), ( + f"At {path!r}: expected at least {len(expected)} elements, got {len(actual)}" + ) + for i, expected_item in enumerate(expected): + current_path = f"{path}[{i}]" + actual_item = actual[i] + if isinstance(expected_item, (dict, list)): + assert_structure(actual_item, expected_item, path=current_path, strict=strict) + else: + assert actual_item == expected_item, ( + f"At {current_path}: expected {expected_item!r}, got {actual_item!r}" + ) + return + + # Check for extra keys in strict mode + if strict: + expected_keys = set(expected.keys()) + actual_keys = set(actual.keys()) + extra_keys = actual_keys - expected_keys + if extra_keys: + raise AssertionError( + f"At {path!r}: unexpected keys: {sorted(extra_keys)}" + ) + + for key, expected_value in expected.items(): + current_path = f"{path}.{key}" if path else key + + if expected_value is NOT_EXISTS: + assert key not in actual, ( + f"Key {current_path!r} should not exist, but found: {actual[key]!r}" + ) + continue + + assert key in actual, f"Key {current_path!r} not found in dict" + + if expected_value is EXISTS: + continue + + actual_value = actual[key] + + if isinstance(expected_value, (dict, list)): + assert_structure(actual_value, expected_value, path=current_path, strict=strict) + else: + assert actual_value == expected_value, ( + f"At {current_path!r}: expected {expected_value!r}, got {actual_value!r}" + ) diff --git a/tests/unit/morphing/json_schema/__init__.py b/tests/unit/morphing/json_schema/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unit/morphing/json_schema/test_mangling.py b/tests/unit/morphing/json_schema/test_mangling.py new file mode 100644 index 00000000..58d06899 --- /dev/null +++ b/tests/unit/morphing/json_schema/test_mangling.py @@ -0,0 +1,141 @@ +import pytest + +from adaptix._internal.morphing.json_schema.definitions import JSONSchema, LocalRefSource +from adaptix._internal.morphing.json_schema.mangling import CompoundRefMangler, IndexRefMangler, QualnameRefMangler +from adaptix._internal.morphing.json_schema.resolver import LocalRefSourceGroup +from adaptix._internal.provider.loc_stack_filtering import LocStack +from adaptix._internal.provider.location import TypeHintLoc + + +def make_group(*types, pinned=None): + sources = [ + LocalRefSource(value=pinned, json_schema=JSONSchema(), loc_stack=LocStack(TypeHintLoc(tp))) + for tp in types + ] + return LocalRefSourceGroup(sources) + + +class _Outer: + class _Inner: + pass + + +@pytest.mark.parametrize( + ["kwargs", "expected_g1", "expected_g2"], + [ + ({}, "Foo-1", "Foo-2"), + ({"start": 0}, "Foo-0", "Foo-1"), + ({"separator": "."}, "Foo.1", "Foo.2"), + ], +) +def test_index_mangler_two_groups(kwargs, expected_g1, expected_g2): + g1, g2 = make_group(int), make_group(str) + + result = IndexRefMangler(**kwargs).mangle_refs((), "Foo", [g1, g2]) + + assert result == {g1: expected_g1, g2: expected_g2} + + +def test_index_mangler_three_groups(): + g1, g2, g3 = make_group(int), make_group(str), make_group(float) + + result = IndexRefMangler().mangle_refs((), "T", [g1, g2, g3]) + + assert result == {g1: "T-1", g2: "T-2", g3: "T-3"} + + +def test_index_mangler_skips_occupied_ref(): + g1, g2 = make_group(int), make_group(str) + + result = IndexRefMangler().mangle_refs(("Foo-1",), "Foo", [g1, g2]) + + assert result == {g1: "Foo-2", g2: "Foo-3"} + + +def test_index_mangler_skips_multiple_occupied_refs(): + g1 = make_group(int) + + result = IndexRefMangler().mangle_refs(("Foo-1", "Foo-2", "Foo-3"), "Foo", [g1]) + + assert result == {g1: "Foo-4"} + + +def test_qualname_mangler_builtin_types(): + g_int, g_str = make_group(int), make_group(str) + + result = QualnameRefMangler().mangle_refs((), "common", [g_int, g_str]) + + assert result == {g_int: "int", g_str: "str"} + + +def test_qualname_mangler_nested_class(): + g = make_group(_Outer._Inner) + + result = QualnameRefMangler().mangle_refs((), "common", [g]) + + assert result == {g: "_Outer._Inner"} + + +def test_qualname_mangler_mixed_types_in_group_falls_back_to_common_ref(): + g_mixed = LocalRefSourceGroup([ + LocalRefSource(value=None, json_schema=JSONSchema(), loc_stack=LocStack(TypeHintLoc(int))), + LocalRefSource(value=None, json_schema=JSONSchema(), loc_stack=LocStack(TypeHintLoc(str))), + ]) + + result = QualnameRefMangler().mangle_refs((), "FallbackName", [g_mixed]) + + assert result == {g_mixed: "FallbackName"} + + +def test_qualname_mangler_same_type_multiple_sources_uses_qualname(): + g = LocalRefSourceGroup([ + LocalRefSource(value=None, json_schema=JSONSchema(), loc_stack=LocStack(TypeHintLoc(int))), + LocalRefSource(value=None, json_schema=JSONSchema(), loc_stack=LocStack(TypeHintLoc(int))), + ]) + + result = QualnameRefMangler().mangle_refs((), "common", [g]) + + assert result == {g: "int"} + + +def test_compound_mangler_no_conflict_uses_base_result(): + g_int, g_str = make_group(int), make_group(str) + cm = CompoundRefMangler(QualnameRefMangler(), IndexRefMangler()) + + result = cm.mangle_refs((), "common", [g_int, g_str]) + + assert result == {g_int: "int", g_str: "str"} + + +def test_compound_mangler_conflict_appends_index_suffix(): + g1, g2 = make_group(int), make_group(int) + cm = CompoundRefMangler(QualnameRefMangler(), IndexRefMangler()) + + result = cm.mangle_refs((), "common", [g1, g2]) + + assert result == {g1: "int-1", g2: "int-2"} + + +def test_compound_mangler_partial_conflict_leaves_unique_refs_unchanged(): + g_int1, g_int2, g_str = make_group(int), make_group(int), make_group(str) + cm = CompoundRefMangler(QualnameRefMangler(), IndexRefMangler()) + + result = cm.mangle_refs((), "common", [g_int1, g_int2, g_str]) + + assert result == {g_int1: "int-1", g_int2: "int-2", g_str: "str"} + + +@pytest.mark.parametrize( + ["start", "separator", "expected_g1", "expected_g2"], + [ + (1, "-", "int-1", "int-2"), + (0, "_", "int_0", "int_1"), + ], +) +def test_compound_mangler_wrapper_params_propagate(start, separator, expected_g1, expected_g2): + g1, g2 = make_group(int), make_group(int) + cm = CompoundRefMangler(QualnameRefMangler(), IndexRefMangler(start=start, separator=separator)) + + result = cm.mangle_refs((), "common", [g1, g2]) + + assert result == {g1: expected_g1, g2: expected_g2} diff --git a/tests/unit/morphing/json_schema/test_patch.py b/tests/unit/morphing/json_schema/test_patch.py new file mode 100644 index 00000000..dc18c8b6 --- /dev/null +++ b/tests/unit/morphing/json_schema/test_patch.py @@ -0,0 +1,192 @@ +import pytest + +from adaptix import Chain +from adaptix._internal.morphing.json_schema.definitions import JSONSchema +from adaptix._internal.morphing.json_schema.patch import JSONSchemaPatch + + +def apply_patch(patch: JSONSchemaPatch, schema: JSONSchema) -> JSONSchema: + for patcher in patch.get_patchers(): + schema = patcher(schema) + return schema + + +def test_replace_sets_field(): + schema = JSONSchema(title="old") + + result = apply_patch(JSONSchemaPatch().replace("title", lambda _: "new"), schema) + + assert result.title == "new" + + +def test_replace_receives_old_value(): + schema = JSONSchema(title="base") + received = [] + + apply_patch(JSONSchemaPatch().replace("title", lambda v: received.append(v) or v), schema) + + assert received == ["base"] + + +def test_replace_omitted_field(): + schema = JSONSchema() + + result = apply_patch(JSONSchemaPatch().replace("description", lambda _: "added"), schema) + + assert result.description == "added" + + +def test_mutate_copy_does_not_modify_original(): + schema = JSONSchema(required=["a", "b"]) + + apply_patch( + JSONSchemaPatch().mutate_copy("required", lambda lst: lst.append("c")), + schema, + ) + + assert schema.required == ["a", "b"] + + +def test_mutate_copy_new_schema_has_mutation(): + schema = JSONSchema(required=["a"]) + + result = apply_patch( + JSONSchemaPatch().mutate_copy("required", lambda lst: lst.append("b")), + schema, + ) + + assert result.required == ["a", "b"] + + +def test_mutate_copy_shallow_does_not_deep_copy(): + inner = {"x": [1, 2]} + schema = JSONSchema(extra_keywords={"nested": inner}) + + result = apply_patch( + JSONSchemaPatch().mutate_copy("extra_keywords", lambda d: d.update({"new_key": "val"})), + schema, + ) + + assert result.extra_keywords == {"nested": inner, "new_key": "val"} + + +def test_mutate_deepcopy_does_not_modify_original(): + inner_list = [1, 2] + schema = JSONSchema(extra_keywords={"items": inner_list}) + + apply_patch( + JSONSchemaPatch().mutate_deepcopy("extra_keywords", lambda d: d["items"].append(99)), + schema, + ) + + assert inner_list == [1, 2] + + +def test_mutate_deepcopy_inner_is_separate_object(): + inner_list = [1, 2] + schema = JSONSchema(extra_keywords={"items": inner_list}) + + result = apply_patch( + JSONSchemaPatch().mutate_deepcopy("extra_keywords", lambda _: None), + schema, + ) + + assert result.extra_keywords["items"] == [1, 2] + assert result.extra_keywords["items"] is not inner_list + + +def test_merge_with_chain_first_override_wins(): + base = JSONSchema(title="base", description="base_desc") + override = JSONSchema(description="new_desc") + + result = apply_patch(JSONSchemaPatch().merge_with(override, Chain.FIRST), base) + + assert result.title == "base" + assert result.description == "new_desc" + + +def test_merge_with_chain_last_base_wins(): + base = JSONSchema(title="base", description="base_desc") + override = JSONSchema(description="new_desc") + + result = apply_patch(JSONSchemaPatch().merge_with(override, Chain.LAST), base) + + assert result.title == "base" + assert result.description == "base_desc" + + +def test_merge_with_chain_first_omitted_field_does_not_overwrite(): + base = JSONSchema(title="keep_me") + override = JSONSchema(description="added") + + result = apply_patch(JSONSchemaPatch().merge_with(override, Chain.FIRST), base) + + assert result.title == "keep_me" + assert result.description == "added" + + +def test_merge_with_default_chain_is_first(): + base = JSONSchema(title="base") + override = JSONSchema(title="override") + + result = apply_patch(JSONSchemaPatch().merge_with(override), base) + + assert result.title == "override" + + +def test_merge_with_empty_override_leaves_base_unchanged(): + base = JSONSchema(title="t", description="d") + + result = apply_patch(JSONSchemaPatch().merge_with(JSONSchema(), Chain.FIRST), base) + + assert result.title == "t" + assert result.description == "d" + + +def test_multiple_patchers_applied_in_order(): + schema = JSONSchema(title="a") + patch = ( + JSONSchemaPatch() + .replace("title", lambda t: t + "b") + .replace("title", lambda t: t + "c") + ) + + result = apply_patch(patch, schema) + + assert result.title == "abc" + + +def test_builder_is_immutable(): + original = JSONSchemaPatch() + + extended = original.replace("title", lambda _: "x") + + assert list(original.get_patchers()) == [] + assert len(list(extended.get_patchers())) == 1 + + +def test_builder_each_method_returns_new_instance(): + p = JSONSchemaPatch() + instances = [ + p, + p.replace("title", lambda _: "x"), + p.mutate_copy("required", lambda lst: lst), + p.merge_with(JSONSchema()), + p.mutate_deepcopy("extra_keywords", lambda d: d), + ] + assert len(set(instances)) == len(instances) + + +def test_get_patchers_empty_by_default(): + assert list(JSONSchemaPatch().get_patchers()) == [] + + +def test_get_patchers_returns_all_patchers(): + patch = ( + JSONSchemaPatch() + .replace("title", lambda _: "x") + .replace("description", lambda _: "y") + .merge_with(JSONSchema()) + ) + + assert len(list(patch.get_patchers())) == 3 diff --git a/tests/unit/morphing/json_schema/test_ref_generator.py b/tests/unit/morphing/json_schema/test_ref_generator.py new file mode 100644 index 00000000..ac1ac525 --- /dev/null +++ b/tests/unit/morphing/json_schema/test_ref_generator.py @@ -0,0 +1,71 @@ +import re + +import pytest + +from adaptix._internal.morphing.json_schema.definitions import JSONSchema +from adaptix._internal.morphing.json_schema.ref_generator import BuiltinRefGenerator +from adaptix._internal.provider.loc_stack_filtering import LocStack +from adaptix._internal.provider.location import TypeHintLoc + + +generator = BuiltinRefGenerator() + + +def ref_for(tp): + return generator.generate_ref(JSONSchema(), LocStack(TypeHintLoc(tp))) + + +class _SimpleModel: + pass + + +class _OuterModel: + class _InnerModel: + pass + + +@pytest.mark.parametrize( + ["tp", "expected"], + [ + (int, "int"), + (str, "str"), + (float, "float"), + (bool, "bool"), + (bytes, "bytes"), + ], +) +def test_builtin_type_ref(tp, expected): + assert ref_for(tp) == expected + + +def test_simple_class_uses_qualname(): + assert ref_for(_SimpleModel) == "_SimpleModel" + + +def test_nested_class_uses_full_qualname(): + assert ref_for(_OuterModel._InnerModel) == "_OuterModel._InnerModel" + + +@pytest.mark.parametrize( + ["tp", "expected"], + [ + (list[int], "list_int"), + (dict[str, int], "dict_str_int"), + (list[dict[str, list[int]]], "list_dict_str_list_int"), + ], +) +def test_generic_type_ref_replaces_unsafe_chars_with_underscores(tp, expected): + assert ref_for(tp) == expected + + +URI_SAFE_PATTERN = re.compile(r"[A-Za-z0-9\-._~]+") + + +@pytest.mark.parametrize( + "tp", + [list[int], dict[str, int], list[dict[str, list[int]]]], +) +def test_generic_type_ref_contains_only_uri_safe_chars(tp): + result = ref_for(tp) + + assert URI_SAFE_PATTERN.fullmatch(result) diff --git a/tests/unit/morphing/json_schema/test_resolver.py b/tests/unit/morphing/json_schema/test_resolver.py new file mode 100644 index 00000000..e0e74ca0 --- /dev/null +++ b/tests/unit/morphing/json_schema/test_resolver.py @@ -0,0 +1,179 @@ +import pytest + +from adaptix._internal.morphing.json_schema.definitions import JSONSchema, LocalRefSource, RemoteRef +from adaptix._internal.morphing.json_schema.mangling import IndexRefMangler +from adaptix._internal.morphing.json_schema.ref_generator import BuiltinRefGenerator +from adaptix._internal.morphing.json_schema.resolver import ( + BuiltinJSONSchemaResolver, + CustomJSONSchemaHasher, + RefMangler, +) +from adaptix._internal.provider.loc_stack_filtering import LocStack +from adaptix._internal.provider.location import TypeHintLoc + + +def make_src(tp, schema=None, pinned=None): + if schema is None: + schema = JSONSchema(title=getattr(tp, "__name__", str(tp))) + return LocalRefSource(value=pinned, json_schema=schema, loc_stack=LocStack(TypeHintLoc(tp))) + + +def make_resolver(mangler=None): + return BuiltinJSONSchemaResolver( + ref_generator=BuiltinRefGenerator(), + ref_mangler=mangler or IndexRefMangler(), + ) + + +PREFIX = "#/$defs/" + + +class _TypeA: + pass + + +class _TypeB: + pass + + +def test_single_ref_produces_one_def(): + src = make_src(_TypeA) + root = JSONSchema(ref=src) + + defs, schemas = make_resolver().resolve([root], local_ref_prefix=PREFIX, occupied_refs=()) + + assert list(defs.keys()) == ["_TypeA"] + assert schemas[0].ref == f"{PREFIX}_TypeA" + + +def test_remote_ref_passes_through_unchanged(): + root = JSONSchema(ref=RemoteRef("https://example.com/schema")) + + defs, schemas = make_resolver().resolve([root], local_ref_prefix=PREFIX, occupied_refs=()) + + assert defs == {} + assert schemas[0].ref == "https://example.com/schema" + + +def test_no_refs_produces_empty_defs(): + root = JSONSchema(title="plain") + + defs, schemas = make_resolver().resolve([root], local_ref_prefix=PREFIX, occupied_refs=()) + + assert defs == {} + assert schemas[0].title == "plain" + + +def test_same_type_twice_deduplicates_to_one_def(): + shared_schema = JSONSchema(title="_TypeA") + src1 = LocalRefSource(value=None, json_schema=shared_schema, loc_stack=LocStack(TypeHintLoc(_TypeA))) + src2 = LocalRefSource(value=None, json_schema=shared_schema, loc_stack=LocStack(TypeHintLoc(_TypeA))) + root1 = JSONSchema(ref=src1) + root2 = JSONSchema(ref=src2) + + defs, schemas = make_resolver().resolve([root1, root2], local_ref_prefix=PREFIX, occupied_refs=()) + + assert len(defs) == 1 + assert schemas[0].ref == schemas[1].ref + + +def test_different_schemas_same_type_produce_mangled_defs(): + src1 = LocalRefSource(value=None, json_schema=JSONSchema(title="V1"), loc_stack=LocStack(TypeHintLoc(_TypeA))) + src2 = LocalRefSource(value=None, json_schema=JSONSchema(title="V2"), loc_stack=LocStack(TypeHintLoc(_TypeA))) + root1 = JSONSchema(ref=src1) + root2 = JSONSchema(ref=src2) + + defs, schemas = make_resolver().resolve([root1, root2], local_ref_prefix=PREFIX, occupied_refs=()) + + assert len(defs) == 2 + assert schemas[0].ref != schemas[1].ref + + +def test_pinned_ref_used_as_def_key(): + src = make_src(_TypeA, pinned="MyCustomRef") + root = JSONSchema(ref=src) + + defs, schemas = make_resolver().resolve([root], local_ref_prefix=PREFIX, occupied_refs=()) + + assert list(defs.keys()) == ["MyCustomRef"] + assert schemas[0].ref == f"{PREFIX}MyCustomRef" + + +def test_pinning_conflict_raises(): + src_a = make_src(_TypeA, schema=JSONSchema(title="A"), pinned="Conflict") + src_b = make_src(_TypeB, schema=JSONSchema(title="B"), pinned="Conflict") + root_a = JSONSchema(ref=src_a) + root_b = JSONSchema(ref=src_b) + + with pytest.raises(ValueError, match="different sub schemas with pinned ref"): + make_resolver().resolve([root_a, root_b], local_ref_prefix=PREFIX, occupied_refs=()) + + +def test_occupied_refs_forces_mangled_name(): + src = make_src(_TypeA) + root = JSONSchema(ref=src) + + defs, _ = make_resolver(IndexRefMangler()).resolve([root], local_ref_prefix=PREFIX, occupied_refs=("_TypeA",)) + + assert list(defs.keys()) == ["_TypeA-1"] + + +def test_custom_local_ref_prefix(): + src = make_src(_TypeA) + root = JSONSchema(ref=src) + custom_prefix = "#/components/schemas/" + + _, schemas = make_resolver().resolve([root], local_ref_prefix=custom_prefix, occupied_refs=()) + + assert schemas[0].ref == f"{custom_prefix}_TypeA" + + +class _IdentityMangler(RefMangler): + def mangle_refs(self, occupied_refs, common_ref, sources_groups): + return {g: common_ref for g in sources_groups} + + +def test_mangling_failure_raises(): + src1 = LocalRefSource(value=None, json_schema=JSONSchema(title="V1"), loc_stack=LocStack(TypeHintLoc(_TypeA))) + src2 = LocalRefSource(value=None, json_schema=JSONSchema(title="V2"), loc_stack=LocStack(TypeHintLoc(_TypeA))) + root1 = JSONSchema(ref=src1) + root2 = JSONSchema(ref=src2) + resolver = BuiltinJSONSchemaResolver(BuiltinRefGenerator(), _IdentityMangler()) + + with pytest.raises(ValueError, match="cannot mangle some refs"): + resolver.resolve([root1, root2], local_ref_prefix=PREFIX, occupied_refs=()) + + +def test_multiple_unrelated_refs_produce_separate_defs(): + src_a = make_src(_TypeA) + src_b = make_src(_TypeB) + root_a = JSONSchema(ref=src_a) + root_b = JSONSchema(ref=src_b) + + defs, schemas = make_resolver().resolve([root_a, root_b], local_ref_prefix=PREFIX, occupied_refs=()) + + assert len(defs) == 2 + assert len(schemas) == 2 + + +def test_hasher_equal_schemas_are_equal(): + h1 = CustomJSONSchemaHasher(JSONSchema(title="Foo", description="bar")) + h2 = CustomJSONSchemaHasher(JSONSchema(title="Foo", description="bar")) + + assert h1 == h2 + assert hash(h1) == hash(h2) + + +def test_hasher_different_schemas_are_not_equal(): + h1 = CustomJSONSchemaHasher(JSONSchema(title="Foo")) + h2 = CustomJSONSchemaHasher(JSONSchema(title="Bar")) + + assert h1 != h2 + + +def test_hasher_empty_schemas_are_equal(): + h1 = CustomJSONSchemaHasher(JSONSchema()) + h2 = CustomJSONSchemaHasher(JSONSchema()) + + assert h1 == h2 + assert hash(h1) == hash(h2) diff --git a/tests/unit/morphing/json_schema/test_schema_tools.py b/tests/unit/morphing/json_schema/test_schema_tools.py new file mode 100644 index 00000000..4651c798 --- /dev/null +++ b/tests/unit/morphing/json_schema/test_schema_tools.py @@ -0,0 +1,221 @@ +import pytest + +from adaptix import Omitted +from adaptix._internal.morphing.json_schema.definitions import JSONSchema, LocalRefSource, RemoteRef, ResolvedJSONSchema +from adaptix._internal.morphing.json_schema.schema_tools import ( + approx_hash_json_schema, + replace_json_schema_ref, + traverse_json_schema, +) +from adaptix._internal.provider.loc_stack_filtering import LocStack +from adaptix._internal.provider.location import TypeHintLoc + + +def make_src(tp, schema=None): + if schema is None: + schema = JSONSchema() + return LocalRefSource(value=None, json_schema=schema, loc_stack=LocStack(TypeHintLoc(tp))) + + +def test_traverse_flat_schema_yields_itself(): + schema = JSONSchema(title="flat") + + result = list(traverse_json_schema(schema)) + + assert result == [schema] + + +def test_traverse_items_yields_root_and_item(): + item = JSONSchema(title="Item") + root = JSONSchema(items=item) + + result = list(traverse_json_schema(root)) + + assert result == [root, item] + + +def test_traverse_not_yields_root_and_subschema(): + sub = JSONSchema(title="sub") + root = JSONSchema(not_=sub) + + result = list(traverse_json_schema(root)) + + assert result == [root, sub] + + +def test_traverse_bool_schema_not_yielded(): + result = list(traverse_json_schema(True)) + + assert result == [] + + +def test_traverse_nested_deep_visits_all_levels(): + leaf = JSONSchema(title="leaf") + mid = JSONSchema(items=leaf) + root = JSONSchema(items=mid) + + result = list(traverse_json_schema(root)) + + assert result == [root, mid, leaf] + + +def test_traverse_properties_yields_root_and_all_values(): + prop_a = JSONSchema(title="A") + prop_b = JSONSchema(title="B") + root = JSONSchema(properties={"a": prop_a, "b": prop_b}) + + result = list(traverse_json_schema(root)) + + assert result == [root, prop_a, prop_b] + + +def test_traverse_all_of_yields_root_and_all_elements(): + s1, s2, s3 = JSONSchema(title="1"), JSONSchema(title="2"), JSONSchema(title="3") + root = JSONSchema(all_of=[s1, s2, s3]) + + result = list(traverse_json_schema(root)) + + assert result == [root, s1, s2, s3] + + +def test_traverse_any_of_yields_root_and_all_elements(): + s1, s2 = JSONSchema(title="x"), JSONSchema(title="y") + root = JSONSchema(any_of=[s1, s2]) + + result = list(traverse_json_schema(root)) + + assert result == [root, s1, s2] + + +def test_traverse_ref_to_local_source_visits_referenced_schema(): + inner = JSONSchema(title="inner") + src = make_src(int, schema=inner) + root = JSONSchema(ref=src) + + result = list(traverse_json_schema(root)) + + assert result == [root, inner] + + +@pytest.mark.parametrize( + "field_name", + ["content_schema", "if_", "then", "else_", "contains", "additional_properties", "property_names"], +) +def test_traverse_single_nested_schema_field(field_name): + child = JSONSchema(title="child") + root = JSONSchema(**{field_name: child}) + + result = list(traverse_json_schema(root)) + + assert result == [root, child] + + +@pytest.mark.parametrize( + "field_name", + ["one_of", "prefix_items"], +) +def test_traverse_sequence_nested_schema_field(field_name): + child1 = JSONSchema(title="child1") + child2 = JSONSchema(title="child2") + root = JSONSchema(**{field_name: [child1, child2]}) + + result = list(traverse_json_schema(root)) + + assert result == [root, child1, child2] + + +@pytest.mark.parametrize( + "field_name", + ["pattern_properties", "dependent_schemas", "defs"], +) +def test_traverse_mapping_nested_schema_field(field_name): + child = JSONSchema(title="child") + root = JSONSchema(**{field_name: {"key": child}}) + + result = list(traverse_json_schema(root)) + + assert result == [root, child] + + +@pytest.mark.parametrize( + ["s1", "s2"], + [ + (JSONSchema(title="foo", description="bar"), JSONSchema(title="foo", description="bar")), + (JSONSchema(), JSONSchema()), + (JSONSchema(extra_keywords={"a": 1, "b": 2}), JSONSchema(extra_keywords={"b": 2, "a": 1})), + ], + ids=["equal_fields", "empty", "extra_keywords_order_independent"], +) +def test_hash_equal_schemas_same_hash(s1, s2): + assert approx_hash_json_schema(s1) == approx_hash_json_schema(s2) + + +def test_hash_different_title_different_hash(): + s1 = JSONSchema(title="foo") + s2 = JSONSchema(title="bar") + + assert approx_hash_json_schema(s1) != approx_hash_json_schema(s2) + + +@pytest.mark.parametrize("value", [True, False]) +def test_hash_bool_schema(value): + assert approx_hash_json_schema(value) == hash(value) + + +def test_hash_nested_schema_field_contributes(): + s_with_items = JSONSchema(items=JSONSchema(title="Item")) + s_without = JSONSchema() + + assert approx_hash_json_schema(s_with_items) != approx_hash_json_schema(s_without) + + +def test_replace_local_ref_with_prefix_and_ctx(): + inner = JSONSchema(title="Inner") + src = make_src(int, schema=inner) + root = JSONSchema(ref=src) + ctx = {src: "MyInt"} + + result = replace_json_schema_ref(root, "#/$defs/", ctx) + + assert result.ref == "#/$defs/MyInt" + + +def test_replace_remote_ref_passes_through(): + root = JSONSchema(ref=RemoteRef("https://example.com/schema")) + + result = replace_json_schema_ref(root, "#/$defs/", {}) + + assert result.ref == "https://example.com/schema" + + +def test_replace_no_ref_stays_omitted(): + root = JSONSchema(title="plain") + + result = replace_json_schema_ref(root, "#/$defs/", {}) + + assert isinstance(result.ref, Omitted) + + +@pytest.mark.parametrize("value", [True, False]) +def test_replace_bool_schema_returns_bool(value): + assert replace_json_schema_ref(value, "#/$defs/", {}) is value + + +def test_replace_returns_resolved_json_schema_type(): + root = JSONSchema(title="plain") + + result = replace_json_schema_ref(root, "#/$defs/", {}) + + assert isinstance(result, ResolvedJSONSchema) + + +def test_replace_nested_properties_recursively_resolved(): + inner = JSONSchema(title="Inner") + src = make_src(str, schema=inner) + prop = JSONSchema(ref=src) + root = JSONSchema(properties={"field": prop}) + ctx = {src: "StrType"} + + result = replace_json_schema_ref(root, "#/$defs/", ctx) + + assert result.properties["field"].ref == "#/$defs/StrType" From 9a7ca0272fd3b60c0068d5308ae99aa5e121045a Mon Sep 17 00:00:00 2001 From: niccolum Date: Sat, 20 Jun 2026 10:59:21 +0300 Subject: [PATCH 2/6] fix linter --- .../tests_helpers/structure_tools.py | 52 +++++++++++-------- tests/unit/morphing/json_schema/test_patch.py | 2 - .../json_schema/test_ref_generator.py | 1 - .../morphing/json_schema/test_schema_tools.py | 3 +- 4 files changed, 33 insertions(+), 25 deletions(-) diff --git a/tests/tests_helpers/tests_helpers/structure_tools.py b/tests/tests_helpers/tests_helpers/structure_tools.py index f1911b78..cd45297b 100644 --- a/tests/tests_helpers/tests_helpers/structure_tools.py +++ b/tests/tests_helpers/tests_helpers/structure_tools.py @@ -16,6 +16,35 @@ class NotExists: NOT_EXISTS = NotExists() +def _assert_list_structure( + actual: list, + expected: list, + *, + path: str, + strict: bool, +) -> None: + assert isinstance(actual, list), ( + f"At {path!r}: expected list, got {type(actual).__name__}" + ) + if strict: + assert len(actual) == len(expected), ( + f"At {path!r}: expected list of length {len(expected)}, got {len(actual)}" + ) + else: + assert len(actual) >= len(expected), ( + f"At {path!r}: expected at least {len(expected)} elements, got {len(actual)}" + ) + for i, expected_item in enumerate(expected): + current_path = f"{path}[{i}]" + actual_item = actual[i] + if isinstance(expected_item, (dict, list)): + assert_structure(actual_item, expected_item, path=current_path, strict=strict) + else: + assert actual_item == expected_item, ( + f"At {current_path}: expected {expected_item!r}, got {actual_item!r}" + ) + + def assert_structure( actual: Union[Mapping[str, Any], list], expected: Union[Mapping[str, Any], list], @@ -48,26 +77,7 @@ def assert_structure( assert_structure([{"a": 1}, {"b": 2}], [{"a": 1}, {"b": EXISTS}]) """ if isinstance(expected, list): - assert isinstance(actual, list), ( - f"At {path!r}: expected list, got {type(actual).__name__}" - ) - if strict: - assert len(actual) == len(expected), ( - f"At {path!r}: expected list of length {len(expected)}, got {len(actual)}" - ) - else: - assert len(actual) >= len(expected), ( - f"At {path!r}: expected at least {len(expected)} elements, got {len(actual)}" - ) - for i, expected_item in enumerate(expected): - current_path = f"{path}[{i}]" - actual_item = actual[i] - if isinstance(expected_item, (dict, list)): - assert_structure(actual_item, expected_item, path=current_path, strict=strict) - else: - assert actual_item == expected_item, ( - f"At {current_path}: expected {expected_item!r}, got {actual_item!r}" - ) + _assert_list_structure(actual, expected, path=path, strict=strict) return # Check for extra keys in strict mode @@ -77,7 +87,7 @@ def assert_structure( extra_keys = actual_keys - expected_keys if extra_keys: raise AssertionError( - f"At {path!r}: unexpected keys: {sorted(extra_keys)}" + f"At {path!r}: unexpected keys: {sorted(extra_keys)}", ) for key, expected_value in expected.items(): diff --git a/tests/unit/morphing/json_schema/test_patch.py b/tests/unit/morphing/json_schema/test_patch.py index dc18c8b6..889689b3 100644 --- a/tests/unit/morphing/json_schema/test_patch.py +++ b/tests/unit/morphing/json_schema/test_patch.py @@ -1,5 +1,3 @@ -import pytest - from adaptix import Chain from adaptix._internal.morphing.json_schema.definitions import JSONSchema from adaptix._internal.morphing.json_schema.patch import JSONSchemaPatch diff --git a/tests/unit/morphing/json_schema/test_ref_generator.py b/tests/unit/morphing/json_schema/test_ref_generator.py index ac1ac525..72b3cc15 100644 --- a/tests/unit/morphing/json_schema/test_ref_generator.py +++ b/tests/unit/morphing/json_schema/test_ref_generator.py @@ -7,7 +7,6 @@ from adaptix._internal.provider.loc_stack_filtering import LocStack from adaptix._internal.provider.location import TypeHintLoc - generator = BuiltinRefGenerator() diff --git a/tests/unit/morphing/json_schema/test_schema_tools.py b/tests/unit/morphing/json_schema/test_schema_tools.py index 4651c798..b6ad845a 100644 --- a/tests/unit/morphing/json_schema/test_schema_tools.py +++ b/tests/unit/morphing/json_schema/test_schema_tools.py @@ -44,7 +44,8 @@ def test_traverse_not_yields_root_and_subschema(): def test_traverse_bool_schema_not_yielded(): - result = list(traverse_json_schema(True)) + bool_schema = True + result = list(traverse_json_schema(bool_schema)) assert result == [] From 79dd4c88d22adaf7f2ddfe6a57547bb145bd4f31 Mon Sep 17 00:00:00 2001 From: niccolum Date: Wed, 15 Jul 2026 00:03:07 +0300 Subject: [PATCH 3/6] fix by comments --- src/adaptix/_internal/morphing/facade/func.py | 3 +- .../_internal/morphing/json_schema/patch.py | 4 +- .../morphing/test_json_schema_facade.py | 245 ++++++++---------- .../morphing/test_json_schema_overrides.py | 95 ++++--- .../tests_helpers/structure_tools.py | 114 -------- tests/unit/morphing/json_schema/test_patch.py | 27 -- .../morphing/json_schema/test_schema_tools.py | 51 +--- 7 files changed, 183 insertions(+), 356 deletions(-) delete mode 100644 tests/tests_helpers/tests_helpers/structure_tools.py diff --git a/src/adaptix/_internal/morphing/facade/func.py b/src/adaptix/_internal/morphing/facade/func.py index 2863399b..3ff88948 100644 --- a/src/adaptix/_internal/morphing/facade/func.py +++ b/src/adaptix/_internal/morphing/facade/func.py @@ -61,7 +61,8 @@ def _format_loader(data): if isinstance(data, str): with suppress(ValueError): return JSONSchemaBuiltinFormat(data) - return data + return data + raise TypeLoadError(expected_type=str, input_value=data) _global_resolver = BuiltinJSONSchemaResolver( diff --git a/src/adaptix/_internal/morphing/json_schema/patch.py b/src/adaptix/_internal/morphing/json_schema/patch.py index 98a0a2ae..2353ec8b 100644 --- a/src/adaptix/_internal/morphing/json_schema/patch.py +++ b/src/adaptix/_internal/morphing/json_schema/patch.py @@ -16,8 +16,8 @@ def __init__(self) -> None: self._patchers: list[Patcher] = [] def _append_with_patcher(self: S, patcher: Callable[[JSONSchema], JSONSchema]) -> S: - self_copy = deepcopy(self) - self_copy._patchers.append(patcher) + self_copy = copy(self) + self_copy._patchers = [*self._patchers, patcher] return self_copy def mutate_copy( diff --git a/tests/integration/morphing/test_json_schema_facade.py b/tests/integration/morphing/test_json_schema_facade.py index f7c65bc1..a703155a 100644 --- a/tests/integration/morphing/test_json_schema_facade.py +++ b/tests/integration/morphing/test_json_schema_facade.py @@ -1,93 +1,35 @@ from dataclasses import dataclass -from typing import Any, Optional +from typing import Optional import pytest -from tests_helpers.structure_tools import EXISTS, NOT_EXISTS, assert_structure +from tests_helpers.morphing import JSONSchemaFork, assert_morphing -from adaptix import Retort +from adaptix import Retort, name_mapping from adaptix._internal.definitions import Direction from adaptix._internal.morphing.facade.func import ( + DIALECT_2020_12 as DIALECT_URI, generate_json_schema, generate_json_schemas_namespace, load_json_schema, ) -from adaptix._internal.morphing.json_schema.definitions import JSONSchema from adaptix._internal.morphing.json_schema.mangling import IndexRefMangler from adaptix._internal.morphing.json_schema.ref_generator import BuiltinRefGenerator from adaptix._internal.morphing.json_schema.resolver import BuiltinJSONSchemaResolver -from adaptix._internal.morphing.json_schema.schema_model import ( - JSONSchemaBuiltinFormat, - JSONSchemaType, -) +from adaptix._internal.morphing.json_schema.schema_model import JSONSchemaBuiltinFormat from adaptix.load_error import AggregateLoadError -DIALECT_URI = "https://json-schema.org/draft/2020-12/schema" - @pytest.mark.parametrize( ["raw", "expected"], [ - ("null", JSONSchemaType.NULL), - ("boolean", JSONSchemaType.BOOLEAN), - ("object", JSONSchemaType.OBJECT), - ("array", JSONSchemaType.ARRAY), - ("number", JSONSchemaType.NUMBER), - ("integer", JSONSchemaType.INTEGER), - ("string", JSONSchemaType.STRING), + ("date-time", JSONSchemaBuiltinFormat.DATE_TIME), + ("my-custom-format", "my-custom-format"), ], ) -def test_load_json_schema_type_enum_values(raw: str, expected: JSONSchemaType): - schema = load_json_schema({"type": raw}) +def test_load_json_schema_format_uses_builtin_enum_or_falls_back_to_str(raw, expected): + schema = load_json_schema({"type": "string", "format": raw}) - assert schema.type == expected - - -@pytest.mark.parametrize( - ["schema_data", "expected"], - [ - ( - {"type": "string", "title": "MyStr", "description": "A string"}, - JSONSchema(type=JSONSchemaType.STRING, title="MyStr", description="A string"), - ), - ( - {"type": "string", "format": "date-time"}, - JSONSchema(type=JSONSchemaType.STRING, format=JSONSchemaBuiltinFormat.DATE_TIME), - ), - ( - {"type": "string", "format": "my-custom-format"}, - JSONSchema(type=JSONSchemaType.STRING, format="my-custom-format"), - ), - ( - { - "type": "object", - "properties": { - "name": {"type": "string"}, - "age": {"type": "integer"}, - }, - "required": ["name"], - }, - JSONSchema( - type=JSONSchemaType.OBJECT, - properties={ - "name": JSONSchema(type=JSONSchemaType.STRING), - "age": JSONSchema(type=JSONSchemaType.INTEGER), - }, - required=["name"], - ), - ), - ( - {"anyOf": [{"type": "string"}, {"type": "integer"}]}, - JSONSchema(any_of=[ - JSONSchema(type=JSONSchemaType.STRING), - JSONSchema(type=JSONSchemaType.INTEGER), - ]), - ), - ], -) -def test_load_json_schema_parses_dict_to_object(schema_data, expected): - schema = load_json_schema(schema_data) - - assert schema == expected + assert schema.format == expected def test_load_json_schema_strict_unknown_field_raises(): @@ -101,12 +43,6 @@ def test_load_json_schema_lax_unknown_field_goes_to_extra_keywords(): assert schema.extra_keywords == {"x-custom": "value"} -def test_load_json_schema_empty_dict(): - schema = load_json_schema({}) - - assert schema == JSONSchema() - - @dataclass class SimpleModel: name: str @@ -120,29 +56,45 @@ class ModelWithOptional: @pytest.mark.parametrize( - ["with_dialect_uri", "expected_schema"], - [ - (True, DIALECT_URI), - (False, NOT_EXISTS), - ], + "with_dialect_uri", + [True, False], ids=["with_uri", "without_uri"], ) -def test_generate_json_schema_dialect_uri(with_dialect_uri: bool, expected_schema: Any): # noqa: FBT001 +def test_generate_json_schema_dialect_uri(with_dialect_uri: bool): # noqa: FBT001 schema = generate_json_schema(Retort(), SimpleModel, Direction.INPUT, with_dialect_uri=with_dialect_uri) - assert_structure(schema, {"$schema": expected_schema}) - - -def test_generate_json_schema_top_level_keys(): - schema = generate_json_schema(Retort(), SimpleModel, Direction.INPUT) - - assert_structure(schema, { - "$schema": EXISTS, - "$ref": EXISTS, - "$defs": { - "SimpleModel": EXISTS, - }, - }, strict=True) + assert ("$schema" in schema) == with_dialect_uri + + +def test_generate_json_schema_matches_full_expected_shape(): + def _make(extra: dict) -> dict: + return { + "$ref": "#/$defs/SimpleModel", + "$schema": DIALECT_URI, + "$defs": { + "SimpleModel": { + "title": "SimpleModel", + "type": "object", + "required": ["name", "value"], + "properties": { + "name": {"type": "string"}, + "value": {"type": "integer"}, + }, + **extra, + }, + }, + } + + assert_morphing( + retort=Retort(), + tp=SimpleModel, + data={"name": "foo", "value": 1}, + loaded=SimpleModel(name="foo", value=1), + json_schema=JSONSchemaFork( + input=_make({"additionalProperties": True}), + output=_make({}), + ), + ) def test_generate_json_schema_custom_ref_prefix(): @@ -151,7 +103,7 @@ def test_generate_json_schema_custom_ref_prefix(): local_ref_prefix="#/components/schemas/", ) - assert_structure(schema, {"$ref": "#/components/schemas/SimpleModel"}) + assert schema["$ref"] == "#/components/schemas/SimpleModel" def test_generate_json_schema_occupied_refs_triggers_mangling(): @@ -161,12 +113,8 @@ def test_generate_json_schema_occupied_refs_triggers_mangling(): resolver=BuiltinJSONSchemaResolver(BuiltinRefGenerator(), IndexRefMangler()), ) - assert_structure(schema, { - "$defs": { - "SimpleModel": NOT_EXISTS, - "SimpleModel-1": EXISTS, - }, - }) + assert "SimpleModel" not in schema["$defs"] + assert "SimpleModel-1" in schema["$defs"] def test_generate_json_schema_custom_resolver(): @@ -175,30 +123,68 @@ def test_generate_json_schema_custom_resolver(): resolver=BuiltinJSONSchemaResolver(BuiltinRefGenerator(), IndexRefMangler()), ) - assert_structure(schema, {"$defs": {"SimpleModel": EXISTS}}) + assert "SimpleModel" in schema["$defs"] def test_generate_json_schema_optional_field_not_required_on_input(): - input_schema = generate_json_schema(Retort(), ModelWithOptional, Direction.INPUT) - output_schema = generate_json_schema(Retort(), ModelWithOptional, Direction.OUTPUT) - - # Input schema: only required_field is required (optional_field has default) - assert_structure(input_schema, { - "$defs": { - "ModelWithOptional": { - "required": ["required_field"], - }, - }, - }) - - # Output schema: both fields are required (dumping doesn't care about defaults) - assert_structure(output_schema, { - "$defs": { - "ModelWithOptional": { - "required": ["required_field", "optional_field"], + def _make(required: list, extra: dict) -> dict: + return { + "$ref": "#/$defs/ModelWithOptional", + "$schema": DIALECT_URI, + "$defs": { + "ModelWithOptional": { + "title": "ModelWithOptional", + "type": "object", + "required": required, + "properties": { + "required_field": {"type": "string"}, + "optional_field": { + "default": None, + "anyOf": [{"type": "string"}, {"type": "null"}], + }, + }, + **extra, + }, }, - }, - }) + } + + assert_morphing( + retort=Retort(), + tp=ModelWithOptional, + data={"required_field": "x"}, + loaded=ModelWithOptional(required_field="x"), + dumped={"required_field": "x", "optional_field": None}, + json_schema=JSONSchemaFork( + # Input schema: only required_field is required (optional_field has a default) + input=_make(["required_field"], {"additionalProperties": True}), + # Output schema: both fields are required (dumping doesn't care about defaults) + output=_make(["required_field", "optional_field"], {}), + ), + ) + + +@dataclass +class ListItem: + x: int + y: int + + +@dataclass +class ListModel: + a: ListItem + b: ListItem + + +def test_generate_json_schema_prefix_items_dedupes_shared_ref(): + retort = Retort(recipe=[name_mapping(ListModel, as_list=True)]) + + schema = generate_json_schema(retort, ListModel, Direction.INPUT) + + assert schema["$defs"]["ListModel"]["prefixItems"] == [ + {"$ref": "#/$defs/ListItem"}, + {"$ref": "#/$defs/ListItem"}, + ] + assert "ListItem" in schema["$defs"] @dataclass @@ -240,12 +226,7 @@ def test_generate_schemas_namespace_deduplicates_shared_types(resolver): defs, schemas = generate_json_schemas_namespace(query, **kwargs) assert len(schemas) == 2 - - assert_structure(defs, { - "Article": EXISTS, - "Post": EXISTS, - "Tag": EXISTS, - }, strict=True) + assert defs.keys() == {"Article", "Post", "Tag"} @pytest.mark.parametrize( @@ -259,9 +240,9 @@ def test_generate_schemas_namespace_dialect_uri(with_dialect_uri: bool): # noqa query = _namespace_query() _, schemas = generate_json_schemas_namespace(query, with_dialect_uri=with_dialect_uri) - expected_schema = DIALECT_URI if with_dialect_uri else NOT_EXISTS for schema in schemas: - assert_structure(schema, {"$schema": expected_schema, "$ref": EXISTS}) + assert ("$schema" in schema) == with_dialect_uri + assert "$ref" in schema def test_generate_schemas_namespace_custom_ref_prefix(): @@ -270,7 +251,7 @@ def test_generate_schemas_namespace_custom_ref_prefix(): query, local_ref_prefix="#/components/schemas/", ) - assert_structure(schemas, [ - {"$ref": "#/components/schemas/Article"}, - {"$ref": "#/components/schemas/Post"}, - ]) + assert [schema["$ref"] for schema in schemas] == [ + "#/components/schemas/Article", + "#/components/schemas/Post", + ] diff --git a/tests/integration/morphing/test_json_schema_overrides.py b/tests/integration/morphing/test_json_schema_overrides.py index a222d4e1..17d969d6 100644 --- a/tests/integration/morphing/test_json_schema_overrides.py +++ b/tests/integration/morphing/test_json_schema_overrides.py @@ -2,11 +2,10 @@ import pytest from tests_helpers.morphing import JSONSchemaFork, assert_morphing -from tests_helpers.structure_tools import EXISTS, NOT_EXISTS, assert_structure from adaptix import Chain, P, ProviderNotFoundError, Retort, dumper, json_schema, loader from adaptix._internal.definitions import Direction -from adaptix._internal.morphing.facade.func import generate_json_schema +from adaptix._internal.morphing.facade.func import DIALECT_2020_12, generate_json_schema from adaptix._internal.morphing.json_schema.definitions import JSONSchema from adaptix._internal.morphing.json_schema.patch import JSONSchemaPatch from adaptix._internal.morphing.json_schema.providers import EraseJSONSchema, KeepJSONSchema @@ -19,7 +18,6 @@ class Product: price: float -_DIALECT_URI = "https://json-schema.org/draft/2020-12/schema" _PRODUCT_DATA = {"name": "test", "price": 1.5} _PRODUCT_LOADED = Product(name="test", price=1.5) @@ -28,7 +26,7 @@ def _product_schema(name_prop: dict, price_prop: dict) -> JSONSchemaFork: def _make(product_extra: dict = {}) -> dict: # noqa: B006 return { "$ref": "#/$defs/Product", - "$schema": _DIALECT_URI, + "$schema": DIALECT_2020_12, "$defs": { "Product": { "title": "Product", @@ -52,15 +50,25 @@ def _make(product_extra: dict = {}) -> dict: # noqa: B006 _DEFAULT_SCHEMA = _product_schema({"type": "string"}, {"type": "number"}) +def _product_schema_with_str_def(name_prop: dict, price_prop: dict, def_name: str) -> JSONSchemaFork: + fork = _product_schema(name_prop, price_prop) + str_def = {def_name: {"type": "string"}} + return JSONSchemaFork( + input={**fork.input, "$defs": {**fork.input["$defs"], **str_def}}, + output={**fork.output, "$defs": {**fork.output["$defs"], **str_def}}, + ) + + def test_json_schema_explicit_replaces_inferred(): + # explicit override is applied verbatim, even though it no longer matches + # the field's actual (str) load/dump behavior -- that mismatch is the point of this test, + # so it can't go through assert_morphing's data round-trip + jsonschema validation. retort = Retort(recipe=[ json_schema(P[Product].name, JSONSchema(type=JSONSchemaType.INTEGER)), ]) schema = generate_json_schema(retort, Product, Direction.INPUT) - assert_structure(schema, { - "$defs": {"Product": {"properties": {"name": {"type": "integer"}}}}, - }) + assert schema["$defs"]["Product"]["properties"]["name"] == {"type": "integer"} def test_json_schema_explicit_empty_schema(): @@ -153,9 +161,27 @@ def test_patch_merge_with_chain_first_overrides_inferred(): ]) schema = generate_json_schema(retort, Product, Direction.INPUT) - assert_structure(schema, { - "$defs": {"Product": {"properties": {"price": {"type": "string"}}}}, - }) + assert schema["$defs"]["Product"]["properties"]["price"] == {"type": "string"} + + +def test_patch_merge_with_chain_last_base_wins_over_conflicting_override(): + retort = Retort(recipe=[ + json_schema( + P[Product].name, + JSONSchemaPatch().merge_with(JSONSchema(type=JSONSchemaType.INTEGER), Chain.LAST), + ), + ]) + + assert_morphing( + retort=retort, + tp=Product, + data=_PRODUCT_DATA, + loaded=_PRODUCT_LOADED, + json_schema=_product_schema( + {"type": "string"}, + {"type": "number"}, + ), + ) def test_patch_replace_title(): @@ -178,33 +204,42 @@ def test_patch_replace_title(): ) -@pytest.mark.parametrize( - ["inline", "expected_ref"], - [ - pytest.param(True, NOT_EXISTS, id="inline_true_embeds_directly"), - pytest.param(False, EXISTS, id="inline_false_uses_ref"), - ], -) -def test_json_schema_inline_controls_ref_embedding(inline, expected_ref): - retort = Retort(recipe=[json_schema(str, inline=inline)]) - schema = generate_json_schema(retort, Product, Direction.INPUT) +def test_json_schema_inline_embeds_directly(): + retort = Retort(recipe=[json_schema(str, inline=True)]) - assert_structure( - schema["$defs"]["Product"]["properties"]["name"], - {"$ref": expected_ref}, + assert_morphing( + retort=retort, + tp=Product, + data=_PRODUCT_DATA, + loaded=_PRODUCT_LOADED, + json_schema=_DEFAULT_SCHEMA, + ) + + +def test_json_schema_not_inline_uses_ref(): + retort = Retort(recipe=[json_schema(str, inline=False)]) + + assert_morphing( + retort=retort, + tp=Product, + data=_PRODUCT_DATA, + loaded=_PRODUCT_LOADED, + json_schema=_product_schema_with_str_def({"$ref": "#/$defs/str"}, {"type": "number"}, "str"), ) def test_json_schema_pinned_ref(): retort = Retort(recipe=[json_schema(str, ref="MyString", inline=False)]) - schema = generate_json_schema(retort, Product, Direction.INPUT) - assert_structure(schema, { - "$defs": { - "MyString": EXISTS, - "Product": {"properties": {"name": {"$ref": "#/$defs/MyString"}}}, - }, - }) + assert_morphing( + retort=retort, + tp=Product, + data=_PRODUCT_DATA, + loaded=_PRODUCT_LOADED, + json_schema=_product_schema_with_str_def( + {"$ref": "#/$defs/MyString"}, {"type": "number"}, "MyString", + ), + ) def test_loader_custom_json_schema_kwarg_reflected_in_schema(): diff --git a/tests/tests_helpers/tests_helpers/structure_tools.py b/tests/tests_helpers/tests_helpers/structure_tools.py deleted file mode 100644 index cd45297b..00000000 --- a/tests/tests_helpers/tests_helpers/structure_tools.py +++ /dev/null @@ -1,114 +0,0 @@ -from dataclasses import dataclass -from typing import Any, Mapping, Union - - -@dataclass(slots=True) -class Exists: - """Marker to check that a key exists in the dict (value type not checked).""" - - -@dataclass(slots=True) -class NotExists: - """Marker to check that a key does NOT exist in the dict.""" - - -EXISTS = Exists() -NOT_EXISTS = NotExists() - - -def _assert_list_structure( - actual: list, - expected: list, - *, - path: str, - strict: bool, -) -> None: - assert isinstance(actual, list), ( - f"At {path!r}: expected list, got {type(actual).__name__}" - ) - if strict: - assert len(actual) == len(expected), ( - f"At {path!r}: expected list of length {len(expected)}, got {len(actual)}" - ) - else: - assert len(actual) >= len(expected), ( - f"At {path!r}: expected at least {len(expected)} elements, got {len(actual)}" - ) - for i, expected_item in enumerate(expected): - current_path = f"{path}[{i}]" - actual_item = actual[i] - if isinstance(expected_item, (dict, list)): - assert_structure(actual_item, expected_item, path=current_path, strict=strict) - else: - assert actual_item == expected_item, ( - f"At {current_path}: expected {expected_item!r}, got {actual_item!r}" - ) - - -def assert_structure( - actual: Union[Mapping[str, Any], list], - expected: Union[Mapping[str, Any], list], - *, - path: str = "", - strict: bool = False, -) -> None: - """ - Assert that a dict or list matches the expected structure. - - Supports nested dicts and lists. - For dict values, supports: - - EXISTS: check that key exists (any value) - - NOT_EXISTS: check that key does NOT exist - - Other values: direct equality check or recursive structure check - - Args: - actual: The actual dict or list to check - expected: Dict or list describing expected structure - path: Current path for error messages (used internally for recursion) - strict: If True, fail if actual has keys/elements not in expected - - Example: - assert_structure(schema, { - "$ref": "#/$defs/SimpleModel", - "$defs": { - "SimpleModel": {}, - }, - }) - assert_structure([{"a": 1}, {"b": 2}], [{"a": 1}, {"b": EXISTS}]) - """ - if isinstance(expected, list): - _assert_list_structure(actual, expected, path=path, strict=strict) - return - - # Check for extra keys in strict mode - if strict: - expected_keys = set(expected.keys()) - actual_keys = set(actual.keys()) - extra_keys = actual_keys - expected_keys - if extra_keys: - raise AssertionError( - f"At {path!r}: unexpected keys: {sorted(extra_keys)}", - ) - - for key, expected_value in expected.items(): - current_path = f"{path}.{key}" if path else key - - if expected_value is NOT_EXISTS: - assert key not in actual, ( - f"Key {current_path!r} should not exist, but found: {actual[key]!r}" - ) - continue - - assert key in actual, f"Key {current_path!r} not found in dict" - - if expected_value is EXISTS: - continue - - actual_value = actual[key] - - if isinstance(expected_value, (dict, list)): - assert_structure(actual_value, expected_value, path=current_path, strict=strict) - else: - assert actual_value == expected_value, ( - f"At {current_path!r}: expected {expected_value!r}, got {actual_value!r}" - ) diff --git a/tests/unit/morphing/json_schema/test_patch.py b/tests/unit/morphing/json_schema/test_patch.py index 889689b3..5620748e 100644 --- a/tests/unit/morphing/json_schema/test_patch.py +++ b/tests/unit/morphing/json_schema/test_patch.py @@ -9,13 +9,6 @@ def apply_patch(patch: JSONSchemaPatch, schema: JSONSchema) -> JSONSchema: return schema -def test_replace_sets_field(): - schema = JSONSchema(title="old") - - result = apply_patch(JSONSchemaPatch().replace("title", lambda _: "new"), schema) - - assert result.title == "new" - def test_replace_receives_old_value(): schema = JSONSchema(title="base") @@ -93,26 +86,6 @@ def test_mutate_deepcopy_inner_is_separate_object(): assert result.extra_keywords["items"] is not inner_list -def test_merge_with_chain_first_override_wins(): - base = JSONSchema(title="base", description="base_desc") - override = JSONSchema(description="new_desc") - - result = apply_patch(JSONSchemaPatch().merge_with(override, Chain.FIRST), base) - - assert result.title == "base" - assert result.description == "new_desc" - - -def test_merge_with_chain_last_base_wins(): - base = JSONSchema(title="base", description="base_desc") - override = JSONSchema(description="new_desc") - - result = apply_patch(JSONSchemaPatch().merge_with(override, Chain.LAST), base) - - assert result.title == "base" - assert result.description == "base_desc" - - def test_merge_with_chain_first_omitted_field_does_not_overwrite(): base = JSONSchema(title="keep_me") override = JSONSchema(description="added") diff --git a/tests/unit/morphing/json_schema/test_schema_tools.py b/tests/unit/morphing/json_schema/test_schema_tools.py index b6ad845a..6d677fdf 100644 --- a/tests/unit/morphing/json_schema/test_schema_tools.py +++ b/tests/unit/morphing/json_schema/test_schema_tools.py @@ -1,28 +1,12 @@ import pytest from adaptix import Omitted -from adaptix._internal.morphing.json_schema.definitions import JSONSchema, LocalRefSource, RemoteRef, ResolvedJSONSchema +from adaptix._internal.morphing.json_schema.definitions import JSONSchema, RemoteRef, ResolvedJSONSchema from adaptix._internal.morphing.json_schema.schema_tools import ( approx_hash_json_schema, replace_json_schema_ref, traverse_json_schema, ) -from adaptix._internal.provider.loc_stack_filtering import LocStack -from adaptix._internal.provider.location import TypeHintLoc - - -def make_src(tp, schema=None): - if schema is None: - schema = JSONSchema() - return LocalRefSource(value=None, json_schema=schema, loc_stack=LocStack(TypeHintLoc(tp))) - - -def test_traverse_flat_schema_yields_itself(): - schema = JSONSchema(title="flat") - - result = list(traverse_json_schema(schema)) - - assert result == [schema] def test_traverse_items_yields_root_and_item(): @@ -88,16 +72,6 @@ def test_traverse_any_of_yields_root_and_all_elements(): assert result == [root, s1, s2] -def test_traverse_ref_to_local_source_visits_referenced_schema(): - inner = JSONSchema(title="inner") - src = make_src(int, schema=inner) - root = JSONSchema(ref=src) - - result = list(traverse_json_schema(root)) - - assert result == [root, inner] - - @pytest.mark.parametrize( "field_name", ["content_schema", "if_", "then", "else_", "contains", "additional_properties", "property_names"], @@ -170,17 +144,6 @@ def test_hash_nested_schema_field_contributes(): assert approx_hash_json_schema(s_with_items) != approx_hash_json_schema(s_without) -def test_replace_local_ref_with_prefix_and_ctx(): - inner = JSONSchema(title="Inner") - src = make_src(int, schema=inner) - root = JSONSchema(ref=src) - ctx = {src: "MyInt"} - - result = replace_json_schema_ref(root, "#/$defs/", ctx) - - assert result.ref == "#/$defs/MyInt" - - def test_replace_remote_ref_passes_through(): root = JSONSchema(ref=RemoteRef("https://example.com/schema")) @@ -208,15 +171,3 @@ def test_replace_returns_resolved_json_schema_type(): result = replace_json_schema_ref(root, "#/$defs/", {}) assert isinstance(result, ResolvedJSONSchema) - - -def test_replace_nested_properties_recursively_resolved(): - inner = JSONSchema(title="Inner") - src = make_src(str, schema=inner) - prop = JSONSchema(ref=src) - root = JSONSchema(properties={"field": prop}) - ctx = {src: "StrType"} - - result = replace_json_schema_ref(root, "#/$defs/", ctx) - - assert result.properties["field"].ref == "#/$defs/StrType" From b250b238387af8e2d1f639c0a5f53cca204456e2 Mon Sep 17 00:00:00 2001 From: niccolum Date: Tue, 6 Oct 2026 19:43:21 +0300 Subject: [PATCH 4/6] fix tests --- .gitignore | 1 - .../morphing/test_json_schema_facade.py | 173 ++++++------------ .../morphing/test_json_schema_overrides.py | 77 ++++---- 3 files changed, 104 insertions(+), 147 deletions(-) diff --git a/.gitignore b/.gitignore index 127bf229..1a935416 100644 --- a/.gitignore +++ b/.gitignore @@ -54,4 +54,3 @@ dmypy.json /docs/build .venv/ -uv.lock \ No newline at end of file diff --git a/tests/integration/morphing/test_json_schema_facade.py b/tests/integration/morphing/test_json_schema_facade.py index a703155a..09167ec8 100644 --- a/tests/integration/morphing/test_json_schema_facade.py +++ b/tests/integration/morphing/test_json_schema_facade.py @@ -1,22 +1,25 @@ from dataclasses import dataclass from typing import Optional +import jsonschema import pytest -from tests_helpers.morphing import JSONSchemaFork, assert_morphing +from tests_helpers import raises_exc +from tests_helpers.morphing import JSONSchemaOptItem, assert_morphing from adaptix import Retort, name_mapping from adaptix._internal.definitions import Direction from adaptix._internal.morphing.facade.func import ( - DIALECT_2020_12 as DIALECT_URI, + DIALECT_2020_12, generate_json_schema, generate_json_schemas_namespace, load_json_schema, ) +from adaptix._internal.morphing.json_schema.definitions import JSONSchema from adaptix._internal.morphing.json_schema.mangling import IndexRefMangler from adaptix._internal.morphing.json_schema.ref_generator import BuiltinRefGenerator -from adaptix._internal.morphing.json_schema.resolver import BuiltinJSONSchemaResolver +from adaptix._internal.morphing.json_schema.resolver import BuiltinJSONSchemaResolver, RefGenerator from adaptix._internal.morphing.json_schema.schema_model import JSONSchemaBuiltinFormat -from adaptix.load_error import AggregateLoadError +from adaptix.load_error import AggregateLoadError, ExtraFieldsLoadError @pytest.mark.parametrize( @@ -33,8 +36,15 @@ def test_load_json_schema_format_uses_builtin_enum_or_falls_back_to_str(raw, exp def test_load_json_schema_strict_unknown_field_raises(): - with pytest.raises(AggregateLoadError): - load_json_schema({"type": "string", "x-custom": "value"}, error_on_extra=True) + data = {"type": "string", "x-custom": "value"} + + raises_exc( + AggregateLoadError( + f"while loading model {JSONSchema}", + [ExtraFieldsLoadError({"x-custom"}, data)], + ), + lambda: load_json_schema(data, error_on_extra=True), + ) def test_load_json_schema_lax_unknown_field_goes_to_extra_keywords(): @@ -67,10 +77,14 @@ def test_generate_json_schema_dialect_uri(with_dialect_uri: bool): # noqa: FBT0 def test_generate_json_schema_matches_full_expected_shape(): - def _make(extra: dict) -> dict: - return { + assert_morphing( + retort=Retort(), + tp=SimpleModel, + data={"name": "foo", "value": 1}, + loaded=SimpleModel(name="foo", value=1), + json_schema={ "$ref": "#/$defs/SimpleModel", - "$schema": DIALECT_URI, + "$schema": DIALECT_2020_12, "$defs": { "SimpleModel": { "title": "SimpleModel", @@ -80,20 +94,10 @@ def _make(extra: dict) -> dict: "name": {"type": "string"}, "value": {"type": "integer"}, }, - **extra, + "additionalProperties": JSONSchemaOptItem(input=True), }, }, - } - - assert_morphing( - retort=Retort(), - tp=SimpleModel, - data={"name": "foo", "value": 1}, - loaded=SimpleModel(name="foo", value=1), - json_schema=JSONSchemaFork( - input=_make({"additionalProperties": True}), - output=_make({}), - ), + }, ) @@ -106,62 +110,48 @@ def test_generate_json_schema_custom_ref_prefix(): assert schema["$ref"] == "#/components/schemas/SimpleModel" -def test_generate_json_schema_occupied_refs_triggers_mangling(): - schema = generate_json_schema( - Retort(), SimpleModel, Direction.INPUT, - occupied_refs=("SimpleModel",), - resolver=BuiltinJSONSchemaResolver(BuiltinRefGenerator(), IndexRefMangler()), - ) +class _PrefixRefGenerator(RefGenerator): + def generate_ref(self, json_schema, loc_stack): + return "Custom" + BuiltinRefGenerator().generate_ref(json_schema, loc_stack) - assert "SimpleModel" not in schema["$defs"] - assert "SimpleModel-1" in schema["$defs"] - -def test_generate_json_schema_custom_resolver(): +@pytest.mark.parametrize( + ["resolver", "occupied_refs", "expected_ref"], + [ + pytest.param( + BuiltinJSONSchemaResolver(BuiltinRefGenerator(), IndexRefMangler()), + ("SimpleModel",), + "SimpleModel-1", + id="mangling_of_occupied_ref", + ), + pytest.param( + BuiltinJSONSchemaResolver(_PrefixRefGenerator(), IndexRefMangler()), + (), + "CustomSimpleModel", + id="custom_ref_generator", + ), + ], +) +def test_generate_json_schema_custom_resolver(resolver, occupied_refs, expected_ref): schema = generate_json_schema( Retort(), SimpleModel, Direction.INPUT, - resolver=BuiltinJSONSchemaResolver(BuiltinRefGenerator(), IndexRefMangler()), + resolver=resolver, + occupied_refs=occupied_refs, ) - assert "SimpleModel" in schema["$defs"] + assert schema["$ref"] == f"#/$defs/{expected_ref}" def test_generate_json_schema_optional_field_not_required_on_input(): - def _make(required: list, extra: dict) -> dict: - return { - "$ref": "#/$defs/ModelWithOptional", - "$schema": DIALECT_URI, - "$defs": { - "ModelWithOptional": { - "title": "ModelWithOptional", - "type": "object", - "required": required, - "properties": { - "required_field": {"type": "string"}, - "optional_field": { - "default": None, - "anyOf": [{"type": "string"}, {"type": "null"}], - }, - }, - **extra, - }, - }, - } - - assert_morphing( - retort=Retort(), - tp=ModelWithOptional, - data={"required_field": "x"}, - loaded=ModelWithOptional(required_field="x"), - dumped={"required_field": "x", "optional_field": None}, - json_schema=JSONSchemaFork( - # Input schema: only required_field is required (optional_field has a default) - input=_make(["required_field"], {"additionalProperties": True}), - # Output schema: both fields are required (dumping doesn't care about defaults) - output=_make(["required_field", "optional_field"], {}), - ), + retort = Retort() + input_validator = jsonschema.Draft202012Validator( + generate_json_schema(retort, ModelWithOptional, Direction.INPUT), ) + # a field with a default can be omitted on loading + assert input_validator.is_valid({"required_field": "x"}) + assert not input_validator.is_valid({"optional_field": "y"}) + @dataclass class ListItem: @@ -204,54 +194,13 @@ class Post: tag: Tag -def _namespace_query(): +def test_generate_schemas_namespace_deduplicates_shared_types(): retort = Retort() - return [(retort, Direction.INPUT, Article), (retort, Direction.INPUT, Post)] - -@pytest.mark.parametrize( - "resolver", - [ - pytest.param(None, id="default"), - pytest.param( - BuiltinJSONSchemaResolver(BuiltinRefGenerator(), IndexRefMangler()), - id="custom_resolver", - ), - ], -) -def test_generate_schemas_namespace_deduplicates_shared_types(resolver): - query = _namespace_query() - kwargs = {"resolver": resolver} if resolver else {} - - defs, schemas = generate_json_schemas_namespace(query, **kwargs) + defs, schemas = generate_json_schemas_namespace([ + (retort, Direction.INPUT, Article), + (retort, Direction.INPUT, Post), + ]) assert len(schemas) == 2 assert defs.keys() == {"Article", "Post", "Tag"} - - -@pytest.mark.parametrize( - "with_dialect_uri", - [ - pytest.param(True, id="with_uri"), - pytest.param(False, id="without_uri"), - ], -) -def test_generate_schemas_namespace_dialect_uri(with_dialect_uri: bool): # noqa: FBT001 - query = _namespace_query() - _, schemas = generate_json_schemas_namespace(query, with_dialect_uri=with_dialect_uri) - - for schema in schemas: - assert ("$schema" in schema) == with_dialect_uri - assert "$ref" in schema - - -def test_generate_schemas_namespace_custom_ref_prefix(): - query = _namespace_query() - _, schemas = generate_json_schemas_namespace( - query, local_ref_prefix="#/components/schemas/", - ) - - assert [schema["$ref"] for schema in schemas] == [ - "#/components/schemas/Article", - "#/components/schemas/Post", - ] diff --git a/tests/integration/morphing/test_json_schema_overrides.py b/tests/integration/morphing/test_json_schema_overrides.py index 17d969d6..8be32016 100644 --- a/tests/integration/morphing/test_json_schema_overrides.py +++ b/tests/integration/morphing/test_json_schema_overrides.py @@ -1,7 +1,7 @@ from dataclasses import dataclass -import pytest -from tests_helpers.morphing import JSONSchemaFork, assert_morphing +from tests_helpers import raises_exc +from tests_helpers.morphing import JSONSchemaOptItem, assert_morphing from adaptix import Chain, P, ProviderNotFoundError, Retort, dumper, json_schema, loader from adaptix._internal.definitions import Direction @@ -22,41 +22,34 @@ class Product: _PRODUCT_LOADED = Product(name="test", price=1.5) -def _product_schema(name_prop: dict, price_prop: dict) -> JSONSchemaFork: - def _make(product_extra: dict = {}) -> dict: # noqa: B006 - return { - "$ref": "#/$defs/Product", - "$schema": DIALECT_2020_12, - "$defs": { - "Product": { - "title": "Product", - "type": "object", - "required": ["name", "price"], - "properties": { - "name": name_prop, - "price": price_prop, - }, - **product_extra, +def _product_schema(name_prop: dict, price_prop: dict) -> dict: + return { + "$ref": "#/$defs/Product", + "$schema": DIALECT_2020_12, + "$defs": { + "Product": { + "title": "Product", + "type": "object", + "required": ["name", "price"], + "properties": { + "name": name_prop, + "price": price_prop, }, + "additionalProperties": JSONSchemaOptItem(input=True), }, - } - - return JSONSchemaFork( - input=_make({"additionalProperties": True}), - output=_make(), - ) + }, + } _DEFAULT_SCHEMA = _product_schema({"type": "string"}, {"type": "number"}) -def _product_schema_with_str_def(name_prop: dict, price_prop: dict, def_name: str) -> JSONSchemaFork: - fork = _product_schema(name_prop, price_prop) - str_def = {def_name: {"type": "string"}} - return JSONSchemaFork( - input={**fork.input, "$defs": {**fork.input["$defs"], **str_def}}, - output={**fork.output, "$defs": {**fork.output["$defs"], **str_def}}, - ) +def _product_schema_with_str_def(name_prop: dict, price_prop: dict, def_name: str) -> dict: + schema = _product_schema(name_prop, price_prop) + return { + **schema, + "$defs": {**schema["$defs"], def_name: {"type": "string"}}, + } def test_json_schema_explicit_replaces_inferred(): @@ -119,8 +112,16 @@ def test_erase_json_schema_on_loader_raises_error(): loader(str, lambda x: x, json_schema=EraseJSONSchema()), ]) - with pytest.raises(ProviderNotFoundError): - generate_json_schema(retort, Product, Direction.INPUT) + raises_exc( + ProviderNotFoundError( + f"Cannot produce JSONSchema for type {Product}", + " × Cannot create JSON Schema for model. JSON Schemas for some fields cannot be created\n" + " │ Location: ‹Product›\n" + " ╰──▷ JSON Schema is erased\n" + " Location: ‹Product.name: str›", + ), + lambda: generate_json_schema(retort, Product, Direction.INPUT), + ) def test_erase_json_schema_on_dumper_raises_error(): @@ -128,8 +129,16 @@ def test_erase_json_schema_on_dumper_raises_error(): dumper(str, lambda x: x, json_schema=EraseJSONSchema()), ]) - with pytest.raises(ProviderNotFoundError): - generate_json_schema(retort, Product, Direction.OUTPUT) + raises_exc( + ProviderNotFoundError( + f"Cannot produce JSONSchema for type {Product}", + " × \n" + " │ Location: ‹Product›\n" + " ╰──▷ JSON Schema is erased\n" + " Location: ‹Product.name: str›", + ), + lambda: generate_json_schema(retort, Product, Direction.OUTPUT), + ) def test_patch_merge_with_adds_description(): From 9e8955bf73b25d2551b69deecbda9a967ac7301d Mon Sep 17 00:00:00 2001 From: niccolum Date: Tue, 6 Oct 2026 19:51:02 +0300 Subject: [PATCH 5/6] delete unuseful test --- .../morphing/test_json_schema_facade.py | 19 ------------------- 1 file changed, 19 deletions(-) diff --git a/tests/integration/morphing/test_json_schema_facade.py b/tests/integration/morphing/test_json_schema_facade.py index 09167ec8..da16cf76 100644 --- a/tests/integration/morphing/test_json_schema_facade.py +++ b/tests/integration/morphing/test_json_schema_facade.py @@ -1,7 +1,5 @@ from dataclasses import dataclass -from typing import Optional -import jsonschema import pytest from tests_helpers import raises_exc from tests_helpers.morphing import JSONSchemaOptItem, assert_morphing @@ -59,12 +57,6 @@ class SimpleModel: value: int -@dataclass -class ModelWithOptional: - required_field: str - optional_field: Optional[str] = None - - @pytest.mark.parametrize( "with_dialect_uri", [True, False], @@ -142,17 +134,6 @@ def test_generate_json_schema_custom_resolver(resolver, occupied_refs, expected_ assert schema["$ref"] == f"#/$defs/{expected_ref}" -def test_generate_json_schema_optional_field_not_required_on_input(): - retort = Retort() - input_validator = jsonschema.Draft202012Validator( - generate_json_schema(retort, ModelWithOptional, Direction.INPUT), - ) - - # a field with a default can be omitted on loading - assert input_validator.is_valid({"required_field": "x"}) - assert not input_validator.is_valid({"optional_field": "y"}) - - @dataclass class ListItem: x: int From 51fc547ff6e9626cdeae8d65930df7851d44eb7d Mon Sep 17 00:00:00 2001 From: niccolum Date: Tue, 6 Oct 2026 20:14:11 +0300 Subject: [PATCH 6/6] fix test --- tests/unit/type_tools/test_normalize_type.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/unit/type_tools/test_normalize_type.py b/tests/unit/type_tools/test_normalize_type.py index 680c0d46..eabaaf73 100644 --- a/tests/unit/type_tools/test_normalize_type.py +++ b/tests/unit/type_tools/test_normalize_type.py @@ -834,11 +834,12 @@ def test_forward_ref_at_type_var_limit(): ), ) + # ParamSpec keeps a string bound as is or wraps it into ForwardRef depending on the Python version assert_norm_tv( P_FR1, NormParamSpec( P_FR1, - limit=Bound(nt_zero(int, source=ForwardRef("int"))), + limit=Bound(nt_zero(int, source=P_FR1.__bound__)), source=P_FR1, default=None, ), @@ -847,7 +848,7 @@ def test_forward_ref_at_type_var_limit(): P_FR2, NormParamSpec( P_FR2, - limit=Bound(nt_zero(MyForwardClass, source=ForwardRef("MyForwardClass"))), + limit=Bound(nt_zero(MyForwardClass, source=P_FR2.__bound__)), source=P_FR2, default=None, ),