learningcircuit--local-deep-research/tests/web/services/test_report_assembly_service.py

"""Tests for the report_assembly_service module."""

from datetime import datetime, UTC

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,
)
from local_deep_research.web.services.report_assembly_service import (
    _build_metrics_markdown,
    _build_sources_markdown,
    assemble_full_report,
    get_research_source_links,
    get_research_source_links_batch,
)


@pytest.fixture
def db_session(tmp_path):
    """Fresh per-test SQLite session with all models created."""
    db_path = tmp_path / "assembly_test.db"
    engine = create_engine(f"sqlite:///{db_path}")
    Base.metadata.create_all(engine)
    SessionLocal = sessionmaker(bind=engine)
    sess = SessionLocal()
    yield sess
    sess.close()
    engine.dispose()


def _mk_research(db_session, **kwargs):
    """Insert a ResearchHistory row with sensible defaults."""
    defaults = dict(
        id=kwargs.pop("id", "test-research-1"),
        query=kwargs.pop("query", "What is X?"),
        mode=kwargs.pop("mode", "quick"),
        status=kwargs.pop("status", "completed"),
        created_at=kwargs.pop("created_at", "2026-04-25T12:00:00+00:00"),
        completed_at=kwargs.pop("completed_at", "2026-04-25T12:05:00+00:00"),
        report_content=kwargs.pop("report_content", "Answer body."),
        research_meta=kwargs.pop("research_meta", None),
    )
    defaults.update(kwargs)
    research = ResearchHistory(**defaults)
    db_session.add(research)
    db_session.commit()
    return research


def _mk_resource(
    db_session,
    research_id,
    *,
    url="https://example.com/a",
    title="Example",
    index="1",
    journal_quality=None,
    resource_metadata=None,
):
    """Insert a ResearchResource row mirroring save_research_sources's shape."""
    if resource_metadata is None:
        original_data = {"index": index, "url": url, "title": title}
        if journal_quality is not None:
            original_data["journal_quality"] = journal_quality
        resource_metadata = {"original_data": original_data}
    r = ResearchResource(
        research_id=research_id,
        title=title,
        url=url,
        source_type="web",
        resource_metadata=resource_metadata,
        created_at=datetime.now(UTC).isoformat(),
    )
    db_session.add(r)
    db_session.commit()
    return r


# ---------------------------------------------------------------------------
# assemble_full_report
# ---------------------------------------------------------------------------


