osint-dashboard/tests/test_api_news.py

"""Integration tests for the news pipeline API (GET /api/news + summaries).

DB-backed: marked `requires_db` and auto-skip when the test database is
unreachable (see tests/conftest.py). Seeding writes directly to the shared
`articles` / `article_summaries` tables, exactly as the scraper + summarizer
services would.
"""

from __future__ import annotations

import asyncio
import os
from datetime import datetime, timezone

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() -> None:
    async def run():
        conn = await asyncpg.connect(**_conn_kwargs())
        try:
            await conn.execute("TRUNCATE articles, article_summaries")
        finally:
            await conn.close()

    asyncio.run(run())


@pytest.fixture()
def clean_news():
    _truncate()
    yield
    _truncate()


def _seed_article(title: str, url: str, domain: str, ts: str, content: str = "body text") -> None:
    async def run():
        conn = await asyncpg.connect(**_conn_kwargs())
        try:
            await conn.execute(
                "INSERT INTO articles (title, url, content, domain, timestamp) "
                "VALUES ($1, $2, $3, $4, $5)",
                title, url, content, domain, datetime.fromisoformat(ts),
            )
        finally:
            await conn.close()

    asyncio.run(run())


def _seed_summary(text: str, ts: str) -> None:
    async def run():
        conn = await asyncpg.connect(**_conn_kwargs())
        try:
            await conn.execute(
                "INSERT INTO article_summaries (summary_text, batch_timestamp) "
                "VALUES ($1, $2)",
                text, datetime.fromisoformat(ts),
            )
        finally:
            await conn.close()

    asyncio.run(run())


def _get(path: str) -> httpx.Response:
    async def _get_async() -> httpx.Response:
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
            return await client.get(path)

    return asyncio.run(_get_async())


@requires_db
def test_api_news_returns_newest_first(clean_news):
    _seed_article("older", "https://a.example/1", "a.example", "2026-08-24T17:00:00+00:00")
    _seed_article("newer", "https://a.example/2", "a.example", "2026-08-24T18:00:00+00:00")
    body = _get("/api/news").json()
    assert isinstance(body, list)
    assert len(body) == 2
    # newest first
    assert [a["title"] for a in body] == ["newer", "older"]
    # exact JSON contract the frontend news panel needs
    assert set(body[0].keys()) == {
        "id", "title", "url", "content", "domain", "timestamp",
    }
    assert body[0]["domain"] == "a.example"


@requires_db
def test_api_news_domain_filter(clean_news):
    _seed_article("x", "https://x.example/1", "x.example", "2026-08-24T17:00:00+00:00")
    _seed_article("y", "https://y.example/1", "y.example", "2026-08-24T17:00:00+00:00")
    body = _get("/api/news?domain=y.example").json()
    assert len(body) == 1
    assert body[0]["domain"] == "y.example"


@requires_db
def test_api_news_since_filter(clean_news):
    _seed_article("before", "https://a.example/1", "a.example", "2026-08-24T17:00:00+00:00")
    _seed_article("after", "https://a.example/2", "a.example", "2026-08-24T18:30:00+00:00")
    body = _get("/api/news?since=2026-08-24T18:00:00Z").json()
    assert len(body) == 1
    assert body[0]["title"] == "after"


@requires_db
def test_api_news_summaries_contract(clean_news):
    _seed_summary("master summary markdown…", "2026-08-24T18:05:00+00:00")
    body = _get("/api/news/summaries").json()
    assert isinstance(body, list)
    assert len(body) == 1
    s = body[0]
    assert set(s.keys()) == {"id", "summary_text", "batch_timestamp"}
    assert s["summary_text"] == "master summary markdown…"
    assert s["batch_timestamp"].startswith("2026-08-24T18:05")


@requires_db
def test_api_news_empty(clean_news):
    assert _get("/api/news").json() == []
    assert _get("/api/news/summaries").json() == []