diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_llm.py | 101 | ||||
| -rw-r--r-- | tests/test_logging.py | 88 | ||||
| -rw-r--r-- | tests/test_settings.py | 25 | ||||
| -rw-r--r-- | tests/test_tools.py | 49 |
4 files changed, 205 insertions, 58 deletions
diff --git a/tests/test_llm.py b/tests/test_llm.py index edd9d61..0e1b146 100644 --- a/tests/test_llm.py +++ b/tests/test_llm.py @@ -9,13 +9,11 @@ from typing import Any import pytest from ollama import Message from ollama._types import Image as OllamaImage -from ollama._types import ResponseError from ollama_web.llm import ( _embed_images_in_content, _is_image_message, _token_counts, - chat_with_tools, stream_chat_with_tools, ) from ollama_web.tools.registry import ToolRegistry @@ -44,46 +42,44 @@ def test_embed_images_in_content_adds_base64_markdown() -> None: assert not getattr(embedded, "images", None) -def test_chat_with_tools_skips_tools_for_non_tool_model() -> None: - """Tools are not passed when the model does not advertise tool support.""" - messages = [Message(role="user", content="hi")] - # Use an empty registry so any tool call would fail if requested. +def test_stream_chat_with_tools_disables_unsupported_capabilities( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Unsupported tools and thinking options are not sent to Ollama.""" + from ollama_web import llm + registry = ToolRegistry() - # Without tool capability, the iterator should finish immediately with no - # actual network request. We only verify it accepts the call and does not - # raise due to tool-related setup. - gen = chat_with_tools( - messages, - model="nonexistent-model", - capabilities=set(), - registry=registry, - ) - events = list(gen) - # We expect either an error (model not found) or a done event. - assert any(e.get("type") in {"error", "done"} for e in events) + def lookup(query: str) -> str: + return query + registry.register(lookup) -def test_chat_with_tools_forces_think_none_for_non_thinking_model() -> None: - messages = [Message(role="user", content="hi")] - gen = chat_with_tools( - messages, - model="nonexistent-model", - capabilities=set(), - ) - events = list(gen) - assert any(e.get("type") in {"error", "done"} for e in events) + class FakeClient: + def __init__(self) -> None: + self.calls: list[dict[str, Any]] = [] + def chat(self, **kwargs: Any) -> list[SimpleNamespace]: + self.calls.append(kwargs) + return [SimpleNamespace(message={"content": "done"})] + + client = FakeClient() + monkeypatch.setattr(llm, "get_client", lambda _host=None: client) -def test_stream_chat_with_tools_accepts_capability_set() -> None: - messages = [Message(role="user", content="hi")] - gen = stream_chat_with_tools( - messages, - model="nonexistent-model", - capabilities={"tools", "thinking"}, + events = list( + stream_chat_with_tools( + [Message(role="user", content="hi")], + model="test-model", + think=True, + capabilities=set(), + registry=registry, + ) ) - with pytest.raises(ResponseError): - list(gen) + + assert client.calls[0]["tools"] == [] + assert client.calls[0]["think"] is None + assert client.calls[0]["stream"] is True + assert events[-1] == {"type": "done"} def test_token_counts_extracts_from_dict() -> None: @@ -109,7 +105,7 @@ def test_token_counts_returns_none_when_missing() -> None: assert _token_counts(None) is None -def test_chat_with_tools_disables_tools_for_final_answer_after_limit( +def test_stream_chat_with_tools_disables_tools_for_final_answer_after_limit( monkeypatch: pytest.MonkeyPatch, ) -> None: from ollama_web import llm @@ -128,26 +124,33 @@ def test_chat_with_tools_disables_tools_for_final_answer_after_limit( def __init__(self) -> None: self.calls: list[dict[str, Any]] = [] - def chat(self, **kwargs: Any) -> dict[str, Any]: + def chat(self, **kwargs: Any) -> list[SimpleNamespace]: self.calls.append(kwargs) if len(self.calls) == 1: - return { - "message": { - "role": "assistant", - "content": "", - "tool_calls": [ - {"function": {"name": "lookup", "arguments": {"query": "test"}}} - ], - } - } - return {"message": {"role": "assistant", "content": "final answer"}} + return [ + SimpleNamespace( + message={ + "role": "assistant", + "content": "", + "tool_calls": [ + { + "function": { + "name": "lookup", + "arguments": {"query": "test"}, + } + } + ], + } + ) + ] + return [SimpleNamespace(message={"content": "final answer"})] client = FakeClient() monkeypatch.setattr(llm, "get_client", lambda _host=None: client) monkeypatch.setattr(llm, "MAX_TOOL_ROUNDS", 1) events = list( - llm.chat_with_tools( + llm.stream_chat_with_tools( [Message(role="user", content="research this")], model="test-model", registry=registry, @@ -160,6 +163,8 @@ def test_chat_with_tools_disables_tools_for_final_answer_after_limit( assert len(client.calls) == 2 assert client.calls[0]["tools"] assert client.calls[1]["tools"] == [] + assert client.calls[0]["stream"] is True + assert client.calls[1]["stream"] is True assert any(event.get("type") == "status" for event in events) assert any(event == {"type": "delta", "content": "final answer"} for event in events) assert events[-1] == {"type": "done"} diff --git a/tests/test_logging.py b/tests/test_logging.py new file mode 100644 index 0000000..ef6fbd2 --- /dev/null +++ b/tests/test_logging.py @@ -0,0 +1,88 @@ +"""Tests for application file logging.""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path +from unittest.mock import patch + +import pytest +from starlette.testclient import TestClient + +from ollama_web.app import create_app +from ollama_web.config import settings +from ollama_web.tools.registry import default_registry + + +def test_server_log_is_scoped_to_data_dir_and_app_lifespan( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + data_dir = tmp_path / "data" + monkeypatch.setattr(settings, "data_dir", str(data_dir)) + app_logger = logging.getLogger("ollama_web") + original_handlers = tuple(app_logger.handlers) + original_level = app_logger.level + + app = create_app() + log_path = data_dir / "server.log" + assert not log_path.exists() + + with TestClient(app): + assert len(app_logger.handlers) == len(original_handlers) + 1 + logging.getLogger("ollama_web.test").info("lifespan log marker") + + assert tuple(app_logger.handlers) == original_handlers + assert app_logger.level == original_level + assert log_path.read_text(encoding="utf-8").count("lifespan log marker") == 1 + + # Closing the lifespan must release the file on Windows as well. + log_path.unlink() + assert not log_path.exists() + + +def test_repeated_app_lifespans_do_not_duplicate_log_handlers( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + data_dir = tmp_path / "data" + monkeypatch.setattr(settings, "data_dir", str(data_dir)) + + with TestClient(create_app()): + logging.getLogger("ollama_web.test").info("first app marker") + with TestClient(create_app()): + logging.getLogger("ollama_web.test").info("second app marker") + + content = (data_dir / "server.log").read_text(encoding="utf-8") + assert content.count("first app marker") == 1 + assert content.count("second app marker") == 1 + + +def test_web_search_provider_error_is_written_to_server_log( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + data_dir = tmp_path / "data" + monkeypatch.setattr(settings, "data_dir", str(data_dir)) + + class FailingDDGS: + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + def text(self, query, max_results=5): + raise RuntimeError("403 Forbidden") + + with patch("ollama_web.tools.search.DDGS", FailingDDGS), TestClient(create_app()): + result = json.loads( + default_registry().execute("web_search", {"query": "private search terms"}) + ) + + assert result == {"error": "RuntimeError: 403 Forbidden"} + content = (data_dir / "server.log").read_text(encoding="utf-8") + assert "DuckDuckGo web search failed" in content + assert "RuntimeError: 403 Forbidden" in content + assert "private search terms" not in content diff --git a/tests/test_settings.py b/tests/test_settings.py index 0f334f4..d19f64d 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -4,6 +4,7 @@ from __future__ import annotations import json from pathlib import Path +from types import SimpleNamespace from typing import Any import pytest @@ -151,24 +152,28 @@ def test_options_are_forwarded_to_tool_and_final_rounds(monkeypatch: pytest.Monk def __init__(self) -> None: self.calls: list[dict[str, Any]] = [] - def chat(self, **kwargs: Any) -> dict[str, Any]: + def chat(self, **kwargs: Any) -> list[SimpleNamespace]: self.calls.append(kwargs) if len(self.calls) == 1: - return { - "message": { - "role": "assistant", - "content": "", - "tool_calls": [{"function": {"name": "missing_tool", "arguments": {}}}], - } - } - return {"message": {"role": "assistant", "content": "done"}} + return [ + SimpleNamespace( + message={ + "role": "assistant", + "content": "", + "tool_calls": [ + {"function": {"name": "missing_tool", "arguments": {}}} + ], + } + ) + ] + return [SimpleNamespace(message={"content": "done"})] client = FakeClient() monkeypatch.setattr(llm, "get_client", lambda _host=None: client) monkeypatch.setattr(llm, "MAX_TOOL_ROUNDS", 1) options = {"temperature": 0.8, "num_ctx": 8192, "top_p": 0.9} events = list( - llm.chat_with_tools( + llm.stream_chat_with_tools( [Message(role="user", content="hello")], "test-model", capabilities={"tools"}, diff --git a/tests/test_tools.py b/tests/test_tools.py index a94ea52..387cbe2 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -114,6 +114,55 @@ def test_web_search_mocked(): assert result[0]["title"] == "T" +def test_web_search_returns_empty_list_for_no_provider_results() -> None: + from ollama_web.tools import search as search_mod + + class EmptyDDGS: + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + def text(self, query, max_results=5): + return [] + + with patch.object(search_mod, "DDGS", EmptyDDGS): + assert search_mod.web_search("nothing found") == [] + + +def test_web_search_logs_and_propagates_provider_errors( + caplog: pytest.LogCaptureFixture, +) -> None: + from ollama_web.tools import search as search_mod + + class FailingDDGS: + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + def text(self, query, max_results=5): + raise RuntimeError("403 Forbidden") + + with ( + patch.object(search_mod, "DDGS", FailingDDGS), + caplog.at_level("ERROR", logger="ollama_web.tools.search"), + ): + with pytest.raises(RuntimeError, match="403 Forbidden"): + search_mod.web_search("private search terms") + + assert "DuckDuckGo web search failed" in caplog.text + assert "private search terms" not in caplog.text + + registry = ToolRegistry() + registry.register(search_mod.web_search) + with patch.object(search_mod, "DDGS", FailingDDGS): + result = json.loads(registry.execute("web_search", {"query": "test"})) + assert result == {"error": "RuntimeError: 403 Forbidden"} + + # --------------------------------------------------------------------------- # scrape_url tests (existing + pagination) # --------------------------------------------------------------------------- |