litellm/tests/unit/interactions/test_openapi_compliance.py

"""
OpenAPI compliance tests for Google Interactions API.

Validates that our SDK requests/responses match the OpenAPI spec snapshot in
interactions.openapi.json, taken from https://ai.google.dev/static/api/interactions.openapi.json
(litellm/types/interactions/README.md says how to refresh it). The scheduled
interactions-spec-canary workflow runs the same tests against the live spec by passing
--interactions-spec-url, so a change Google makes shows up there, not in every PR.

Run with: pytest tests/unit/interactions/test_openapi_compliance.py -v
"""

import json
import re
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Any, Dict, Final, get_args
from unittest.mock import MagicMock, patch

import httpx
import pytest
from openapi_core import OpenAPI
from pydantic import BaseModel, TypeAdapter

OPENAPI_SPEC_URL = "https://ai.google.dev/static/api/interactions.openapi.json"
OPENAPI_SPEC_SNAPSHOT: Final = Path(__file__).with_name("interactions.openapi.json")
SPEC_URL_OPTION: Final = TypeAdapter(str | None)


@dataclass(frozen=True, slots=True)
class LoadedSpec:
    source: str
    spec: Dict[str, Any]


def _status_enum_of(model: type[BaseModel]) -> type[Enum]:
    annotation: Final = model.model_fields["status"].annotation
    candidates: Final = (annotation, *get_args(annotation))
    return next(candidate for candidate in candidates if isinstance(candidate, type) and issubclass(candidate, Enum))


def _requested_spec_url(request: pytest.FixtureRequest) -> str | None:
    return SPEC_URL_OPTION.validate_python(request.config.getoption("interactions_spec_url"))


def _load_openapi_spec(live_url: str | None) -> LoadedSpec:
    if live_url is None:
        return LoadedSpec(source=str(OPENAPI_SPEC_SNAPSHOT), spec=json.loads(OPENAPI_SPEC_SNAPSHOT.read_text()))
    with httpx.Client(transport=httpx.HTTPTransport(retries=3), timeout=30.0) as client:
        response: Final = client.get(live_url)
    response.raise_for_status()
    return LoadedSpec(source=str(response.url), spec=response.json())


def _model_create_request_schema(spec_dict: Dict[str, Any]) -> Dict[str, Any]:
    schemas = spec_dict["components"]["schemas"]
    create_path = next(path for path in spec_dict["paths"] if path.endswith("/interactions"))
    body_schema = spec_dict["paths"][create_path]["post"]["requestBody"]["content"]["application/json"]["schema"]
    variants = [schemas[option["$ref"].split("/")[-1]] for option in body_schema.get("oneOf", []) if "$ref" in option]
    return next(variant for variant in variants if "model" in variant.get("properties", {}))


def _interaction_resource_path(spec_dict: Dict[str, Any], method: str) -> str | None:
    return next(
        (
            path
            for path, methods in spec_dict["paths"].items()
            if re.search(r"/interactions/\{[^}]+\}$", path) and method in methods
        ),
        None,
    )


def _declared_type_value(variant_schema: Dict[str, Any]) -> Any:
    """The single `type` value a union variant pins, whether spelled as a const or a 1-item enum."""
    type_property = variant_schema.get("properties", {}).get("type", {})
    enum_values = type_property.get("enum") or []
    return type_property.get("const") or (enum_values[0] if len(enum_values) == 1 else None)


@pytest.fixture(scope="module")
def loaded_spec(request: pytest.FixtureRequest) -> LoadedSpec:
    return _load_openapi_spec(_requested_spec_url(request))


@pytest.fixture(scope="module")
def spec_dict(loaded_spec: LoadedSpec) -> Dict[str, Any]:
    return loaded_spec.spec


@pytest.fixture(scope="module")
def openapi_spec(spec_dict: Dict[str, Any]) -> OpenAPI:
    """Load the OpenAPI spec as an OpenAPI object."""
    return OpenAPI.from_dict(spec_dict)


def test_spec_comes_from_the_requested_source(loaded_spec: LoadedSpec, request: pytest.FixtureRequest) -> None:
    assert loaded_spec.source == (_requested_spec_url(request) or str(OPENAPI_SPEC_SNAPSHOT))


def test_interaction_status_export_carries_the_interaction_models_statuses() -> None:
    from litellm.types.interactions import InteractionStatus, generated

    exported: Final = frozenset(member.value for member in InteractionStatus)
    modelled: Final = frozenset(member.value for member in _status_enum_of(generated.Interaction))
    assert exported == modelled, sorted(exported ^ modelled)


