/tests/news/test_subscription_like_escaping.py

"""Regression: subscription_id LIKE wildcards must be literal (#3094).

``get_news_feed`` filters research history by substring-matching the
caller-supplied ``subscription_id`` into a ``research_meta LIKE`` pattern.
Without escaping, a ``subscription_id`` containing ``%`` or ``_`` acts as a
SQL wildcard and pulls research rows belonging to *other* subscriptions
(a wildcard-injection / enumeration bug — reachable unvalidated via the
``news/flask_api.py`` blueprint).

These use a real in-memory SQLite session (only the per-user DB session is
patched) so SQLite itself evaluates the ``LIKE ... ESCAPE`` pattern; they
fail if the escaping at ``news/api.py`` is reverted.
"""

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

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from local_deep_research.database.models import Base, ResearchHistory


@contextmanager
def _patched_user_db(seed_fn):
    """Seed a fresh in-memory DB via ``seed_fn(session)`` and patch the
    per-user DB session so ``news.api`` uses it. Only the session is
    patched; SQLite itself evaluates the real ``LIKE ... ESCAPE`` pattern.
    """
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    session_local = sessionmaker(bind=engine)

    seed = session_local()
    seed_fn(seed)
    seed.commit()
    seed.close()

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

    with patch(
        "local_deep_research.database.session_context.get_user_db_session",
        _fake_user_db,
    ):
        yield
    engine.dispose()


def _news_row(research_id, subscription_id, now):
    """A completed news-style research row bound to a subscription."""
    return ResearchHistory(
        id=research_id,
        query="latest news on the topic",
        title="Topic",
        mode="quick",
        status="completed",
        created_at=now,
        completed_at=now,
        report_content="Answer.",
        research_meta={
            "is_news_search": True,
            "subscription_id": subscription_id,
        },
    )


@contextmanager
def _seeded_feed(rows):
    """Seed ResearchHistory news rows and patch the per-user DB session.

    ``rows`` is an iterable of ``(research_id, subscription_id)``. The query
    text and ``is_news_search`` flag make ``get_news_feed`` surface each row
    as a news item, so the only thing distinguishing target from decoy is
    the ``subscription_id`` the LIKE filter matches on.
    """
    now = datetime.now(UTC).isoformat()

    def _seed(session):
        for research_id, subscription_id in rows:
            session.add(_news_row(research_id, subscription_id, now))

    with _patched_user_db(_seed):
        yield


def _feed_research_ids(subscription_id):
    from local_deep_research.news.api import get_news_feed

    result = get_news_feed(
        user_id="testuser",
        limit=20,
        use_cache=False,
        subscription_id=subscription_id,
    )
    return [it.get("research_id") for it in result.get("news_items", [])]


def test_subscription_underscore_treated_as_literal():
    """Filtering on 'a_b' must not pull the 'aXb' subscription's rows."""
    with _seeded_feed([("target", "a_b"), ("decoy", "aXb")]):
        ids = _feed_research_ids("a_b")

    assert "target" in ids
    assert "decoy" not in ids


def test_subscription_percent_treated_as_literal():
    """Filtering on 'sub-100%' must not pull 'sub-100X-extra' rows."""
    with _seeded_feed([("target", "sub-100%"), ("decoy", "sub-100X-extra")]):
        ids = _feed_research_ids("sub-100%")

    assert "target" in ids
    assert "decoy" not in ids


def test_get_subscriptions_run_count_ignores_wildcard_matches():
    """get_subscriptions' per-subscription total_runs must count only the
    exact subscription: id 'a_b' must not also count the 'aXb' run.

    Real-DB fail-on-revert coverage for the get_subscriptions LIKE site.
    """
    from local_deep_research.database.models.news import NewsSubscription
    from local_deep_research.news.api import get_subscriptions

    def _seed(session):
        session.add(
            NewsSubscription(
                id="a_b", subscription_type="search", query_or_topic="q"
            )
        )
        session.add(_news_row("target", "a_b", "2026-01-01T00:00:00"))
        session.add(_news_row("decoy", "aXb", "2026-01-01T00:00:00"))

    with _patched_user_db(_seed):
        subscriptions = get_subscriptions("testuser")["subscriptions"]

    by_id = {s["id"]: s for s in subscriptions}
    assert by_id["a_b"]["total_runs"] == 1


def test_get_subscription_history_ignores_wildcard_matches():
    """get_subscription_history for 'a_b' must not return the 'aXb'
    subscription's research runs.

    Real-DB fail-on-revert coverage for the get_subscription_history LIKE
    site. Because it exercises the full return path it also guards the
    created_at/completed_at Text handling (a reverted `.isoformat()` on the
    Text column would raise here).
    """
    from local_deep_research.database.models.news import NewsSubscription
    from local_deep_research.news.api import get_subscription_history

    def _seed(session):
        session.add(
            NewsSubscription(
                id="a_b", subscription_type="search", query_or_topic="q"
            )
        )
        session.add(_news_row("target", "a_b", "2026-01-01T00:00:00"))
        session.add(_news_row("decoy", "aXb", "2026-01-01T00:00:00"))

    with _patched_user_db(_seed):
        history = get_subscription_history("a_b")["history"]

    ids = [h["research_id"] for h in history]
    assert "target" in ids
    assert "decoy" not in ids