58ff43bdee
Swap Nous Portal for the newsPipeline local_llm generate path (POST /api/generate, GET /api/tags). No API key. Settings model selector lists Ollama tags; idle if the host is down.
111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
from unittest.mock import MagicMock
|
|
|
|
import httpx
|
|
from ollama_client import chat, list_tags, normalize_base, ollama_ready
|
|
|
|
DEFAULT_UA = "osint-dashboard-news-summarizer"
|
|
BASE = "http://127.0.0.1:11434"
|
|
|
|
|
|
def _ok_response(content="hello"):
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
resp.json.return_value = {"response": content}
|
|
return resp
|
|
|
|
|
|
def _install_fake(monkeypatch, post_impl=None, get_impl=None):
|
|
captured = {}
|
|
|
|
class FakeClient:
|
|
def __init__(self, timeout=None, **kwargs):
|
|
captured["timeout"] = timeout
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def post(self, url, *, headers=None, json=None, **kwargs):
|
|
captured["url"] = url
|
|
captured["headers"] = headers
|
|
captured["json"] = json
|
|
if post_impl is None:
|
|
raise AssertionError("post not expected")
|
|
return post_impl(url, headers, json)
|
|
|
|
def get(self, url, *, headers=None, **kwargs):
|
|
captured["url"] = url
|
|
captured["headers"] = headers
|
|
if get_impl is None:
|
|
raise AssertionError("get not expected")
|
|
return get_impl(url, headers)
|
|
|
|
monkeypatch.setattr(httpx, "Client", FakeClient)
|
|
return captured
|
|
|
|
|
|
def test_normalize_base_strips_generate_path():
|
|
assert normalize_base("http://ollama:11434/api/generate") == "http://ollama:11434"
|
|
assert normalize_base("http://ollama:11434/api/") == "http://ollama:11434"
|
|
assert normalize_base("http://ollama:11434") == "http://ollama:11434"
|
|
|
|
|
|
def test_posts_generate_and_returns_response(monkeypatch):
|
|
captured = _install_fake(monkeypatch, post_impl=lambda *a: _ok_response("the-content"))
|
|
out = chat("summarize this", model="qwen3:30b-a3b", base_url=BASE)
|
|
assert out == "the-content"
|
|
assert captured["url"] == f"{BASE}/api/generate"
|
|
assert captured["headers"]["User-Agent"] == DEFAULT_UA
|
|
assert captured["json"]["model"] == "qwen3:30b-a3b"
|
|
assert captured["json"]["prompt"] == "summarize this"
|
|
assert captured["json"]["stream"] is False
|
|
assert "format" not in captured["json"]
|
|
assert captured["json"]["options"]["num_ctx"] == 32768
|
|
assert captured["timeout"] == 300.0
|
|
|
|
|
|
def test_json_mode_sets_format_json(monkeypatch):
|
|
captured = _install_fake(monkeypatch, post_impl=lambda *a: _ok_response("{}"))
|
|
chat("p", model="m", base_url=BASE, json_mode=True)
|
|
assert captured["json"]["format"] == "json"
|
|
|
|
|
|
def test_empty_model_returns_empty(monkeypatch):
|
|
assert chat("p", model="", base_url=BASE) == ""
|
|
|
|
|
|
def test_5xx_returns_empty_string(monkeypatch):
|
|
def post_impl(*a):
|
|
resp = MagicMock()
|
|
resp.status_code = 503
|
|
return resp
|
|
|
|
_install_fake(monkeypatch, post_impl=post_impl)
|
|
assert chat("p", model="m", base_url=BASE) == ""
|
|
|
|
|
|
def test_timeout_returns_empty_string(monkeypatch):
|
|
def post_impl(*a):
|
|
raise httpx.TimeoutException("timed out")
|
|
|
|
_install_fake(monkeypatch, post_impl=post_impl)
|
|
assert chat("p", model="m", base_url=BASE) == ""
|
|
|
|
|
|
def test_list_tags_and_ready(monkeypatch):
|
|
def get_impl(*a):
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
resp.json.return_value = {
|
|
"models": [{"name": "qwen3:30b-a3b"}, {"name": "llama3.2:latest"}],
|
|
}
|
|
return resp
|
|
|
|
captured = _install_fake(monkeypatch, get_impl=get_impl)
|
|
names = list_tags(BASE)
|
|
assert names == ["qwen3:30b-a3b", "llama3.2:latest"]
|
|
assert captured["url"] == f"{BASE}/api/tags"
|
|
assert ollama_ready("qwen3:30b-a3b", base_url=BASE) is True
|
|
assert ollama_ready("missing", base_url=BASE) is False
|