class TestRequestCompliance:
    """Tests that our request bodies match the OpenAPI spec."""

    def test_create_model_interaction_request_schema(self, spec_dict):
        schema = _model_create_request_schema(spec_dict)

        assert "model" in schema["required"]
        assert "input" in schema["properties"]

        # Check our supported optional fields exist in spec
        our_optional_fields = [
            "tools",
            "system_instruction",
            "generation_config",
            "stream",
            "store",
            "background",
            "response_modalities",
            "response_format",
            "response_mime_type",
            "previous_interaction_id",
        ]

        spec_properties = schema["properties"]
        for field in our_optional_fields:
            assert field in spec_properties, f"Field '{field}' not in OpenAPI spec"
            print(f"✓ Field '{field}' exists in spec")

    def test_input_types_match_spec(self, spec_dict):
        """Verify input field supports string, Content, Content[], Turn[]."""
        schema = _model_create_request_schema(spec_dict)
        input_schema = schema["properties"]["input"]

        # The input property may be inline oneOf or a $ref to InteractionsInput
        if "$ref" in input_schema:
            ref_name = input_schema["$ref"].split("/")[-1]
            input_schema = spec_dict["components"]["schemas"][ref_name]

        # Should be oneOf with multiple types
        assert "oneOf" in input_schema

        input_types = []
        for option in input_schema["oneOf"]:
            if option.get("type") == "string":
                input_types.append("string")
            elif option.get("type") == "array":
                input_types.append("array")
            elif "$ref" in option:
                input_types.append(option["$ref"])

        print(f"Input supports types: {input_types}")
        assert "string" in input_types, "Input should support string"
        assert "array" in input_types, "Input should support array"

    def test_content_variants_are_identified_by_their_type_field(self, spec_dict):
        """Verify a Content part can be told apart by its `type`, however the spec spells that.

        Our transformation reads `type` off each content part to route it, so what has to hold is
        that every variant of the union pins a distinct `type` value and that text is one of them.
        A spec may express that with an OpenAPI `discriminator` on the union or with a `const` on
        each member's own `type`; both are equivalent for us, so accepting only the first makes
        this test fail on a stylistic change upstream that costs us nothing.
        """
        content_schema = spec_dict["components"]["schemas"]["Content"]

        discriminator = content_schema.get("discriminator")
        if discriminator is not None:
            assert (
                discriminator.get("propertyName") == "type"
            ), f"Content is discriminated on {discriminator.get('propertyName')!r}, not 'type'"

        variant_names = [
            option["$ref"].split("/")[-1]
            for option in content_schema.get("oneOf", [])
            if "$ref" in option
        ]
        assert variant_names, f"Content is not a union of named variants: {content_schema}"

        mapping = (discriminator or {}).get("mapping") or {}
        type_values = {
            variant: mapping_value
            for mapping_value, ref in mapping.items()
            for variant in [ref.split("/")[-1]]
        } or {
            variant: _declared_type_value(spec_dict["components"]["schemas"].get(variant, {}))
            for variant in variant_names
        }

        assert set(type_values) == set(variant_names) and all(type_values.values()), (
            f"every Content variant needs a discoverable type value, "
            f"got {type_values} for variants {sorted(variant_names)}"
        )
        assert len(set(type_values.values())) == len(type_values), (
            f"Content variants must pin DISTINCT type values, got {type_values}"
        )
        assert type_values.get("TextContent") == "text", (
            f"TextContent must be reachable as type 'text', got {type_values}"
        )
        print(f"Content variants by type: {type_values}")

    def test_text_content_schema(self, spec_dict):
        """Verify TextContent schema."""
        text_schema = spec_dict["components"]["schemas"]["TextContent"]

        assert "type" in text_schema["required"]
        assert "text" in text_schema["properties"]
        assert text_schema["properties"]["type"].get("const") == "text"
        print("✓ TextContent schema is correct")

    def test_step_schema(self, spec_dict):
        """Verify step-based multi-turn input.

        Google replaced the role-carrying `Turn` schema with typed steps
        (spec update of Aug 13, 2026): conversation history is now a `Step[]`
        where `UserInputStep`/`ModelOutputStep` pin `type` values that our
        transformations read to recover the role. Assert exactly what our code
        depends on: `InteractionsInput` accepts a Step array, both step kinds
        are part of the `Step` union, each pins its `type` const, and each
        carries a `Content[]` content field.
        """
        input_schema = spec_dict["components"]["schemas"]["InteractionsInput"]
        step_array_items = [
            option["items"]["$ref"].split("/")[-1]
            for option in input_schema["oneOf"]
            if option.get("type") == "array" and "$ref" in option.get("items", {})
        ]
        assert "Step" in step_array_items, f"InteractionsInput should accept Step[], got arrays of {step_array_items}"

        step_variants = {
            option["$ref"].split("/")[-1]
            for option in spec_dict["components"]["schemas"]["Step"]["oneOf"]
            if "$ref" in option
        }
        assert {"UserInputStep", "ModelOutputStep"} <= step_variants, f"Step union is missing role steps: {step_variants}"

        for step_name, type_value in [("UserInputStep", "user_input"), ("ModelOutputStep", "model_output")]:
            step_schema = spec_dict["components"]["schemas"][step_name]
            assert step_schema["properties"]["type"].get("const") == type_value
            assert "type" in step_schema["required"]
            content_items = step_schema["properties"]["content"]["items"]
            assert content_items["$ref"].split("/")[-1] == "Content"
            print(f"✓ {step_name} pins type '{type_value}' with Content[] content")


