vitya
6bbbefaa6f
- api_client.py: APIClient class with ProviderConfig types, chat_stream() returns str instead of printing, module-level wrappers for backward compat - tools.py: ToolRegistry class with instance methods, pure Python list_files (os.walk) and search_in_files (re.search) replacing find/grep for Windows compat, English tool descriptions and error messages, module-level wrappers for backward compat - 74 new tests (21 api_client + 53 tools) Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
359 lines
14 KiB
Python
359 lines
14 KiB
Python
"""Tests for meeting_room.api_client — APIClient class and module-level wrappers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from meeting_room.api_client import APIClient, chat, chat_stream, init_providers
|
|
from meeting_room.models import ProviderConfig
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures & helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
PROVIDERS = {
|
|
"testprov": ProviderConfig(
|
|
base_url="https://api.test.example.com/v1",
|
|
api_key="sk-test-key-123",
|
|
),
|
|
}
|
|
|
|
|
|
def _make_chat_response(
|
|
content: str = "Hello!",
|
|
tool_calls: list[dict] | None = None,
|
|
) -> dict:
|
|
"""Build a minimal OpenAI-style chat completion response."""
|
|
message: dict = {"content": content}
|
|
if tool_calls:
|
|
message["tool_calls"] = tool_calls
|
|
return {"choices": [{"message": message}]}
|
|
|
|
|
|
def _make_stream_lines(chunks: list[str]) -> list[str]:
|
|
"""Build SSE lines that look like what OpenAI streaming returns.
|
|
|
|
Each chunk becomes a ``data: {json}`` line. A ``data: [DONE]``
|
|
sentinel is appended automatically.
|
|
"""
|
|
lines: list[str] = []
|
|
for text in chunks:
|
|
delta = {"choices": [{"delta": {"content": text}}]}
|
|
lines.append(f"data: {json.dumps(delta)}")
|
|
lines.append("data: [DONE]")
|
|
return lines
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# APIClient — construction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAPIClientInit:
|
|
def test_stores_providers(self) -> None:
|
|
client = APIClient(PROVIDERS)
|
|
assert "testprov" in client.providers
|
|
assert client.providers["testprov"].base_url == "https://api.test.example.com/v1"
|
|
|
|
def test_empty_providers_ok(self) -> None:
|
|
client = APIClient({})
|
|
assert client.providers == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# APIClient.chat
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAPIClientChat:
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_basic_text_response(self, mock_client_cls: MagicMock) -> None:
|
|
mock_resp = MagicMock()
|
|
mock_resp.json.return_value = _make_chat_response("Hi there")
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_client_instance = mock_client_cls.return_value
|
|
mock_client_instance.post.return_value = mock_resp
|
|
|
|
client = APIClient(PROVIDERS)
|
|
result = client.chat("testprov", "gpt-4o-mini", [{"role": "user", "content": "Hi"}])
|
|
|
|
assert result == {"content": "Hi there", "tool_calls": None}
|
|
mock_client_instance.post.assert_called_once()
|
|
call_args = mock_client_instance.post.call_args
|
|
assert call_args[0][0] == "/chat/completions"
|
|
payload = call_args[1]["json"]
|
|
assert payload["model"] == "gpt-4o-mini"
|
|
assert payload["messages"] == [{"role": "user", "content": "Hi"}]
|
|
assert payload["temperature"] == 0.7
|
|
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_custom_temperature(self, mock_client_cls: MagicMock) -> None:
|
|
mock_resp = MagicMock()
|
|
mock_resp.json.return_value = _make_chat_response("cold")
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_client_cls.return_value.post.return_value = mock_resp
|
|
|
|
client = APIClient(PROVIDERS)
|
|
client.chat("testprov", "m", [], temperature=0.2)
|
|
|
|
payload = mock_client_cls.return_value.post.call_args[1]["json"]
|
|
assert payload["temperature"] == 0.2
|
|
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_tool_calls_in_response(self, mock_client_cls: MagicMock) -> None:
|
|
tool_calls = [
|
|
{
|
|
"id": "call_abc",
|
|
"function": {"name": "read_file", "arguments": '{"path": "/tmp/x"}'},
|
|
"type": "function",
|
|
}
|
|
]
|
|
mock_resp = MagicMock()
|
|
mock_resp.json.return_value = _make_chat_response("", tool_calls=tool_calls)
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_client_cls.return_value.post.return_value = mock_resp
|
|
|
|
client = APIClient(PROVIDERS)
|
|
result = client.chat("testprov", "m", [])
|
|
|
|
assert result["tool_calls"] == [
|
|
{"id": "call_abc", "name": "read_file", "arguments": '{"path": "/tmp/x"}'}
|
|
]
|
|
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_tools_sent_in_payload(self, mock_client_cls: MagicMock) -> None:
|
|
mock_resp = MagicMock()
|
|
mock_resp.json.return_value = _make_chat_response("ok")
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_client_cls.return_value.post.return_value = mock_resp
|
|
|
|
tools = [{"type": "function", "function": {"name": "search"}}]
|
|
client = APIClient(PROVIDERS)
|
|
client.chat("testprov", "m", [], tools=tools)
|
|
|
|
payload = mock_client_cls.return_value.post.call_args[1]["json"]
|
|
assert "tools" in payload
|
|
assert payload["tool_choice"] == "auto"
|
|
|
|
def test_unknown_provider_raises(self) -> None:
|
|
client = APIClient(PROVIDERS)
|
|
with pytest.raises(ValueError, match="Unknown provider: nonexistent"):
|
|
client.chat("nonexistent", "m", [])
|
|
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_empty_content_yields_empty_string(self, mock_client_cls: MagicMock) -> None:
|
|
mock_resp = MagicMock()
|
|
mock_resp.json.return_value = {"choices": [{"message": {"content": None}}]}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_client_cls.return_value.post.return_value = mock_resp
|
|
|
|
client = APIClient(PROVIDERS)
|
|
result = client.chat("testprov", "m", [])
|
|
assert result["content"] == ""
|
|
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_http_error_propagates(self, mock_client_cls: MagicMock) -> None:
|
|
mock_resp = MagicMock()
|
|
mock_resp.raise_for_status.side_effect = httpx.HTTPStatusError(
|
|
"Bad Request",
|
|
request=MagicMock(),
|
|
response=httpx.Response(400),
|
|
)
|
|
mock_client_cls.return_value.post.return_value = mock_resp
|
|
|
|
client = APIClient(PROVIDERS)
|
|
with pytest.raises(httpx.HTTPStatusError):
|
|
client.chat("testprov", "m", [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# APIClient.chat_stream
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAPIClientChatStream:
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_accumulates_text(self, mock_client_cls: MagicMock) -> None:
|
|
"""chat_stream must return the concatenated text, NOT print."""
|
|
lines = _make_stream_lines(["Hello", " world", "!"])
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.iter_lines.return_value = lines
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
|
mock_response.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_client_instance = mock_client_cls.return_value
|
|
mock_client_instance.stream.return_value = mock_response
|
|
|
|
client = APIClient(PROVIDERS)
|
|
result = client.chat_stream("testprov", "gpt-4o-mini", [{"role": "user", "content": "Hi"}])
|
|
|
|
assert result == "Hello world!"
|
|
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_stream_payload_includes_stream_flag(self, mock_client_cls: MagicMock) -> None:
|
|
lines = _make_stream_lines(["x"])
|
|
mock_response = MagicMock()
|
|
mock_response.iter_lines.return_value = lines
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
|
mock_response.__exit__ = MagicMock(return_value=False)
|
|
mock_client_cls.return_value.stream.return_value = mock_response
|
|
|
|
client = APIClient(PROVIDERS)
|
|
client.chat_stream("testprov", "m", [])
|
|
|
|
call_args = mock_client_cls.return_value.stream.call_args
|
|
payload = call_args[1]["json"]
|
|
assert payload["stream"] is True
|
|
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_empty_stream_returns_empty_string(self, mock_client_cls: MagicMock) -> None:
|
|
lines = _make_stream_lines([])
|
|
mock_response = MagicMock()
|
|
mock_response.iter_lines.return_value = lines
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
|
mock_response.__exit__ = MagicMock(return_value=False)
|
|
mock_client_cls.return_value.stream.return_value = mock_response
|
|
|
|
client = APIClient(PROVIDERS)
|
|
result = client.chat_stream("testprov", "m", [])
|
|
assert result == ""
|
|
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_malformed_lines_skipped(self, mock_client_cls: MagicMock) -> None:
|
|
"""Lines that are not valid SSE data should be silently skipped."""
|
|
raw_lines = [
|
|
"", # empty line — skipped
|
|
"not SSE", # no "data: " prefix — skipped
|
|
"data: {bad json", # malformed JSON — skipped
|
|
'data: {"choices":[{"delta":{"content":"ok"}}]}', # valid
|
|
"data: [DONE]",
|
|
]
|
|
mock_response = MagicMock()
|
|
mock_response.iter_lines.return_value = raw_lines
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
|
mock_response.__exit__ = MagicMock(return_value=False)
|
|
mock_client_cls.return_value.stream.return_value = mock_response
|
|
|
|
client = APIClient(PROVIDERS)
|
|
result = client.chat_stream("testprov", "m", [])
|
|
assert result == "ok"
|
|
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_does_not_print(self, mock_client_cls: MagicMock, capsys: pytest.CaptureFixture) -> None:
|
|
"""chat_stream must NOT write to stdout."""
|
|
lines = _make_stream_lines(["Hello", " world"])
|
|
mock_response = MagicMock()
|
|
mock_response.iter_lines.return_value = lines
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
|
mock_response.__exit__ = MagicMock(return_value=False)
|
|
mock_client_cls.return_value.stream.return_value = mock_response
|
|
|
|
client = APIClient(PROVIDERS)
|
|
result = client.chat_stream("testprov", "m", [])
|
|
|
|
captured = capsys.readouterr()
|
|
assert captured.out == ""
|
|
assert result == "Hello world"
|
|
|
|
def test_unknown_provider_raises(self) -> None:
|
|
client = APIClient(PROVIDERS)
|
|
with pytest.raises(ValueError, match="Unknown provider: nope"):
|
|
client.chat_stream("nope", "m", [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Module-level wrappers — backward compatibility
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestModuleLevelWrappers:
|
|
def test_init_providers_creates_default_client(self) -> None:
|
|
import meeting_room.api_client as mod
|
|
|
|
config = {
|
|
"providers": {
|
|
"myprov": {"base_url": "https://api.example.com/v1", "api_key": "sk-abc"},
|
|
}
|
|
}
|
|
init_providers(config)
|
|
assert mod._default_client is not None
|
|
assert "myprov" in mod._default_client.providers
|
|
assert mod._default_client.providers["myprov"].api_key == "sk-abc"
|
|
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_module_chat_delegates(self, mock_client_cls: MagicMock) -> None:
|
|
import meeting_room.api_client as mod
|
|
|
|
init_providers({"providers": {"prov": {"base_url": "https://x/v1", "api_key": "k"}}})
|
|
mock_resp = MagicMock()
|
|
mock_resp.json.return_value = _make_chat_response("delegated")
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_client_cls.return_value.post.return_value = mock_resp
|
|
|
|
result = mod.chat("prov", "m", [])
|
|
assert result["content"] == "delegated"
|
|
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_module_chat_stream_delegates(self, mock_client_cls: MagicMock) -> None:
|
|
import meeting_room.api_client as mod
|
|
|
|
init_providers({"providers": {"prov": {"base_url": "https://x/v1", "api_key": "k"}}})
|
|
lines = _make_stream_lines(["abc"])
|
|
mock_response = MagicMock()
|
|
mock_response.iter_lines.return_value = lines
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
|
mock_response.__exit__ = MagicMock(return_value=False)
|
|
mock_client_cls.return_value.stream.return_value = mock_response
|
|
|
|
result = mod.chat_stream("prov", "m", [])
|
|
assert result == "abc"
|
|
|
|
def test_module_chat_without_init_raises(self) -> None:
|
|
import meeting_room.api_client as mod
|
|
|
|
mod._default_client = None # ensure clean state
|
|
with pytest.raises(RuntimeError, match="init_providers"):
|
|
mod.chat("prov", "m", [])
|
|
|
|
def test_module_chat_stream_without_init_raises(self) -> None:
|
|
import meeting_room.api_client as mod
|
|
|
|
mod._default_client = None
|
|
with pytest.raises(RuntimeError, match="init_providers"):
|
|
mod.chat_stream("prov", "m", [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# APIClient._get_client — httpx.Client construction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetClient:
|
|
@patch("meeting_room.api_client.httpx.Client")
|
|
def test_client_created_with_correct_params(self, mock_client_cls: MagicMock) -> None:
|
|
client = APIClient(PROVIDERS)
|
|
# Trigger _get_client by calling chat with a mocked response
|
|
mock_resp = MagicMock()
|
|
mock_resp.json.return_value = _make_chat_response("ok")
|
|
mock_resp.raise_for_status = MagicMock()
|
|
mock_client_cls.return_value.post.return_value = mock_resp
|
|
|
|
client.chat("testprov", "m", [])
|
|
|
|
mock_client_cls.assert_called_once_with(
|
|
base_url="https://api.test.example.com/v1",
|
|
headers={"Authorization": "Bearer sk-test-key-123"},
|
|
timeout=120.0,
|
|
) |