diff --git a/src/adaptix/_internal/morphing/facade/func.py b/src/adaptix/_internal/morphing/facade/func.py index 63913e9d..3ff88948 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,14 @@ 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 + raise TypeLoadError(expected_type=str, input_value=data) + + _global_resolver = BuiltinJSONSchemaResolver( ref_generator=BuiltinRefGenerator(), ref_mangler=CompoundRefMangler(QualnameRefMangler(), IndexRefMangler()), @@ -74,6 +83,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..2353ec8b 100644 --- a/src/adaptix/_internal/morphing/json_schema/patch.py +++ b/src/adaptix/_internal/morphing/json_schema/patch.py @@ -17,7 +17,7 @@ def __init__(self) -> None: def _append_with_patcher(self: S, patcher: Callable[[JSONSchema], JSONSchema]) -> S: self_copy = copy(self) - self_copy._patchers.append(patcher) + self_copy._patchers = [*self._patchers, patcher] return self_copy def mutate_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..da16cf76 --- /dev/null +++ b/tests/integration/morphing/test_json_schema_facade.py @@ -0,0 +1,187 @@ +from dataclasses import dataclass + +import pytest +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, + 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, RefGenerator +from adaptix._internal.morphing.json_schema.schema_model import JSONSchemaBuiltinFormat +from adaptix.load_error import AggregateLoadError, ExtraFieldsLoadError + + +@pytest.mark.parametrize( + ["raw", "expected"], + [ + ("date-time", JSONSchemaBuiltinFormat.DATE_TIME), + ("my-custom-format", "my-custom-format"), + ], +) +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.format == expected + + +def test_load_json_schema_strict_unknown_field_raises(): + 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(): + schema = load_json_schema({"type": "string", "x-custom": "value"}, error_on_extra=False) + + assert schema.extra_keywords == {"x-custom": "value"} + + +@dataclass +class SimpleModel: + name: str + value: int + + +@pytest.mark.parametrize( + "with_dialect_uri", + [True, False], + ids=["with_uri", "without_uri"], +) +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 ("$schema" in schema) == with_dialect_uri + + +def test_generate_json_schema_matches_full_expected_shape(): + assert_morphing( + retort=Retort(), + tp=SimpleModel, + data={"name": "foo", "value": 1}, + loaded=SimpleModel(name="foo", value=1), + json_schema={ + "$ref": "#/$defs/SimpleModel", + "$schema": DIALECT_2020_12, + "$defs": { + "SimpleModel": { + "title": "SimpleModel", + "type": "object", + "required": ["name", "value"], + "properties": { + "name": {"type": "string"}, + "value": {"type": "integer"}, + }, + "additionalProperties": JSONSchemaOptItem(input=True), + }, + }, + }, + ) + + +def test_generate_json_schema_custom_ref_prefix(): + schema = generate_json_schema( + Retort(), SimpleModel, Direction.INPUT, + local_ref_prefix="#/components/schemas/", + ) + + assert schema["$ref"] == "#/components/schemas/SimpleModel" + + +class _PrefixRefGenerator(RefGenerator): + def generate_ref(self, json_schema, loc_stack): + return "Custom" + BuiltinRefGenerator().generate_ref(json_schema, loc_stack) + + +@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=resolver, + occupied_refs=occupied_refs, + ) + + assert schema["$ref"] == f"#/$defs/{expected_ref}" + + +@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 +class Tag: + label: str + + +@dataclass +class Article: + title: str + tag: Tag + + +@dataclass +class Post: + body: str + tag: Tag + + +def test_generate_schemas_namespace_deduplicates_shared_types(): + retort = Retort() + + defs, schemas = generate_json_schemas_namespace([ + (retort, Direction.INPUT, Article), + (retort, Direction.INPUT, Post), + ]) + + assert len(schemas) == 2 + assert defs.keys() == {"Article", "Post", "Tag"} 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..8be32016 --- /dev/null +++ b/tests/integration/morphing/test_json_schema_overrides.py @@ -0,0 +1,286 @@ +from dataclasses import dataclass + +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 +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 +from adaptix._internal.morphing.json_schema.schema_model import JSONSchemaType + + +@dataclass +class Product: + name: str + price: float + + +_PRODUCT_DATA = {"name": "test", "price": 1.5} +_PRODUCT_LOADED = Product(name="test", price=1.5) + + +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), + }, + }, + } + + +_DEFAULT_SCHEMA = _product_schema({"type": "string"}, {"type": "number"}) + + +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(): + # 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 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()), + ]) + + 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(): + retort = Retort(recipe=[ + dumper(str, lambda x: x, json_schema=EraseJSONSchema()), + ]) + + 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(): + 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 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(): + 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"}, + ), + ) + + +def test_json_schema_inline_embeds_directly(): + retort = Retort(recipe=[json_schema(str, inline=True)]) + + 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)]) + + 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(): + 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/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..5620748e --- /dev/null +++ b/tests/unit/morphing/json_schema/test_patch.py @@ -0,0 +1,163 @@ +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_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_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..72b3cc15 --- /dev/null +++ b/tests/unit/morphing/json_schema/test_ref_generator.py @@ -0,0 +1,70 @@ +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..6d677fdf --- /dev/null +++ b/tests/unit/morphing/json_schema/test_schema_tools.py @@ -0,0 +1,173 @@ +import pytest + +from adaptix import Omitted +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, +) + + +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(): + bool_schema = True + result = list(traverse_json_schema(bool_schema)) + + 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] + + +@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_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) 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, ),