class TestResponseCompliance:
    """Tests that our response types match the OpenAPI spec."""

    def test_interaction_response_fields(self, spec_dict):
        """Verify our InteractionsAPIResponse has correct fields."""
        # The response is the dedicated `Interaction` schema. Google moved the
        # output-only fields (notably the `steps` array, formerly `outputs`)
        # off `CreateModelInteractionParams` and onto `Interaction`; the request
        # schema no longer carries `steps`. Google later moved `role` off
        # `Interaction` onto the per-turn `Turn` schema (asserted in
        # test_turn_schema), so it is no longer a top-level output field here.
        # Keep this aligned with the live spec.
        schema = spec_dict["components"]["schemas"]["Interaction"]

        # Output fields (readOnly). `role` was removed from the `Interaction`
        # schema by Google; it now lives only on `Turn`.
        output_fields = [
            "id",
            "status",
            "created",
            "updated",
            "steps",
            "usage",
        ]

        for field in output_fields:
            assert field in schema["properties"], f"Output field '{field}' not in spec"
            print(f"✓ Output field '{field}' exists in spec")

    @pytest.mark.parametrize(
        "schema_name",
        [
            "Interaction",
            "InteractionStatusUpdate",
            "CreateModelInteractionParams",
            "CreateAgentInteractionParams",
            "UrlContextResult",
        ],
    )
    def test_status_enum_values(self, spec_dict, schema_name):
        from litellm.types.interactions import generated

        schema = spec_dict["components"]["schemas"][schema_name]
        spec_statuses: Final = frozenset(schema["properties"]["status"]["enum"])
        parsed_statuses: Final = frozenset(member.value for member in _status_enum_of(getattr(generated, schema_name)))
        assert spec_statuses
        assert spec_statuses <= parsed_statuses, sorted(spec_statuses - parsed_statuses)

    def test_usage_schema(self, spec_dict):
        """Verify Usage schema fields."""
        usage_schema = spec_dict["components"]["schemas"]["Usage"]

        # Key usage fields
        expected_fields = ["total_input_tokens", "total_output_tokens", "total_tokens"]

        for field in expected_fields:
            assert (
                field in usage_schema["properties"]
            ), f"Usage field '{field}' not in spec"
            print(f"✓ Usage field '{field}' exists")


class TestToolsCompliance:
    """Tests that our tool types match the OpenAPI spec."""

    def test_tool_schema(self, spec_dict):
        """Verify Tool schema."""
        tool_schema = spec_dict["components"]["schemas"]["Tool"]

        # Tool should be oneOf multiple tool types
        assert "oneOf" in tool_schema or "properties" in tool_schema
        print(f"✓ Tool schema found")

    def test_function_declaration_schema(self, spec_dict):
        """Verify FunctionDeclaration schema for function tools."""
        if "FunctionDeclaration" in spec_dict["components"]["schemas"]:
            func_schema = spec_dict["components"]["schemas"]["FunctionDeclaration"]
            assert "name" in func_schema.get(
                "properties", {}
            ) or "name" in func_schema.get("required", [])
            print("✓ FunctionDeclaration schema found")
        else:
            print("⚠ FunctionDeclaration schema not found (may be nested)")


class TestEndpointCompliance:
    """Tests that our endpoints match the OpenAPI spec."""

    def test_create_endpoint_exists(self, spec_dict):
        """Verify POST /interactions endpoint exists."""
        paths = spec_dict["paths"]

        # Find the create interactions endpoint
        create_path = None
        for path, methods in paths.items():
            if "interactions" in path and "post" in methods:
                create_path = path
                break

        assert create_path is not None, "POST /interactions endpoint not found"
        print(f"✓ Create endpoint: POST {create_path}")

    def test_get_endpoint_exists(self, spec_dict):
        """Verify GET /interactions/{id} endpoint exists."""
        get_path = _interaction_resource_path(spec_dict, "get")

        assert get_path is not None, "GET /interactions/{id} endpoint not found"
        print(f"✓ Get endpoint: GET {get_path}")

    def test_delete_endpoint_exists(self, spec_dict):
        """Verify DELETE /interactions/{id} endpoint exists."""
        delete_path = _interaction_resource_path(spec_dict, "delete")

        assert delete_path is not None, "DELETE /interactions/{id} endpoint not found"
        print(f"✓ Delete endpoint: DELETE {delete_path}")


if __name__ == "__main__":
    # Quick manual test
    import httpx

    print("Loading OpenAPI spec...")
    response = httpx.get(OPENAPI_SPEC_URL)
    spec = response.json()

    print(f"\nSpec version: {spec.get('openapi')}")
    print(f"API title: {spec.get('info', {}).get('title')}")
    print(f"\nEndpoints:")
    for path, methods in spec.get("paths", {}).items():
        for method in methods:
            if method in ["get", "post", "delete", "put", "patch"]:
                print(f"  {method.upper()} {path}")

    print(
        f"\nSchemas: {list(spec.get('components', {}).get('schemas', {}).keys())[:10]}..."
    )