tests/test_llm_integration.py

from __future__ import annotations

import json
import os

import pytest
from loosecanvas.llm_probe import (
    BARE_JSON_SCHEMA,
    JSON_OBJECT_ENUM_SCHEMA,
    OPENAI_WRAPPED_ENUM_SCHEMA,
    SCENEPLAN_CONTEXT,
    SCENEPLAN_SYSTEM_PROMPT,
    TOOL_MESSAGES,
    WEATHER_TOOL,
    assistant_message,
    chat,
    resolve_sceneplan_response_format,
    server_available,
    tool_result_roundtrip_messages,
    valid_enum_payload,
    valid_weather_tool_call,
    validate_sceneplan_payload,
)

RUN_ENV_VAR = "LOOSECANVAS_RUN_LLM_INTEGRATION"

if os.environ.get(RUN_ENV_VAR) != "1":
    pytest.skip(
        f"Set {RUN_ENV_VAR}=1 to run live llama.cpp integration tests.",
        allow_module_level=True,
    )

if not server_available():
    pytest.skip("Live llama.cpp server is not reachable.", allow_module_level=True)


def test_bare_response_format_remains_unenforced() -> None:
    resp = chat(
        [{"role": "user", "content": "Is the sky blue on a clear day? Answer."}],
        response_format=BARE_JSON_SCHEMA,
        max_tokens=20,
        chat_template_kwargs={"enable_thinking": False},
    )
    content = assistant_message(resp).get("content") or ""
    assert not valid_enum_payload(content)


@pytest.mark.parametrize("response_format", [OPENAI_WRAPPED_ENUM_SCHEMA, JSON_OBJECT_ENUM_SCHEMA])  # type: ignore[untyped-decorator]
def test_structured_enum_enforcement(response_format: dict[str, object]) -> None:
    resp = chat(
        [{"role": "user", "content": "Is the sky blue on a clear day? Answer."}],
        response_format=response_format,
        max_tokens=20,
        chat_template_kwargs={"enable_thinking": False},
    )
    content = assistant_message(resp).get("content") or ""
    assert valid_enum_payload(content)


def test_sceneplan_schema_enforcement() -> None:
    response_format, source = resolve_sceneplan_response_format()
    assert source in {
        "fallback_inline_schema",
        "project_model_json_schema",
        "project_response_format",
    }
    resp = chat(
        [
            {"role": "system", "content": SCENEPLAN_SYSTEM_PROMPT},
            {"role": "user", "content": json.dumps(SCENEPLAN_CONTEXT)},
        ],
        response_format=response_format,
        max_tokens=600,
        chat_template_kwargs={"enable_thinking": False},
    )
    content = assistant_message(resp).get("content") or ""
    assert validate_sceneplan_payload(content)


def test_tool_call_generation_and_roundtrip() -> None:
    tool_resp = chat(
        TOOL_MESSAGES,
        tools=[WEATHER_TOOL],
        tool_choice="auto",
        parse_tool_calls=True,
        max_tokens=256,
        chat_template_kwargs={"enable_thinking": False},
    )
    tool_msg = assistant_message(tool_resp)
    assert valid_weather_tool_call(tool_msg)

    final_resp = chat(
        TOOL_MESSAGES + tool_result_roundtrip_messages(tool_msg),
        tools=[WEATHER_TOOL],
        tool_choice="none",
        parse_tool_calls=True,
        max_tokens=128,
        chat_template_kwargs={"enable_thinking": False},
    )
    final_content = (assistant_message(final_resp).get("content") or "").lower()
    assert "seattle" in final_content
    assert (
        "17" in final_content
        or "light rain" in final_content
        or "celsius" in final_content
    )