OpenAI Compatibility Tests

import httpx, json

TOOLS = [{"type": "function", "function": {
    "name": "say_ok", "description": "Reply with 'ok'.",
    "parameters": {"type": "object", "properties": {}, "required": []},
}}]

PERSON_SCHEMA = {
    "type": "object",
    "properties": {"name": {"type": "string"}, "age": {"type": "integer"}},
    "required": ["name", "age"],
}

def test_compat(base_url: str, model: str, label: str = ""):
    """Six checks, one per feature the prose lists. Returns {n: bool}."""
    print(f"\n=== {label or base_url} ===")
    kw = {"timeout": 120.0}
    results: dict[int, bool] = {}

    def report(n: int, name: str, ok: bool, detail: str = ""):
        results[n] = ok
        print(f"{n}. {name:<18} {'PASS' if ok else 'FAIL'}{'  ' + detail if detail else ''}")

    # 1. Non-streaming chat
    r = httpx.post(f"{base_url}/chat/completions", json={
        "model": model,
        "messages": [{"role": "user", "content": "Say 'ok' and nothing else."}],
        "stream": False,
    }, **kw)
    report(1, "non-stream", r.status_code == 200)

    # 2. Streaming chat
    with httpx.stream("POST", f"{base_url}/chat/completions", json={
        "model": model,
        "messages": [{"role": "user", "content": "Count: 1 2 3"}],
        "stream": True,
    }, timeout=None) as r2:
        chunks = sum(1 for line in r2.iter_lines() if line.startswith("data:"))
    report(2, "stream", chunks > 1, f"({chunks} chunks)")

    # 3. Multi-turn (system + user + assistant + user)
    r3 = httpx.post(f"{base_url}/chat/completions", json={
        "model": model,
        "messages": [
            {"role": "system", "content": "You answer with one word."},
            {"role": "user", "content": "My code word is BANANA. Remember it."},
            {"role": "assistant", "content": "Noted."},
            {"role": "user", "content": "What is my code word?"},
        ],
        "stream": False,
    }, **kw)
    ok3 = r3.status_code == 200 and "banana" in r3.text.lower()
    report(3, "multi-turn", ok3)

    # 4. Tool definition + tool loop (call, then feed the result back)
    r4 = httpx.post(f"{base_url}/chat/completions", json={
        "model": model,
        "messages": [{"role": "user", "content": "Use the say_ok tool."}],
        "tools": TOOLS, "stream": False,
    }, **kw)
    calls = (r4.json().get("choices", [{}])[0].get("message", {}) or {}).get("tool_calls") or []
    ok4 = bool(calls)
    if ok4:
        # The loop half: the engine must accept a tool result back.
        r4b = httpx.post(f"{base_url}/chat/completions", json={
            "model": model,
            "messages": [
                {"role": "user", "content": "Use the say_ok tool."},
                {"role": "assistant", "content": "", "tool_calls": calls},
                {"role": "tool", "tool_call_id": calls[0].get("id", "0"), "content": "ok"},
            ],
            "tools": TOOLS, "stream": False,
        }, **kw)
        ok4 = r4b.status_code == 200
    report(4, "tool loop", ok4)

    # 5. Structured output (JSON Schema)
    r5 = httpx.post(f"{base_url}/chat/completions", json={
        "model": model,
        "messages": [{"role": "user", "content": "Invent a person."}],
        "response_format": {"type": "json_schema", "json_schema": {
            "name": "person", "schema": PERSON_SCHEMA}},
        "stream": False,
    }, **kw)
    ok5 = False
    if r5.status_code == 200:
        try:
            obj = json.loads(r5.json()["choices"][0]["message"]["content"])
            ok5 = isinstance(obj.get("name"), str) and isinstance(obj.get("age"), int)
        except (json.JSONDecodeError, KeyError, AttributeError):
            ok5 = False
    report(5, "structured", ok5)

    # 6. Token usage in the response
    usage = r.json().get("usage") if r.status_code == 200 else None
    ok6 = bool(usage) and usage.get("total_tokens", 0) > 0
    report(6, "usage", ok6, f"({usage})" if usage else "")

    return results

# Run against multiple engines
test_compat("http://localhost:11434/v1", "qwen2.5:7b",   label="Ollama")
test_compat("http://localhost:8080/v1",  "qwen-direct",  label="llama-server")
# test_compat("http://localhost:8000/v1", "Qwen/Qwen2.5-7B-Instruct", label="vLLM")