567-labs--instructor/tests/coverage/test_openai_support_coverage.py

from __future__ import annotations

import builtins
import json
import runpy
from pathlib import Path
from typing import Any, cast

import httpx
import openai
import pytest
import requests
from openai.types.responses import Response
from pydantic import BaseModel, Field

from instructor.v2.core.client import AsyncInstructor, Instructor
from instructor.v2.core.errors import ClientError, ModeError
from instructor.v2.core.mode import Mode
from instructor.v2.core.multimodal import Audio, Image, PDF
from instructor.v2.core.providers import Provider
from instructor.v2.providers.openai.client import (
    async_map_chat_completion_to_response,
    from_openai,
    map_chat_completion_to_response,
)
from instructor.v2.providers.openai.multimodal import (
    audio_to_openai,
    image_to_openai,
    pdf_to_openai,
)
from instructor.v2.providers.openai.schema import generate_openai_schema
from instructor.v2.providers.openai.templating import process_message
from instructor.v2.providers.openrouter.client import from_openrouter
from instructor.v2.providers.perplexity.client import from_perplexity


def _response_payload(model: str) -> dict[str, Any]:
    return {
        "id": "resp_coverage",
        "object": "response",
        "created_at": 1,
        "model": model,
        "output": [],
        "parallel_tool_calls": False,
        "tool_choice": "auto",
        "tools": [],
    }


def test_openai_sync_response_mapping_uses_real_sdk_types() -> None:
    seen: list[dict[str, Any]] = []

    def handle(request: httpx.Request) -> httpx.Response:
        assert request.url.path.endswith("/responses")
        body = json.loads(request.content)
        seen.append(body)
        return httpx.Response(200, json=_response_payload(body["model"]))

    http_client = httpx.Client(transport=httpx.MockTransport(handle), trust_env=False)
    client = openai.OpenAI(
        api_key="test-key",
        base_url="https://openai.invalid/v1",
        http_client=http_client,
    )
    messages = [{"role": "user", "content": "hello from sync"}]

    response = map_chat_completion_to_response(
        messages, client, model="gpt-test-sync", metadata={"lane": "sync"}
    )

    assert isinstance(response, Response)
    assert response.id == "resp_coverage"
    assert seen == [
        {
            "input": messages,
            "model": "gpt-test-sync",
            "metadata": {"lane": "sync"},
        }
    ]
    client.close()


@pytest.mark.asyncio
async def test_openai_async_response_mapping_uses_real_sdk_types() -> None:
    seen: list[dict[str, Any]] = []

    async def handle(request: httpx.Request) -> httpx.Response:
        assert request.url.path.endswith("/responses")
        body = json.loads(request.content)
        seen.append(body)
        return httpx.Response(200, json=_response_payload(body["model"]))

    http_client = httpx.AsyncClient(
        transport=httpx.MockTransport(handle), trust_env=False
    )
    client = openai.AsyncOpenAI(
        api_key="test-key",
        base_url="https://openai.invalid/v1",
        http_client=http_client,
    )
    messages = [{"role": "user", "content": "hello from async"}]

    response = await async_map_chat_completion_to_response(
        messages, client, model="gpt-test-async", metadata={"lane": "async"}
    )

    assert isinstance(response, Response)
    assert response.model == "gpt-test-async"
    assert seen == [
        {
            "input": messages,
            "model": "gpt-test-async",
            "metadata": {"lane": "async"},
        }
    ]
    await client.close()


def test_openai_factory_rejects_an_unregistered_mode() -> None:
    client = openai.OpenAI(
        api_key="test-key",
        base_url="https://openai.invalid/v1",
        http_client=httpx.Client(trust_env=False),
    )

    with pytest.raises(ModeError) as error:
        from_openai(client, mode=Mode.GEMINI_TOOLS)

    assert error.value.mode == Mode.GEMINI_TOOLS.value
    assert error.value.provider == Provider.OPENAI.value
    assert Mode.TOOLS.value in error.value.valid_modes
    client.close()


def test_openai_factory_rejects_an_invalid_client() -> None:
    with pytest.raises(ClientError, match="Got: object") as error:
        from_openai(cast(openai.OpenAI, object()), mode=Mode.TOOLS)

    assert "OpenAI, AsyncOpenAI" in str(error.value)


