osint-dashboard/tests/test_api_settings.py

"""API tests for GET/PUT /api/settings and GET /api/news/models."""

from __future__ import annotations

import asyncio
import os

import asyncpg
import httpx
import pytest

from conftest import requires_db

from main import app

BASE = "http://test"


def _conn_kwargs() -> dict:
    return {
        "host": os.environ["DB_HOST"],
        "port": int(os.environ["DB_PORT"]),
        "user": os.environ["DB_USER"],
        "password": os.environ["DB_PASSWORD"],
        "database": os.environ["DB_NAME"],
    }


def _truncate_settings() -> None:
    async def run():
        conn = await asyncpg.connect(**_conn_kwargs())
        try:
            await conn.execute("DROP TABLE IF EXISTS app_settings")
        finally:
            await conn.close()

    asyncio.run(run())


@pytest.fixture()
def clean_settings():
    import settings_store
    settings_store._ensured = False
    _truncate_settings()
    yield
    settings_store._ensured = False
    _truncate_settings()


def _request(method: str, path: str, json: dict | None = None) -> httpx.Response:
    async def _run() -> httpx.Response:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
            return await client.request(method, path, json=json)

    return asyncio.run(_run())


def _get(path: str) -> httpx.Response:
    return _request("GET", path)


def _put(path: str, json: dict) -> httpx.Response:
    return _request("PUT", path, json=json)


def test_get_news_models_without_key_returns_fallback(monkeypatch):
    monkeypatch.delenv("NOUS_API_KEY", raising=False)
    monkeypatch.setattr("keystore.get_api_key", _missing_key)

    resp = _get("/api/news/models")
    assert resp.status_code == 200
    body = resp.json()
    assert body["source"] == "fallback"
    ids = [m["id"] for m in body["models"]]
    assert "Hermes-4.3-36B" in ids
    from settings_store import FALLBACK_MODELS
    assert ids == FALLBACK_MODELS


async def _missing_key(name: str):
    return None


async def _present_key(name: str):
    return "test-nous-api-key-1234"


def test_get_news_models_live_from_upstream(monkeypatch):
    monkeypatch.setattr("keystore.get_api_key", _present_key)
    monkeypatch.setenv("NOUS_BASE_URL", "https://inference-api.nousresearch.com/v1")
    monkeypatch.delenv("OSINT_USER_AGENT", raising=False)

    captured: dict = {}

    class FakeResponse:
        status_code = 200

        def raise_for_status(self):
            return None

        def json(self):
            return {"data": [{"id": "live-model-a"}, {"id": "Hermes-4.3-36B"}]}

    async def fake_http_get(url, *, headers, timeout):
        captured["url"] = url
        captured["headers"] = headers
        captured["timeout"] = timeout
        return FakeResponse()

    import settings_store
    monkeypatch.setattr(settings_store, "_http_get", fake_http_get)
    settings_store._models_cache = None

    resp = _get("/api/news/models")
    assert resp.status_code == 200
    body = resp.json()
    assert body["source"] == "live"
    ids = [m["id"] for m in body["models"]]
    assert ids == ["live-model-a", "Hermes-4.3-36B"]
    assert captured["url"] == "https://inference-api.nousresearch.com/v1/models"
    assert captured["headers"]["User-Agent"] == "osint-dashboard-news-summarizer"
    assert captured["headers"]["Authorization"] == "Bearer test-nous-api-key-1234"
    timeout = captured["timeout"]
    assert timeout == 8 or getattr(timeout, "read", timeout) == 8 or float(timeout) == 8.0


def test_get_news_models_upstream_failure_returns_fallback(monkeypatch):
    monkeypatch.setattr("keystore.get_api_key", _present_key)

    async def boom_http_get(url, *, headers, timeout):
        raise httpx.ConnectError("upstream down")

    import settings_store
    monkeypatch.setattr(settings_store, "_http_get", boom_http_get)
    settings_store._models_cache = None

    resp = _get("/api/news/models")
    assert resp.status_code == 200
    body = resp.json()
    assert body["source"] == "fallback"
    ids = [m["id"] for m in body["models"]]
    assert "Hermes-4.3-36B" in ids


def test_put_settings_empty_returns_422():
    resp = _put("/api/settings", {"summary_model": ""})
    assert resp.status_code == 422


def test_put_settings_whitespace_only_returns_422():
    resp = _put("/api/settings", {"summary_model": "   "})
    assert resp.status_code == 422


@requires_db
def test_put_settings_round_trip(clean_settings, monkeypatch):
    monkeypatch.delenv("SUMMARY_MODEL", raising=False)
    monkeypatch.delenv("NOUS_BASE_URL", raising=False)

    put_resp = _put("/api/settings", {"summary_model": "google/gemini-2.5-flash"})
    assert put_resp.status_code == 200
    put_body = put_resp.json()
    assert put_body["summary_model"] == "google/gemini-2.5-flash"
    assert put_body["nous_base_url"] == "https://inference-api.nousresearch.com/v1"

    get_resp = _get("/api/settings")
    assert get_resp.status_code == 200
    get_body = get_resp.json()
    assert get_body["summary_model"] == "google/gemini-2.5-flash"
    assert get_body["nous_base_url"] == "https://inference-api.nousresearch.com/v1"
    assert set(get_body.keys()) == {"summary_model", "nous_base_url"}