98 lines
3 KiB
Python
98 lines
3 KiB
Python
|
from unittest.mock import MagicMock
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
from nous_client import chat
|
||
|
|
|
||
|
|
DEFAULT_UA = "osint-dashboard-news-summarizer"
|
||
|
|
BASE = "https://inference-api.nousresearch.com/v1"
|
||
|
|
|
||
|
|
|
||
|
|
def _ok_response(content="hello"):
|
||
|
|
resp = MagicMock()
|
||
|
|
resp.status_code = 200
|
||
|
|
resp.json.return_value = {"choices": [{"message": {"content": content}}]}
|
||
|
|
return resp
|
||
|
|
|
||
|
|
|
||
|
|
def _install_fake(monkeypatch, post_impl):
|
||
|
|
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
|
||
|
|
return post_impl(url, headers, json)
|
||
|
|
|
||
|
|
monkeypatch.setattr(httpx, "Client", FakeClient)
|
||
|
|
return captured
|
||
|
|
|
||
|
|
|
||
|
|
def test_posts_chat_completions_with_auth_body_and_returns_content(monkeypatch):
|
||
|
|
captured = _install_fake(monkeypatch, lambda *a: _ok_response("the-content"))
|
||
|
|
out = chat(
|
||
|
|
"summarize this",
|
||
|
|
api_key="secret-key",
|
||
|
|
model="hermes-3",
|
||
|
|
base_url=BASE,
|
||
|
|
)
|
||
|
|
assert out == "the-content"
|
||
|
|
assert captured["url"] == f"{BASE}/chat/completions"
|
||
|
|
assert captured["headers"]["Authorization"] == "Bearer secret-key"
|
||
|
|
assert captured["headers"]["User-Agent"] == DEFAULT_UA
|
||
|
|
assert captured["json"]["model"] == "hermes-3"
|
||
|
|
assert captured["json"]["messages"] == [{"role": "user", "content": "summarize this"}]
|
||
|
|
assert captured["json"]["temperature"] == 0.2
|
||
|
|
assert captured["json"]["max_tokens"] == 4096
|
||
|
|
assert "response_format" not in captured["json"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_user_agent_equals_osint_user_agent_env(monkeypatch):
|
||
|
|
monkeypatch.setenv("OSINT_USER_AGENT", "custom-ua/2.0")
|
||
|
|
captured = _install_fake(monkeypatch, lambda *a: _ok_response("ok"))
|
||
|
|
chat("p", api_key="k", model="m", base_url=BASE)
|
||
|
|
assert captured["headers"]["User-Agent"] == "custom-ua/2.0"
|
||
|
|
|
||
|
|
|
||
|
|
def test_json_mode_sets_response_format(monkeypatch):
|
||
|
|
captured = _install_fake(monkeypatch, lambda *a: _ok_response("{}"))
|
||
|
|
chat("p", api_key="k", model="m", base_url=BASE, json_mode=True)
|
||
|
|
assert captured["json"]["response_format"] == {"type": "json_object"}
|
||
|
|
|
||
|
|
|
||
|
|
def test_401_returns_empty_string(monkeypatch):
|
||
|
|
def post_impl(*a):
|
||
|
|
resp = MagicMock()
|
||
|
|
resp.status_code = 401
|
||
|
|
return resp
|
||
|
|
|
||
|
|
_install_fake(monkeypatch, post_impl)
|
||
|
|
assert chat("p", api_key="bad", model="m", 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)
|
||
|
|
assert chat("p", api_key="k", 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)
|
||
|
|
assert chat("p", api_key="k", model="m", base_url=BASE) == ""
|