Blame for source/tests/test_schema.py

"""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


# ----------------------------------------------------------------------------- examples


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"  # from the $ref target
    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"),  # one-sided bound: a quantity, extracted
        ("/rating", "extract.number"),  # a number: always extracted; bounds only validate
        ("/in_stock", "decide.noul"),
        ("/description", "generate"),
        ("/tags/*", "generate"),  # x-generate on the array applies to its string items
    ]
    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"),  # nullable boolean -> 3-way 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"  # 12 admissible values: extracted
    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")


# ----------------------------------------------------------------------------- errors


@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}}})  # invalid JSON Schema


def test_required_must_name_properties():
    errors = errors_of({"type": "object", "properties": {}, "required": ["ghost"]})
    assert "not in properties" in errors[0]["msg"]


# ----------------------------------------------------------------------------- stability


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  # one generate field: order of the rest is moot


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)
        # required flags match because obj() requires everything, as PRODUCT_LISTING does
        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},  # not x-generate
        {"type": "string", "x-generate": True, "x-max-tokens": 0},
        {"type": "string", "x-levels": {"1": "a"}},  # not a score field
        {"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}})