@pytest.mark.asyncio
async def test_openai_compatible_wrappers_keep_provider_mode_and_client() -> None:
    sync_client = openai.OpenAI(
        api_key="test-key",
        base_url="https://openrouter.invalid/api/v1",
        http_client=httpx.Client(trust_env=False),
    )
    async_client = openai.AsyncOpenAI(
        api_key="test-key",
        base_url="https://perplexity.invalid",
        http_client=httpx.AsyncClient(trust_env=False),
    )

    openrouter = from_openrouter(
        sync_client, mode=Mode.JSON_SCHEMA, model="router-model"
    )
    perplexity = from_perplexity(async_client, model="perplexity-model")

    assert isinstance(openrouter, Instructor)
    assert openrouter.client is sync_client
    assert openrouter.provider is Provider.OPENROUTER
    assert openrouter.mode is Mode.JSON_SCHEMA
    assert isinstance(perplexity, AsyncInstructor)
    assert perplexity.client is async_client
    assert perplexity.provider is Provider.PERPLEXITY
    assert perplexity.mode is Mode.MD_JSON
    sync_client.close()
    await async_client.close()


@pytest.mark.asyncio
async def test_openai_response_wrappers_normalize_the_inbuilt_tools_mode() -> None:
    sync_client = openai.OpenAI(
        api_key="test-key",
        base_url="https://openai.invalid/v1",
        http_client=httpx.Client(trust_env=False),
    )
    async_client = openai.AsyncOpenAI(
        api_key="test-key",
        base_url="https://openai.invalid/v1",
        http_client=httpx.AsyncClient(trust_env=False),
    )

    sync_instructor = from_openai(
        sync_client, mode=Mode.RESPONSES_TOOLS_WITH_INBUILT_TOOLS
    )
    async_instructor = from_openai(
        async_client, mode=Mode.RESPONSES_TOOLS_WITH_INBUILT_TOOLS
    )

    assert isinstance(sync_instructor, Instructor)
    assert sync_instructor.client is sync_client
    assert sync_instructor.mode is Mode.RESPONSES_TOOLS
    assert isinstance(async_instructor, AsyncInstructor)
    assert async_instructor.client is async_client
    assert async_instructor.mode is Mode.RESPONSES_TOOLS
    sync_client.close()
    await async_client.close()