class TestAssembleFullReport:
    def test_includes_answer_sources_metrics(self, db_session):
        research = _mk_research(
            db_session,
            report_content="The answer text [1].",
            research_meta={
                "iterations": 3,
                "generated_at": "2026-04-25T12:05:00+00:00",
            },
        )
        _mk_resource(db_session, research.id, url="https://a.com", title="A")

        out = assemble_full_report(research, db_session)
        assert "The answer text [1]." in out
        assert "## Sources" in out
        assert "https://a.com" in out
        assert "## Research Metrics" in out
        assert "Search Iterations: 3" in out

    def test_omits_sources_when_no_resources(self, db_session):
        research = _mk_research(
            db_session,
            report_content="Just an answer.",
            research_meta={},
            completed_at=None,
        )
        out = assemble_full_report(research, db_session)
        # No resources → no Sources section. No metadata + no completed_at
        # → no Metrics section. Just the answer.
        assert out == "Just an answer."

    def test_metrics_appears_when_completed_at_set_even_without_meta(
        self, db_session
    ):
        """completed_at alone is enough to render the Metrics section."""
        research = _mk_research(
            db_session,
            report_content="Answer body.",
            research_meta=None,
            completed_at="2026-04-25T12:05:00+00:00",
        )
        out = assemble_full_report(research, db_session)
        assert "## Research Metrics" in out
        assert "Generated at: 2026-04-25T12:05:00+00:00" in out

    def test_omits_metrics_when_no_metadata(self, db_session):
        research = _mk_research(
            db_session,
            report_content="Answer.",
            research_meta=None,
            completed_at=None,
        )
        out = assemble_full_report(research, db_session)
        # No sources, no metadata, no completed_at → just the answer.
        assert out == "Answer."

    def test_handles_none_research(self, db_session):
        # None research → None return (distinct from "" which means
        # "research exists but has no body / sources / metrics yet").
        # Callers map None → HTTP 404 and "" → HTTP 200 empty.
        assert assemble_full_report(None, db_session) is None

    def test_handles_none_resource_metadata(self, db_session):
        """Defensive isinstance check survives a row with metadata=None."""
        research = _mk_research(db_session, report_content="x")
        # Bypass _mk_resource's auto-built metadata; insert a row with None.
        r = ResearchResource(
            research_id=research.id,
            url="https://nometa.com",
            title="No Meta",
            source_type="web",
            resource_metadata=None,
            created_at=datetime.now(UTC).isoformat(),
        )
        db_session.add(r)
        db_session.commit()
        out = assemble_full_report(research, db_session)
        assert "https://nometa.com" in out

    def test_legacy_row_with_inline_sources_not_double_rendered(
        self, db_session
    ):
        """Pre-refactor rows already embed ## Sources in report_content.

        Without the legacy-row guard the assembler would append a freshly
        built Sources block on top, producing two `## Sources` headings.
        """
        legacy_body = (
            "The answer body.\n\n"
            "## Sources\n\n"
            "[1] Old Source (source nr: 1)\n"
            "   URL: https://old.com\n"
        )
        research = _mk_research(db_session, report_content=legacy_body)
        # Add a structured resource that, without the guard, would render
        # as a SECOND `## Sources` block.
        r = ResearchResource(
            research_id=research.id,
            url="https://new.com",
            title="New",
            source_type="web",
            resource_metadata={"original_data": {"index": 1}},
            created_at=datetime.now(UTC).isoformat(),
        )
        db_session.add(r)
        db_session.commit()

        out = assemble_full_report(research, db_session)

        assert out.count("## Sources") == 1
        # The legacy body's existing source URL is preserved as-is; the
        # structured `https://new.com` is suppressed because the legacy
        # block already covers that section.
        assert "https://old.com" in out
        assert "https://new.com" not in out

    def test_legacy_row_with_inline_metrics_not_double_rendered(
        self, db_session
    ):
        """Same guard for the `## Research Metrics` section."""
        legacy_body = "Body.\n\n## Research Metrics\n- Generated at: 2025-01-01"
        research = _mk_research(
            db_session,
            report_content=legacy_body,
            research_meta={"iterations": 7},
        )
        out = assemble_full_report(research, db_session)
        assert out.count("## Research Metrics") == 1
        # Iteration count from the new path is NOT appended because the
        # legacy block already owns that section.
        assert (
            "iterations" not in out.lower().replace("research metrics", "")
            or "7" not in out
        )

    def test_inline_sources_in_prose_does_not_trigger_legacy_guard(
        self, db_session
    ):
        """Regression for the loose-substring trap: prose that mentions
        ``## Sources`` mid-line must NOT trip the legacy guard (line-
        anchored regex). Otherwise a new-row LLM answer that quotes a
        markdown snippet would lose its structured Sources block."""
        body = (
            "Use a markdown heading like ## Sources to list references "
            "(this is just an explanation of markdown)."
        )
        research = _mk_research(db_session, report_content=body)
        r = ResearchResource(
            research_id=research.id,
            url="https://kept.com",
            title="Kept",
            source_type="web",
            resource_metadata={"original_data": {"index": 1}},
            created_at=datetime.now(UTC).isoformat(),
        )
        db_session.add(r)
        db_session.commit()

        out = assemble_full_report(research, db_session)
        # The structured Sources block should still render even though the
        # body contains the substring `## Sources`.
        assert "https://kept.com" in out
        # The literal `## Sources` from prose plus the appended header
        # gives count == 2 — both legitimate occurrences.
        assert out.count("## Sources") == 2

    def test_falls_back_to_row_order_when_index_missing(
        self, db_session, caplog
    ):
        research = _mk_research(db_session, report_content="x")
        # Insert two resources where original_data has no index.
        for url in ("https://a.com", "https://b.com"):
            r = ResearchResource(
                research_id=research.id,
                url=url,
                title=url,
                source_type="web",
                resource_metadata={"original_data": {}},
                created_at=datetime.now(UTC).isoformat(),
            )
            db_session.add(r)
        db_session.commit()

        # Capture loguru output by intercepting via the standard logger.
        # _build_sources_markdown logs at DEBUG so we just verify no
        # crash and both rows render.
        out = assemble_full_report(research, db_session)
        assert "https://a.com" in out
        assert "https://b.com" in out


# ---------------------------------------------------------------------------
# _build_metrics_markdown
# ---------------------------------------------------------------------------


class TestBuildMetricsMarkdown:
    def test_renders_iterations_and_generated_at(self, db_session):
        research = _mk_research(
            db_session,
            research_meta={"iterations": 5, "generated_at": "2026-01-01"},
        )
        md = _build_metrics_markdown(research)
        assert "Search Iterations: 5" in md
        assert "Generated at: 2026-01-01" in md

    def test_falls_back_to_completed_at_when_generated_at_missing(
        self, db_session
    ):
        research = _mk_research(
            db_session,
            research_meta={"iterations": 2},
            completed_at="2026-02-02T00:00:00+00:00",
        )
        md = _build_metrics_markdown(research)
        assert "Generated at: 2026-02-02T00:00:00+00:00" in md

    def test_returns_empty_when_no_data(self, db_session):
        research = _mk_research(
            db_session, research_meta=None, completed_at=None
        )
        assert _build_metrics_markdown(research) == ""


