Dirk Riemann
703da9035e
Adds RequestIX.use_case_inline so callers can define ad-hoc extraction schemas in the request itself, bypassing the backend registry. The pipeline builds a fresh (Request, Response) Pydantic class pair per call via ix.use_cases.inline.build_use_case_classes; structural errors (dup field, bad identifier, choices-on-non-str, empty fields) raise IX_001_001 to match the registry-miss path. Inline wins when both use_case and use_case_inline are set. Existing REST callers see no behavioural change. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
313 lines
11 KiB
Python
313 lines
11 KiB
Python
"""Tests for :mod:`ix.use_cases.inline` — dynamic Pydantic class builder.
|
|
|
|
The builder takes an :class:`InlineUseCase` (carried on :class:`RequestIX` as
|
|
``use_case_inline``) and produces a fresh ``(RequestClass, ResponseClass)``
|
|
pair that the pipeline can consume in place of a registered use case.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from pydantic import BaseModel, ValidationError
|
|
|
|
from ix.contracts.request import InlineUseCase, UseCaseFieldDef
|
|
from ix.errors import IXErrorCode, IXException
|
|
from ix.use_cases.inline import build_use_case_classes
|
|
|
|
|
|
class TestUseCaseFieldDef:
|
|
def test_minimal(self) -> None:
|
|
fd = UseCaseFieldDef(name="foo", type="str")
|
|
assert fd.name == "foo"
|
|
assert fd.type == "str"
|
|
assert fd.required is False
|
|
assert fd.description is None
|
|
assert fd.choices is None
|
|
|
|
def test_extra_forbidden(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
UseCaseFieldDef.model_validate(
|
|
{"name": "foo", "type": "str", "bogus": 1}
|
|
)
|
|
|
|
def test_invalid_type_rejected(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
UseCaseFieldDef.model_validate({"name": "foo", "type": "list"})
|
|
|
|
|
|
class TestInlineUseCaseRoundtrip:
|
|
def test_json_roundtrip(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="Vendor Total",
|
|
system_prompt="Extract invoice total and vendor.",
|
|
default_model="qwen3:14b",
|
|
fields=[
|
|
UseCaseFieldDef(name="vendor", type="str", required=True),
|
|
UseCaseFieldDef(
|
|
name="total",
|
|
type="decimal",
|
|
required=True,
|
|
description="total amount due",
|
|
),
|
|
UseCaseFieldDef(
|
|
name="currency",
|
|
type="str",
|
|
choices=["USD", "EUR", "CHF"],
|
|
),
|
|
],
|
|
)
|
|
dumped = iuc.model_dump_json()
|
|
round = InlineUseCase.model_validate_json(dumped)
|
|
assert round == iuc
|
|
# JSON is well-formed
|
|
json.loads(dumped)
|
|
|
|
def test_extra_forbidden(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
InlineUseCase.model_validate(
|
|
{
|
|
"use_case_name": "X",
|
|
"system_prompt": "p",
|
|
"fields": [],
|
|
"bogus": 1,
|
|
}
|
|
)
|
|
|
|
|
|
class TestBuildBasicTypes:
|
|
@pytest.mark.parametrize(
|
|
"type_name, sample_value, bad_value",
|
|
[
|
|
("str", "hello", 123),
|
|
("int", 42, "nope"),
|
|
("float", 3.14, "nope"),
|
|
("bool", True, "nope"),
|
|
],
|
|
)
|
|
def test_simple_type(
|
|
self, type_name: str, sample_value: object, bad_value: object
|
|
) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[UseCaseFieldDef(name="field", type=type_name, required=True)],
|
|
)
|
|
_req_cls, resp_cls = build_use_case_classes(iuc)
|
|
instance = resp_cls(field=sample_value)
|
|
assert instance.field == sample_value
|
|
with pytest.raises(ValidationError):
|
|
resp_cls(field=bad_value)
|
|
|
|
def test_decimal_type(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[UseCaseFieldDef(name="amount", type="decimal", required=True)],
|
|
)
|
|
_req_cls, resp_cls = build_use_case_classes(iuc)
|
|
instance = resp_cls(amount="12.34")
|
|
assert isinstance(instance.amount, Decimal)
|
|
assert instance.amount == Decimal("12.34")
|
|
|
|
def test_date_type(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[UseCaseFieldDef(name="d", type="date", required=True)],
|
|
)
|
|
_req_cls, resp_cls = build_use_case_classes(iuc)
|
|
instance = resp_cls(d="2026-04-18")
|
|
assert instance.d == date(2026, 4, 18)
|
|
|
|
def test_datetime_type(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[UseCaseFieldDef(name="ts", type="datetime", required=True)],
|
|
)
|
|
_req_cls, resp_cls = build_use_case_classes(iuc)
|
|
instance = resp_cls(ts="2026-04-18T10:00:00")
|
|
assert isinstance(instance.ts, datetime)
|
|
|
|
|
|
class TestOptionalVsRequired:
|
|
def test_required_field_cannot_be_missing(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[UseCaseFieldDef(name="must", type="str", required=True)],
|
|
)
|
|
_req_cls, resp_cls = build_use_case_classes(iuc)
|
|
with pytest.raises(ValidationError):
|
|
resp_cls()
|
|
|
|
def test_optional_field_defaults_to_none(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[UseCaseFieldDef(name="maybe", type="str", required=False)],
|
|
)
|
|
_req_cls, resp_cls = build_use_case_classes(iuc)
|
|
instance = resp_cls()
|
|
assert instance.maybe is None
|
|
|
|
def test_optional_field_schema_allows_null(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[UseCaseFieldDef(name="maybe", type="str", required=False)],
|
|
)
|
|
_req_cls, resp_cls = build_use_case_classes(iuc)
|
|
schema = resp_cls.model_json_schema()
|
|
# "maybe" accepts string or null
|
|
prop = schema["properties"]["maybe"]
|
|
# Pydantic may express Optional as anyOf [str, null] or a type list.
|
|
# Either is fine — just assert null is allowed somewhere.
|
|
dumped = json.dumps(prop)
|
|
assert "null" in dumped
|
|
|
|
|
|
class TestChoices:
|
|
def test_choices_for_str_produces_literal(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[
|
|
UseCaseFieldDef(
|
|
name="kind",
|
|
type="str",
|
|
required=True,
|
|
choices=["a", "b", "c"],
|
|
)
|
|
],
|
|
)
|
|
_req_cls, resp_cls = build_use_case_classes(iuc)
|
|
inst = resp_cls(kind="a")
|
|
assert inst.kind == "a"
|
|
with pytest.raises(ValidationError):
|
|
resp_cls(kind="nope")
|
|
schema = resp_cls.model_json_schema()
|
|
# enum or const wind up in a referenced definition; walk the schema
|
|
dumped = json.dumps(schema)
|
|
assert '"a"' in dumped and '"b"' in dumped and '"c"' in dumped
|
|
|
|
def test_choices_for_non_str_raises_ix_001_001(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[
|
|
UseCaseFieldDef(
|
|
name="kind",
|
|
type="int",
|
|
required=True,
|
|
choices=["1", "2"],
|
|
)
|
|
],
|
|
)
|
|
with pytest.raises(IXException) as exc:
|
|
build_use_case_classes(iuc)
|
|
assert exc.value.code is IXErrorCode.IX_001_001
|
|
|
|
def test_empty_choices_list_ignored(self) -> None:
|
|
# An explicitly empty list is as-if choices were unset; builder must
|
|
# not break. If the caller sent choices=[] we treat the field as
|
|
# plain str.
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[
|
|
UseCaseFieldDef(
|
|
name="kind", type="str", required=True, choices=[]
|
|
)
|
|
],
|
|
)
|
|
_req_cls, resp_cls = build_use_case_classes(iuc)
|
|
inst = resp_cls(kind="anything")
|
|
assert inst.kind == "anything"
|
|
|
|
|
|
class TestValidation:
|
|
def test_duplicate_field_names_raise(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[
|
|
UseCaseFieldDef(name="foo", type="str"),
|
|
UseCaseFieldDef(name="foo", type="int"),
|
|
],
|
|
)
|
|
with pytest.raises(IXException) as exc:
|
|
build_use_case_classes(iuc)
|
|
assert exc.value.code is IXErrorCode.IX_001_001
|
|
|
|
def test_invalid_field_name_raises(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[UseCaseFieldDef(name="123abc", type="str")],
|
|
)
|
|
with pytest.raises(IXException) as exc:
|
|
build_use_case_classes(iuc)
|
|
assert exc.value.code is IXErrorCode.IX_001_001
|
|
|
|
def test_empty_fields_list_raises(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X", system_prompt="p", fields=[]
|
|
)
|
|
with pytest.raises(IXException) as exc:
|
|
build_use_case_classes(iuc)
|
|
assert exc.value.code is IXErrorCode.IX_001_001
|
|
|
|
|
|
class TestResponseClassNaming:
|
|
def test_class_name_sanitised(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="Bank / Statement — header!",
|
|
system_prompt="p",
|
|
fields=[UseCaseFieldDef(name="x", type="str")],
|
|
)
|
|
_req_cls, resp_cls = build_use_case_classes(iuc)
|
|
assert resp_cls.__name__.startswith("Inline_")
|
|
# Only alphanumerics and underscores remain.
|
|
assert all(c.isalnum() or c == "_" for c in resp_cls.__name__)
|
|
|
|
def test_fresh_instances_per_call(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="X",
|
|
system_prompt="p",
|
|
fields=[UseCaseFieldDef(name="x", type="str")],
|
|
)
|
|
req1, resp1 = build_use_case_classes(iuc)
|
|
req2, resp2 = build_use_case_classes(iuc)
|
|
assert resp1 is not resp2
|
|
assert req1 is not req2
|
|
|
|
|
|
class TestRequestClassShape:
|
|
def test_request_class_exposes_prompt_and_default(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="My Case",
|
|
system_prompt="Follow directions.",
|
|
default_model="qwen3:14b",
|
|
fields=[UseCaseFieldDef(name="x", type="str")],
|
|
)
|
|
req_cls, _resp_cls = build_use_case_classes(iuc)
|
|
inst = req_cls()
|
|
assert inst.use_case_name == "My Case"
|
|
assert inst.system_prompt == "Follow directions."
|
|
assert inst.default_model == "qwen3:14b"
|
|
assert issubclass(req_cls, BaseModel)
|
|
|
|
def test_default_model_none_when_unset(self) -> None:
|
|
iuc = InlineUseCase(
|
|
use_case_name="My Case",
|
|
system_prompt="Follow directions.",
|
|
fields=[UseCaseFieldDef(name="x", type="str")],
|
|
)
|
|
req_cls, _resp_cls = build_use_case_classes(iuc)
|
|
inst = req_cls()
|
|
assert inst.default_model is None
|