def test_openai_multimodal_encoders_cover_response_and_error_paths(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_url = Image(
        source="https://cdn.example.invalid/image.png", media_type="image/png"
    )
    assert image_to_openai(image_url, Mode.RESPONSES_TOOLS) == {
        "type": "input_image",
        "image_url": "https://cdn.example.invalid/image.png",
    }
    assert image_to_openai(image_url, Mode.TOOLS) == {
        "type": "image_url",
        "image_url": {"url": "https://cdn.example.invalid/image.png"},
    }

    image_data = Image.from_base64("data:image/png;base64,aW1hZ2U=")
    assert image_to_openai(image_data, Mode.RESPONSES_TOOLS) == {
        "type": "input_image",
        "image_url": "data:image/png;base64,aW1hZ2U=",
    }
    assert image_to_openai(image_data, Mode.TOOLS) == {
        "type": "image_url",
        "image_url": {"url": "data:image/png;base64,aW1hZ2U="},
    }

    with pytest.raises(ValueError, match="Image data is missing"):
        image_to_openai(
            Image(source="missing-image", media_type="image/png"), Mode.TOOLS
        )

    audio = Audio(source="clip.wav", media_type="audio/wav", data="YXVkaW8=")
    assert audio_to_openai(audio, Mode.TOOLS) == {
        "type": "input_audio",
        "input_audio": {"data": "YXVkaW8=", "format": "wav"},
    }
    with pytest.raises(ValueError, match="Responses doesn't support audio"):
        audio_to_openai(audio, Mode.RESPONSES_TOOLS_WITH_INBUILT_TOOLS)

    requested: list[str] = []

    def fetch(url: str) -> requests.Response:
        requested.append(url)
        response = requests.Response()
        response.status_code = 200
        response._content = b"%PDF-1.7\ncoverage"  # real response body, no network call
        return response

    monkeypatch.setattr(requests, "get", fetch)
    pdf_url = PDF(source="https://cdn.example.invalid/report.pdf")
    assert pdf_to_openai(pdf_url, Mode.RESPONSES_TOOLS) == {
        "type": "input_file",
        "filename": "https://cdn.example.invalid/report.pdf",
        "file_data": "data:application/pdf;base64,JVBERi0xLjcKY292ZXJhZ2U=",
    }
    assert pdf_to_openai(pdf_url, Mode.TOOLS) == {
        "type": "file",
        "file": {
            "filename": "https://cdn.example.invalid/report.pdf",
            "file_data": "data:application/pdf;base64,JVBERi0xLjcKY292ZXJhZ2U=",
        },
    }
    assert requested == [
        "https://cdn.example.invalid/report.pdf",
        "https://cdn.example.invalid/report.pdf",
    ]

    pdf_data = PDF.from_base64("data:application/pdf;base64,cGRm")
    assert pdf_to_openai(pdf_data, Mode.RESPONSES_TOOLS) == {
        "type": "input_file",
        "filename": "data:application/pdf;base64,cGRm",
        "file_data": "data:application/pdf;base64,cGRm",
    }
    assert pdf_to_openai(pdf_data, Mode.TOOLS) == {
        "type": "file",
        "file": {
            "filename": "data:application/pdf;base64,cGRm",
            "file_data": "data:application/pdf;base64,cGRm",
        },
    }

    with pytest.raises(ValueError, match="PDF data is missing"):
        pdf_to_openai(PDF(source="missing-pdf"), Mode.TOOLS)


def test_openai_schema_merges_only_missing_parameter_descriptions() -> None:
    class Contact(BaseModel):
        """A contact extracted from a note.

        Args:
            name: The contact's full name.
            nickname: A short display name.
            unknown: This parameter is intentionally not a model field.
        """

        name: str
        nickname: str = Field(description="Keep the explicit field description.")
        visits: int = 0

    schema = generate_openai_schema(Contact)
    properties = schema["parameters"]["properties"]

    assert schema["name"] == "Contact"
    assert schema["description"].startswith("A contact extracted from a note.")
    assert properties["name"]["description"] == "The contact's full name."
    assert (
        properties["nickname"]["description"] == "Keep the explicit field description."
    )
    assert "unknown" not in properties
    assert schema["parameters"]["required"] == ["name", "nickname"]


@pytest.mark.parametrize(
    "message",
    [
        {"role": "user"},
        {"role": "user", "content": [{"type": "text", "text": "{{ name }}"}]},
    ],
)
def test_openai_templating_leaves_non_string_content_unchanged(
    message: dict[str, Any],
) -> None:
    calls: list[str] = []

    def apply_template(text: str, _context: dict[str, Any]) -> str:
        calls.append(text)
        return "rendered"

    assert process_message(message, {"name": "Ada"}, apply_template) is message
    assert calls == []


def test_openai_templating_renders_string_content_in_place() -> None:
    message = {"role": "user", "content": "Hello {{ name }}"}
    contexts: list[dict[str, Any]] = []

    def apply_template(text: str, context: dict[str, Any]) -> str:
        contexts.append(context)
        return text.replace("{{ name }}", context["name"])

    assert process_message(message, {"name": "Ada"}, apply_template) is message
    assert message["content"] == "Hello Ada"
    assert contexts == [{"name": "Ada"}]


def test_openai_schema_supplies_a_description_when_the_model_has_no_help_text() -> None:
    class Count(BaseModel):
        value: int

    Count.__doc__ = None
    schema = generate_openai_schema(Count)

    assert schema["description"] == (
        "Correctly extracted `Count` with all the required parameters with correct types"
    )
    assert schema["parameters"]["required"] == ["value"]


def test_openai_package_stays_importable_when_client_dependency_is_missing(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    original_import = builtins.__import__
    blocked: list[str] = []

    def import_without_client(name: str, *args: Any, **kwargs: Any) -> Any:
        if name == "instructor.v2.providers.openai.client":
            blocked.append(name)
            raise ImportError("openai dependency is unavailable")
        return original_import(name, *args, **kwargs)

    monkeypatch.setattr(builtins, "__import__", import_without_client)
    init_path = Path(__file__).parents[2] / "instructor/v2/providers/openai/__init__.py"
    namespace = runpy.run_path(str(init_path), run_name="openai_optional_import_test")

    assert blocked == ["instructor.v2.providers.openai.client"]
    assert namespace["from_openai"] is None
    assert namespace["__all__"] == ["from_openai"]