| """Schema -> fill plan contract (SPEC.md §3-§4).""" |
|
|
| from __future__ import annotations |
|
|
| import copy |
| import json |
|
|
| import pytest |
| from deg_cases import ARRAYS, INVOICE, NESTED, PRODUCT_LISTING, SUPPORT_TICKET |
|
|
| from deg.schema import ITEM, SchemaError, compile_schema, pointer |
|
|
|
|
| def modes(plan): |
| return [(f.pointer, f.mode) for f in plan.fields] |
|
|
|
|
| def obj(**props): |
| return {"type": "object", "properties": props, "required": list(props)} |
|
|
|
|
| def errors_of(schema): |
| with pytest.raises(SchemaError) as info: |
| compile_schema(schema) |
| return info.value.errors |
|
|
|
|
| |
|
|
|
|
| def test_support_ticket_plan(): |
| plan = compile_schema(SUPPORT_TICKET) |
| assert modes(plan) == [ |
| ("/department", "decide.choice"), |
| ("/urgent", "decide.noul"), |
| ("/anger", "decide.score"), |
| ("/customer_email", "extract.span"), |
| ("/order_id", "extract.span"), |
| ("/reply", "generate"), |
| ] |
| dept, _, anger, email, order, reply = plan.fields |
| assert dept.options == ("billing", "technical", "sales") |
| assert dept.description == "Which team should handle this?" |
| assert anger.levels == (0, 1, 2, 3, 4) |
| assert (email.format, email.nullable, email.on_null) == ("email", True, "null") |
| assert (order.on_null, order.constraints) == ("omit", {"pattern": "^[A-Z]{2}-\\d+$"}) |
| assert reply.constraints == {"maxLength": 400} and reply.on_null == "forbid" |
| assert [f.pointer for f in plan.generate_fields] == ["/reply"] |
| assert len(plan.parallel_fields) == 5 |
|
|
|
|
| def test_invoice_plan_refs_oneof_and_arrays(): |
| plan = compile_schema(INVOICE) |
| assert modes(plan) == [ |
| ("/invoice_number", "extract.span"), |
| ("/issued", "extract.date"), |
| ("/due", "extract.date"), |
| ("/vendor/name", "extract.span"), |
| ("/vendor/website", "extract.span"), |
| ("/currency", "decide.choice"), |
| ("/line_items/*/description", "extract.span"), |
| ("/line_items/*/quantity", "extract.number"), |
| ("/line_items/*/amount", "extract.number"), |
| ("/total", "extract.number"), |
| ] |
| currency = plan.field("/currency") |
| assert currency.options == ("USD", "EUR", "GBP") |
| assert currency.option_descriptions == ("US dollars", "Euros", None) |
| amount = plan.field("/line_items/*/amount") |
| assert amount.path == ("line_items", ITEM, "amount") |
| assert amount.description == "Amount in invoice currency" |
| assert amount.constraints == {"minimum": 0} |
| assert plan.field("/line_items/*/quantity").on_null == "omit" |
| assert plan.field("/due").on_null == "null" |
| assert plan.field("/vendor/website").format == "uri" |
| (items,) = plan.arrays |
| assert (items.pointer, items.min_items, items.max_items) == ("/line_items", 1, 50) |
| assert plan.generate_fields == () |
|
|
|
|
| def test_product_listing_generates_only_marked_fields(): |
| plan = compile_schema(PRODUCT_LISTING) |
| assert modes(plan) == [ |
| ("/title", "extract.span"), |
| ("/category", "decide.choice"), |
| ("/condition", "decide.choice"), |
| ("/price", "extract.number"), |
| ("/rating", "extract.number"), |
| ("/in_stock", "decide.noul"), |
| ("/description", "generate"), |
| ("/tags/*", "generate"), |
| ] |
| rating = plan.field("/rating") |
| assert rating.levels is None and rating.constraints == {"minimum": 0, "maximum": 5} |
| assert [f.pointer for f in plan.generate_fields] == ["/description", "/tags/*"] |
| assert plan.field("/tags/*").constraints == {"maxLength": 24} |
| assert plan.arrays[0].max_items == 5 |
|
|
|
|
| def test_nested_objects(): |
| plan = compile_schema(NESTED) |
| assert modes(plan) == [ |
| ("/patient/name", "extract.span"), |
| ("/patient/birth", "extract.date"), |
| ("/patient/contact/email", "extract.span"), |
| ("/patient/contact/id", "extract.span"), |
| ("/visit/at", "extract.date"), |
| ("/visit/follow_up", "decide.choice"), |
| ("/visit/summary", "generate"), |
| ] |
| assert plan.field("/visit/follow_up").options == (True, False, None) |
| assert plan.field("/visit/at").format == "date-time" |
| assert plan.field("/patient/contact/id").format == "uuid" |
|
|
|
|
| def test_arrays_per_item(): |
| plan = compile_schema(ARRAYS) |
| assert modes(plan) == [ |
| ("/labels/*", "decide.choice"), |
| ("/people/*/name", "extract.span"), |
| ("/people/*/aliases/*", "extract.span"), |
| ("/matrix/*/*", "extract.number"), |
| ] |
| assert [a.pointer for a in plan.arrays] == [ |
| "/labels", "/people", "/people/*/aliases", "/matrix", "/matrix/*", |
| ] |
| assert plan.arrays[0].unique_items is True |
|
|
|
|
| def test_type_to_mode_table(): |
| plan = compile_schema(obj( |
| e={"enum": [1, 2, 3]}, |
| mixed={"enum": ["a", 1, None]}, |
| b={"type": "boolean"}, |
| i_small={"type": "integer", "minimum": 1, "maximum": 10}, |
| i_step={"type": "integer", "minimum": 0, "maximum": 100, "multipleOf": 25}, |
| i_excl={"type": "integer", "exclusiveMinimum": 0, "exclusiveMaximum": 4}, |
| i_wide={"type": "integer", "minimum": 0, "maximum": 150}, |
| i_open={"type": "integer"}, |
| n_unit={"type": "number", "minimum": 0, "maximum": 1}, |
| n_excl={"type": "number", "exclusiveMinimum": 0, "exclusiveMaximum": 1}, |
| i_12={"type": "integer", "minimum": 0, "maximum": 11}, |
| n_open={"type": "number"}, |
| s={"type": "string"}, |
| d={"type": "string", "format": "date"}, |
| dt={"type": "string", "format": "date-time"}, |
| u={"type": "string", "format": "uuid"}, |
| g={"type": "string", "x-generate": True}, |
| g_false={"type": "string", "x-generate": False}, |
| )) |
| got = {f.pointer: (f.mode, f.levels, f.options) for f in plan.fields} |
| assert got["/e"] == ("decide.choice", None, (1, 2, 3)) |
| assert got["/mixed"] == ("decide.choice", None, ("a", 1, None)) |
| assert got["/b"][0] == "decide.noul" |
| assert got["/i_small"] == ("decide.score", tuple(range(1, 11)), None) |
| assert got["/i_step"] == ("decide.score", (0, 25, 50, 75, 100), None) |
| assert got["/i_excl"] == ("decide.score", (1, 2, 3), None) |
| assert got["/i_wide"][0] == "extract.number" |
| assert got["/i_open"][0] == "extract.number" |
| assert got["/n_unit"][0] == "extract.number" |
| assert got["/n_excl"][0] == "extract.number" |
| assert got["/i_12"][0] == "extract.number" |
| assert got["/n_open"][0] == "extract.number" |
| assert got["/s"][0] == "extract.span" |
| assert got["/d"][0] == got["/dt"][0] == "extract.date" |
| assert got["/u"][0] == "extract.span" |
| assert got["/g"][0] == "generate" |
| assert got["/g_false"][0] == "extract.span" |
| assert plan.field("/mixed").nullable is True |
|
|
|
|
| def test_nullable_forms(): |
| plan = compile_schema(obj( |
| a={"type": ["string", "null"]}, |
| b={"anyOf": [{"type": "string"}, {"type": "null"}], "description": "B"}, |
| c={"oneOf": [{"type": "null"}, {"type": "integer", "minimum": 0, "maximum": 3}]}, |
| d={"anyOf": [{"enum": ["x", "y"]}, {"type": "null"}]}, |
| e={"type": ["string", "null"], "enum": ["x", "y", None]}, |
| )) |
| a, b, c, d, e = plan.fields |
| assert a.nullable and b.nullable and b.description == "B" |
| assert (c.mode, c.levels, c.null_outcome) == ("decide.score", (0, 1, 2, 3), True) |
| assert d.options == ("x", "y", None) and d.nullable |
| assert e.options == ("x", "y", None) |
|
|
|
|
| def test_optional_fields_and_on_null(): |
| plan = compile_schema({ |
| "type": "object", |
| "properties": {"req": {"type": "string"}, "opt": {"type": "string"}, |
| "decided": {"type": "boolean"}}, |
| "required": ["req"], |
| }) |
| assert [f.on_null for f in plan.fields] == ["forbid", "omit", "omit"] |
| assert [f.required for f in plan.fields] == [True, False, False] |
|
|
|
|
| def test_pointer_escaping(): |
| assert pointer(("a/b", "c~d", ITEM, "*")) == "/a~1b/c~0d/*/~2" |
| plan = compile_schema(obj(**{"a/b": {"type": "string"}})) |
| assert plan.fields[0].pointer == "/a~1b" |
|
|
|
|
| def test_local_ref_with_annotation_override(): |
| schema = { |
| "type": "object", |
| "definitions": {"name": {"type": "string", "description": "orig"}}, |
| "properties": {"n": {"$ref": "#/definitions/name", "description": "override"}}, |
| } |
| (field,) = compile_schema(schema).fields |
| assert (field.mode, field.description) == ("extract.span", "override") |
|
|
|
|
| |
|
|
|
|
| @pytest.mark.parametrize( |
| "bad", |
| [ |
| {"enum": ["a", "b"], "x-generate": True}, |
| {"type": "string", "enum": ["a", "b"], "x-generate": True}, |
| {"type": "boolean", "x-generate": True}, |
| {"type": "number", "x-generate": True}, |
| {"type": "integer", "minimum": 0, "maximum": 3, "x-generate": True}, |
| {"type": "object", "properties": {}, "x-generate": True}, |
| {"type": "array", "items": {"type": "integer"}, "x-generate": True}, |
| {"type": "array", "items": {"enum": ["a", "b"]}, "x-generate": True}, |
| ], |
| ) |
| def test_x_generate_only_on_string(bad): |
| (error,) = errors_of(obj(f=bad)) |
| assert "x-generate is allowed only on string or array of string" in error["msg"] |
|
|
|
|
| def test_x_generate_with_format_and_bad_value(): |
| (error,) = errors_of(obj(f={"type": "string", "format": "email", "x-generate": True})) |
| assert "format" in error["msg"] |
| (error,) = errors_of(obj(f={"type": "string", "x-generate": "yes"})) |
| assert error["msg"] == "x-generate must be a boolean" |
|
|
|
|
| @pytest.mark.parametrize( |
| ("bad", "keyword"), |
| [ |
| ({"allOf": [{"type": "string"}]}, "allOf"), |
| ({"type": "string", "not": {"maxLength": 2}}, "not"), |
| ({"type": "string", "if": {}, "then": {}}, "if"), |
| ({"const": "x"}, "const"), |
| ({"type": "object", "patternProperties": {"^a": {"type": "string"}}}, |
| "patternProperties"), |
| ({"type": "array", "prefixItems": [{"type": "string"}]}, "prefixItems"), |
| ({"type": "array", "items": {"type": "string"}, "contains": {"type": "string"}}, |
| "contains"), |
| ({"type": "object", "properties": {}, "minProperties": 1}, "minProperties"), |
| ({"$dynamicRef": "#x"}, "$dynamicRef"), |
| ], |
| ) |
| def test_unsupported_keywords(bad, keyword): |
| errors = errors_of(obj(f=bad)) |
| assert any(f"unsupported keyword {keyword!r}" in e["msg"] for e in errors), errors |
|
|
|
|
| @pytest.mark.parametrize( |
| ("bad", "fragment"), |
| [ |
| ({"anyOf": [{"type": "string"}, {"type": "integer"}]}, "anyOf is supported only"), |
| ({"oneOf": [{"type": "string"}, {"type": "object"}]}, "oneOf is supported only"), |
| ({"type": ["string", "integer"]}, "union types are not supported"), |
| ({"$ref": "https://example.com/s.json"}, "only local $ref"), |
| ({"$ref": "#/$defs/missing"}, "unresolvable $ref"), |
| ({"type": "object", "additionalProperties": {"type": "string"}}, "additionalProperties"), |
| ({"type": ["object", "null"], "properties": {}}, "nullable objects/arrays"), |
| ({"type": "array"}, "arrays need an items schema"), |
| ({"type": "string", "format": "hostname"}, "unsupported format"), |
| ({"type": "string", "x-foo": 1}, "unknown extension"), |
| ({"type": "string", "frobnicate": 1}, "unknown keyword"), |
| ({"type": "integer", "maxLength": 3}, "does not apply to type 'integer'"), |
| ({"enum": ["only"]}, "a single value has nothing to decide"), |
| ({"enum": ["a", "a"]}, "duplicate enum value"), |
| ({"type": "string", "enum": ["a", 1]}, "does not match type"), |
| ({"type": "integer", "minimum": 3, "maximum": 3}, "single value"), |
| ({"type": "number", "minimum": 2, "maximum": 1}, "admit no value"), |
| ({"type": "number", "minimum": 1, "exclusiveMaximum": 1}, "admit no value"), |
| ({"type": "string", "pattern": "("}, "invalid"), |
| ({}, "type is required"), |
| ], |
| ) |
| def test_validation_errors(bad, fragment): |
| errors = errors_of(obj(f=bad)) |
| assert any(fragment in e["msg"] for e in errors), errors |
| assert all(e["loc"][:3] == ["schema", "properties", "f"] for e in errors), errors |
|
|
|
|
| def test_draft07_tuple_items_rejected(): |
| schema = {"$schema": "http://json-schema.org/draft-07/schema#", **obj( |
| f={"type": "array", "items": [{"type": "string"}]})} |
| (error,) = errors_of(schema) |
| assert "tuple arrays" in error["msg"] |
|
|
|
|
| def test_recursive_ref_rejected(): |
| schema = { |
| "type": "object", |
| "$defs": {"node": {"type": "object", "properties": {"child": {"$ref": "#/$defs/node"}}}}, |
| "properties": {"root": {"$ref": "#/$defs/node"}}, |
| } |
| errors = errors_of(schema) |
| assert any("recursive $ref" in e["msg"] for e in errors) |
|
|
|
|
| def test_root_must_be_object_and_errors_are_collected(): |
| assert "type: object" in errors_of({"type": "string"})[0]["msg"] |
| errors = errors_of(obj(a={"type": "boolean", "x-generate": True}, |
| b={"allOf": [{"type": "string"}]}, |
| c={"type": "string", "format": "ipv4"})) |
| assert sorted(e["loc"][2] for e in errors) == ["a", "b", "c"] |
| with pytest.raises(SchemaError): |
| compile_schema("not a schema") |
| with pytest.raises(SchemaError): |
| compile_schema({"type": "object", "properties": {"a": {"type": 5}}}) |
|
|
|
|
| def test_required_must_name_properties(): |
| errors = errors_of({"type": "object", "properties": {}, "required": ["ghost"]}) |
| assert "not in properties" in errors[0]["msg"] |
|
|
|
|
| |
|
|
|
|
| def test_plan_is_deterministic_and_does_not_mutate_schema(): |
| for schema in (SUPPORT_TICKET, INVOICE, PRODUCT_LISTING, NESTED, ARRAYS): |
| before = copy.deepcopy(schema) |
| a, b = compile_schema(schema), compile_schema(copy.deepcopy(schema)) |
| assert a.to_dict() == b.to_dict() |
| assert json.dumps(a.to_dict(), sort_keys=True) == json.dumps(b.to_dict(), sort_keys=True) |
| assert schema == before |
|
|
|
|
| def _reordered(schema, order): |
| out = copy.deepcopy(schema) |
| out["properties"] = {k: schema["properties"][k] for k in order} |
| return out |
|
|
|
|
| def test_parallel_fields_are_order_independent(): |
| shuffled = _reordered(SUPPORT_TICKET, ["reply", "order_id", "anger", "urgent", |
| "customer_email", "department"]) |
| a, b = compile_schema(SUPPORT_TICKET), compile_schema(shuffled) |
| fa = {f.pointer: f.fingerprint for f in a.fields} |
| fb = {f.pointer: f.fingerprint for f in b.fields} |
| assert fa == fb |
| assert a.fingerprint == b.fingerprint |
|
|
|
|
| def test_generate_order_is_meaningful(): |
| schema = obj( |
| kind={"enum": ["a", "b"]}, |
| first={"type": "string", "x-generate": True}, |
| second={"type": "string", "x-generate": True}, |
| ) |
| swapped = _reordered(schema, ["second", "kind", "first"]) |
| a, b = compile_schema(schema), compile_schema(swapped) |
| assert [f.pointer for f in a.generate_fields] == ["/first", "/second"] |
| assert [f.pointer for f in b.generate_fields] == ["/second", "/first"] |
| assert a.fingerprint != b.fingerprint |
| moved_parallel = _reordered(schema, ["first", "kind", "second"]) |
| assert compile_schema(moved_parallel).fingerprint == a.fingerprint |
|
|
|
|
| def test_field_fingerprint_independent_of_composition(): |
| """A leaf's spec is the same alone, next to other fields, or merged into a bigger schema.""" |
| alone = compile_schema(obj(department=SUPPORT_TICKET["properties"]["department"])) |
| merged = compile_schema(obj( |
| department=SUPPORT_TICKET["properties"]["department"], |
| **{k: v for k, v in PRODUCT_LISTING["properties"].items()}, |
| )) |
| assert alone.fields[0].fingerprint == merged.field("/department").fingerprint |
| product = compile_schema(PRODUCT_LISTING) |
| for f in product.parallel_fields: |
| g = merged.field(f.pointer) |
| |
| assert g.fingerprint == f.fingerprint |
|
|
|
|
| def test_x_levels_and_x_max_tokens_extensions(): |
| ok = compile_schema({"type": "object", "properties": { |
| "s": {"type": "integer", "minimum": 1, "maximum": 3, "x-levels": {"1": "low", "3": "high"}}, |
| "g": {"type": "string", "x-generate": True, "x-max-tokens": 12}, |
| "tags": {"type": "array", "x-generate": True, "x-max-tokens": 5, |
| "items": {"type": "string"}}}}) |
| by = {f.pointer: f for f in ok.fields} |
| assert by["/s"].level_labels == ("low", None, "high") |
| assert by["/g"].max_new_tokens == 12 and by["/tags/*"].max_new_tokens == 5 |
| bad = [ |
| {"type": "string", "x-max-tokens": 4}, |
| {"type": "string", "x-generate": True, "x-max-tokens": 0}, |
| {"type": "string", "x-levels": {"1": "a"}}, |
| {"type": "integer", "minimum": 0, "maximum": 2, "x-levels": {"5": "a"}}, |
| {"type": "integer", "minimum": 0, "maximum": 2, "x-levels": {"1": ""}}, |
| {"type": "string", "x-other": 1}, |
| ] |
| for sub in bad: |
| with pytest.raises(SchemaError): |
| compile_schema({"type": "object", "properties": {"f": sub}}) |
|
|