# ---------------------------------------------------------------------------
# _build_sources_markdown
# ---------------------------------------------------------------------------


class TestBuildSourcesMarkdown:
    def test_uses_original_index_when_present(self, db_session):
        research = _mk_research(db_session)
        _mk_resource(
            db_session, research.id, url="https://x.com", title="X", index="3"
        )
        md = _build_sources_markdown(research, db_session)
        assert "[3]" in md
        assert "https://x.com" in md

    def test_returns_empty_when_no_resources(self, db_session):
        research = _mk_research(db_session)
        assert _build_sources_markdown(research, db_session) == ""

    def test_index_zero_is_not_treated_as_missing(self, db_session):
        """Citation index 0 is a valid value; only None / '' triggers fallback."""
        research = _mk_research(db_session)
        _mk_resource(
            db_session, research.id, url="https://z.com", title="Z", index=0
        )
        md = _build_sources_markdown(research, db_session)
        assert "[0]" in md or "0]" in md  # exact rendering may vary


# ---------------------------------------------------------------------------
# get_research_source_links
# ---------------------------------------------------------------------------


class TestGetResearchSourceLinks:
    def test_returns_top_n_in_row_order(self, db_session):
        research = _mk_research(db_session)
        for i, url in enumerate(
            ["https://a.com", "https://b.com", "https://c.com", "https://d.com"]
        ):
            _mk_resource(
                db_session,
                research.id,
                url=url,
                title=f"T{i}",
                index=str(i + 1),
            )
        links = get_research_source_links(research.id, db_session, limit=3)
        assert len(links) == 3
        assert [link["url"] for link in links] == [
            "https://a.com",
            "https://b.com",
            "https://c.com",
        ]

    def test_returns_empty_when_no_resources(self, db_session):
        research = _mk_research(db_session)
        assert get_research_source_links(research.id, db_session) == []

    def test_skips_non_http_urls(self, db_session):
        research = _mk_research(db_session)
        _mk_resource(db_session, research.id, url="ftp://nope.com", title="X")
        _mk_resource(
            db_session,
            research.id,
            url="https://yes.com",
            title="Y",
            index="2",
        )
        links = get_research_source_links(research.id, db_session)
        assert [link["url"] for link in links] == ["https://yes.com"]

    def test_falls_back_to_domain_when_title_missing(self, db_session):
        research = _mk_research(db_session)
        r = ResearchResource(
            research_id=research.id,
            url="https://www.foo.com/page",
            title="",
            source_type="web",
            resource_metadata={"original_data": {"index": "1"}},
            created_at=datetime.now(UTC).isoformat(),
        )
        db_session.add(r)
        db_session.commit()
        links = get_research_source_links(research.id, db_session)
        assert links[0]["title"] == "foo.com"


# ---------------------------------------------------------------------------
# get_research_source_links_batch
# ---------------------------------------------------------------------------


class TestGetResearchSourceLinksBatch:
    def test_groups_links_by_research_id(self, db_session):
        r1 = _mk_research(db_session, id="r1")
        r2 = _mk_research(db_session, id="r2")
        _mk_resource(db_session, r1.id, url="https://r1a.com", title="r1a")
        _mk_resource(db_session, r2.id, url="https://r2a.com", title="r2a")
        _mk_resource(db_session, r2.id, url="https://r2b.com", title="r2b")

        batch = get_research_source_links_batch(["r1", "r2"], db_session)
        assert len(batch["r1"]) == 1
        assert len(batch["r2"]) == 2
        assert batch["r1"][0]["url"] == "https://r1a.com"

    def test_empty_input_returns_empty_dict(self, db_session):
        assert get_research_source_links_batch([], db_session) == {}

    def test_research_with_no_resources_maps_to_empty_list(self, db_session):
        _mk_research(db_session, id="empty-r")
        batch = get_research_source_links_batch(["empty-r"], db_session)
        assert batch == {"empty-r": []}

    def test_respects_limit_per_research(self, db_session):
        r = _mk_research(db_session)
        for i, url in enumerate(
            ["https://a.com", "https://b.com", "https://c.com"]
        ):
            _mk_resource(
                db_session, r.id, url=url, title=f"T{i}", index=str(i + 1)
            )
        batch = get_research_source_links_batch([r.id], db_session, limit=2)
        assert len(batch[r.id]) == 2