/tests/news/test_news_api_contract.py

"""API contract tests for the news feed report shape (#3665 Fix B).

After the chat-mode-v2 refactor, ``report_content`` stores only the
synthesized answer. The news feed must therefore expose:

* ``findings`` — the answer-only ``report_content`` verbatim (no embedded
  ``## Sources`` block), and
* ``links`` — the structured top-N source URLs read from the
  ``research_resources`` table (not parsed out of ``report_content``).

These exercise the real ``get_news_feed`` research-history path against a
seeded DB; only the per-user-DB session is patched.
"""

from contextlib import contextmanager
from datetime import datetime, UTC
from unittest.mock import patch

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from local_deep_research.database.models import Base
from local_deep_research.database.models.research import (
    ResearchHistory,
    ResearchResource,
)

RESEARCH_ID = "news-contract-research-1"
# "latest news" in the query makes get_news_feed treat the row as a news item.
QUERY = "latest news on quantum computing breakthroughs"
# Answer-only report content — inline [N](url) citations, NO ## Sources block.
ANSWER_ONLY = (
    "Quantum error correction crossed a threshold this year, "
    "citing [1](https://q1.example) and [2](https://q2.example)."
)
N_SOURCES = 4  # > the feed's top-N (3): links must be capped, not unbounded.


@pytest.fixture
def seeded_news_db(tmp_path):
    """Real temp-file SQLite seeded with one completed news-style research."""
    engine = create_engine(f"sqlite:///{tmp_path / 'news_contract.db'}")
    Base.metadata.create_all(engine)
    SessionLocal = sessionmaker(bind=engine)

    seed = SessionLocal()
    seed.add(
        ResearchHistory(
            id=RESEARCH_ID,
            query=QUERY,
            title="Quantum Computing Breakthroughs",
            mode="quick",
            status="completed",
            created_at="2026-04-25T12:00:00+00:00",
            completed_at="2026-04-25T12:05:00+00:00",
            report_content=ANSWER_ONLY,
            research_meta={"is_news_search": True},
        )
    )
    for i in range(1, N_SOURCES + 1):
        url = f"https://q{i}.example"
        seed.add(
            ResearchResource(
                research_id=RESEARCH_ID,
                title=f"Quantum Source {i}",
                url=url,
                source_type="web",
                resource_metadata={
                    "original_data": {
                        "index": str(i),
                        "url": url,
                        "title": f"Quantum Source {i}",
                    }
                },
                created_at=datetime.now(UTC).isoformat(),
            )
        )
    seed.commit()
    seed.close()

    @contextmanager
    def _fake_user_db(username=None, password=None):
        db = SessionLocal()
        try:
            yield db
        finally:
            db.close()

    # get_news_feed imports get_user_db_session from session_context at call
    # time, so patch it at the source module.
    with patch(
        "local_deep_research.database.session_context.get_user_db_session",
        _fake_user_db,
    ):
        yield

    engine.dispose()


def _feed_item(seeded_news_db):
    from local_deep_research.news.api import get_news_feed

    result = get_news_feed(user_id="testuser", limit=10, use_cache=False)
    items = result["news_items"]
    matches = [it for it in items if it.get("research_id") == RESEARCH_ID]
    assert matches, f"seeded research not surfaced in feed: {items}"
    return matches[0]


def test_news_feed_findings_is_answer_only(seeded_news_db):
    item = _feed_item(seeded_news_db)
    findings = item["findings"]
    assert findings == ANSWER_ONLY
    # The post-refactor contract: no assembled ## Sources blob in findings.
    assert "## Sources" not in findings


def test_news_feed_links_array_populated_from_research_resources(
    seeded_news_db,
):
    item = _feed_item(seeded_news_db)
    links = item["links"]
    assert isinstance(links, list)
    # Exactly the top-N cap: the feed passes limit=3 (news/api.py), so 3 of
    # the 4 seeded resources come back. == 3 (not <= 3) fences the cap AND
    # the source: report_content only cites [1]/[2] inline, so a regression
    # that parsed links out of report_content could yield at most 2.
    assert len(links) == 3
    urls = {link.get("url") for link in links}
    # At least one link is a research_resources-only source (q3/q4), absent
    # from the two inline citations in report_content — proves the links are
    # read FROM the table, not parsed out of the answer text.
    assert urls - {"https://q1.example", "https://q2.example"}, urls
    for link in links:
        assert link.get("url")