newt-agent-py 0.8.2

Python bindings for newt-agent
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

"""Smoke tests for newt_agent.inference.

The async surfaces require a running asyncio loop; pytest-asyncio is
configured via `asyncio_mode = "auto"` in the root pyproject.

We deliberately do NOT exercise real Ollama / vLLM HTTP — those
tests live in the Rust side with `wiremock`. The Python suite confirms
the bindings *exist*, can be constructed, and their async return
shape is awaitable.
"""

from __future__ import annotations

import pytest

import newt_agent.inference as inference


def test_chat_request_builder_chain() -> None:
    req = inference.ChatRequest()
    req.system("You are helpful.")
    req.user("Hello")
    req.with_max_tokens(256)
    assert len(req.messages) == 2
    assert req.messages[0].role == "system"
    assert req.messages[1].role == "user"
    assert req.max_tokens == 256


def test_chat_reply_construction_and_audit_string() -> None:
    reply = inference.ChatReply("hello", "llama3.1:8b")
    assert reply.content == "hello"
    assert reply.model_id == "llama3.1:8b"
    assert reply.audit_string("ollama-local") == "backend=ollama-local model_id=llama3.1:8b"


def test_message_classmethods() -> None:
    m = inference.Message.system("sys")
    assert m.role == "system" and m.content == "sys"
    m = inference.Message.user("u")
    assert m.role == "user"
    m = inference.Message.assistant("a")
    assert m.role == "assistant"


def test_local_ollama_backend_construct() -> None:
    backend = inference.LocalOllamaBackend("http://127.0.0.1:11434", "llama3.1:8b")
    assert backend.endpoint() == "http://127.0.0.1:11434"
    assert backend.model_id() == "llama3.1:8b"
    assert backend.name() == "ollama-local"


def test_local_ollama_default_endpoints_include_localhost() -> None:
    eps = inference.LocalOllamaBackend.default_endpoints()
    assert any("127.0.0.1" in ep for ep in eps)


def test_local_vllm_backend_construct() -> None:
    backend = inference.LocalVllmBackend("http://127.0.0.1:8000", "meta/llama")
    assert backend.endpoint() == "http://127.0.0.1:8000"
    assert backend.name() == "vllm-local"


def test_local_vllm_backend_accepts_api_key() -> None:
    # Hosted OpenAI-compatible endpoints (any vendor or self-hosted gateway)
    # require a bearer token; the constructor must accept it as an optional
    # kwarg. Wire-level header behavior is covered by the Rust
    # `openai_auth_tests`; here we only assert the binding threads it through.
    backend = inference.LocalVllmBackend(
        "https://inference.example.com/v1",
        "example/model-a",
        api_key="secret-abc",
    )
    assert backend.endpoint() == "https://inference.example.com/v1"
    assert backend.name() == "vllm-local"


def test_local_vllm_backend_api_key_defaults_none() -> None:
    # Positional, no key — the default path stays unauthenticated.
    backend = inference.LocalVllmBackend("http://127.0.0.1:8000", "meta/llama")
    assert backend.endpoint() == "http://127.0.0.1:8000"


async def test_backend_registry_register_and_names() -> None:
    registry = inference.BackendRegistry()
    backend = inference.LocalOllamaBackend("http://127.0.0.1:11434", "llama3.1:8b")
    await registry.register_ollama(backend)
    assert await registry.len() == 1
    names = await registry.names()
    assert names == ["ollama-local"]


async def test_local_ollama_complete_against_unreachable_endpoint_raises() -> None:
    # Bind to an unreachable port so we exercise the awaitable path
    # without standing up a real server. The backend retries and
    # eventually raises; we just check we get an exception.
    backend = inference.LocalOllamaBackend("http://127.0.0.1:1", "llama3.1:8b")
    req = inference.ChatRequest()
    req.user("hello")
    with pytest.raises(Exception):
        await backend.complete(req)