test_process_runner.py

import datetime
import json
import re
from typing import Any, NamedTuple, cast
from unittest.mock import AsyncMock, MagicMock

import pytest

from dank.config import RedditSourceConfig, SourceConfig
from dank.model import Post
from dank.process import runner
from dank.process.rss import convert_raw_post
from dank.process.runner import (
    _recover_saved_pages,  # pyright: ignore[reportPrivateUsage]
    parse_age_window,
    process_source_assets,
    process_source_posts,
)


class _Result(NamedTuple):
    rows: list[dict[str, Any]]


class _DummyClient:
    def __init__(self) -> None:
        self.query = ""
        self.params: dict[str, object] = {}

    async def fetch_json(
        self,
        query: str,
        params: dict[str, object] | None = None,
    ) -> _Result:
        self.query = query
        self.params = params or {}

        return _Result(rows=[])


class _DummyEmbedder:
    model_name = "test-model"

    def embed_texts(self, items: list[str]) -> list[tuple[float, ...]]:
        return [() for _ in items]


class _InsertClient:
    def __init__(self) -> None:
        self.query = ""
        self.params: dict[str, object] = {}
        self.table = ""
        self.rows: list[dict[str, Any]] = []

    async def fetch_json(
        self,
        query: str,
        params: dict[str, object] | None = None,
    ) -> _Result:
        self.query = query
        self.params = params or {}

        return _Result(
            rows=[
                {
                    "domain": "x.com",
                    "post_id": "1",
                    "url": "https://x.com/i/status/1",
                    "post_created_at": None,
                    "scraped_at": datetime.datetime(
                        2026,
                        2,
                        1,
                        tzinfo=datetime.UTC,
                    ),
                    "source": "x",
                    "request_url": "https://x.com/i/api/graphql/Example",
                    "payload": "{}",
                },
            ],
        )

    async def insert_rows(
        self,
        table: str,
        rows: list[dict[str, Any]],
    ) -> None:
        self.table = table
        self.rows = rows


class _DummyAssetClient:
    def __init__(self) -> None:
        self.query = ""
        self.params: dict[str, object] = {}

    async def fetch_json(
        self,
        query: str,
        params: dict[str, object] | None = None,
    ) -> _Result:
        self.query = query
        self.params = params or {}

        return _Result(rows=[])

    async def insert_rows(
        self,
        _table: str,
        _rows: list[dict[str, Any]],
    ) -> None:
        return


def test_parse_age_window_seconds() -> None:
    assert parse_age_window("30s") == datetime.timedelta(seconds=30)
    assert parse_age_window("15") == datetime.timedelta(seconds=15)


def test_parse_age_window_minutes() -> None:
    assert parse_age_window("10m") == datetime.timedelta(minutes=10)
    assert parse_age_window("5min") == datetime.timedelta(minutes=5)


def test_parse_age_window_hours() -> None:
    assert parse_age_window("2h") == datetime.timedelta(hours=2)
    assert parse_age_window("1hour") == datetime.timedelta(hours=1)


@pytest.mark.parametrize("value", ["", "0", "-5m", "3d", "abc"])
def test_parse_age_window_invalid(value: str) -> None:
    with pytest.raises(ValueError):
        parse_age_window(value)


async def test_process_source_posts_filters_by_scraped_at() -> None:
    client = _DummyClient()
    since = datetime.datetime.now(datetime.UTC)

    converted = await process_source_posts(
        cast(Any, client),
        "x.com",
        lambda _row: None,
        since=since,
        embedder=cast(Any, _DummyEmbedder()),
    )

    assert converted == 0
    assert "AND scraped_at >= %(since)s" in client.query
    assert "coalesce(post_created_at, scraped_at)" not in client.query


async def test_process_source_posts_only_selects_unprocessed_posts() -> None:
    client = _DummyClient()
    since = datetime.datetime.now(datetime.UTC)

    converted = await process_source_posts(
        cast(Any, client),
        "x.com",
        lambda _row: None,
        since=since,
        embedder=cast(Any, _DummyEmbedder()),
    )

    assert converted == 0
    assert "LIMIT 1 BY post_id" in client.query
    assert "LEFT JOIN" in client.query
    assert "raw.scraped_at > processed.updated_at" in client.query
    assert client.params["reprocess"] is False


async def test_reprocess_keeps_domain_and_age_bounds() -> None:
    client = _DummyClient()
    since = datetime.datetime(2026, 10, 1, tzinfo=datetime.UTC)

    await process_source_posts(
        cast(Any, client), "publisher.test", lambda _row: None,
        since=since, embedder=cast(Any, _DummyEmbedder()), reprocess=True,
    )

    assert "WHERE %(reprocess)s OR" in client.query
    assert (
        "WHERE domain = %(domain)s AND NOT is_reference "
        "AND scraped_at >= %(since)s"
    ) in client.query
    assert "LIMIT 1 BY post_id" in client.query
    assert client.params == {
        "domain": "publisher.test", "since": since, "reprocess": True,
    }


async def test_process_run_scopes_reprocessing_to_selected_sources(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    settings = MagicMock()
    settings.sources = (
        SourceConfig(domain="www.theregister.com", accounts=()),
        SourceConfig(domain="other.test", accounts=()),
    )
    posts = AsyncMock(return_value=1)
    assets = AsyncMock(return_value=0)
    monkeypatch.setattr(runner, "process_source_posts", posts)
    monkeypatch.setattr(runner, "process_source_assets", assets)
    monkeypatch.setattr(runner, "ClickHouseClient", MagicMock())

    assert await runner.run_process(
        settings, age="168h", reprocess=True,
        domain_regex=re.compile(r"^www\.theregister\.com$"),
    ) == 1

    posts.assert_awaited_once()
    assert posts.call_args.args[1] == "www.theregister.com"
    assert posts.call_args.kwargs["reprocess"] is True
    assets.assert_awaited_once()
    assert assets.call_args.args[1] == "www.theregister.com"
    assert "reprocess" not in assets.call_args.kwargs


async def test_all_age_processes_every_configured_source(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    settings = MagicMock()
    settings.sources = (
        SourceConfig(domain="first.test", accounts=()),
        SourceConfig(domain="second.test", accounts=()),
    )
    posts = AsyncMock(return_value=1)
    assets = AsyncMock(return_value=0)
    monkeypatch.setattr(runner, "process_source_posts", posts)
    monkeypatch.setattr(runner, "process_source_assets", assets)
    monkeypatch.setattr(runner, "ClickHouseClient", MagicMock())

    assert await runner.run_process(settings, age="all", reprocess=True) == 2
    assert [call.args[1] for call in posts.call_args_list] == [
        "first.test", "second.test",
    ]

    for call in posts.call_args_list + assets.call_args_list:
        assert call.kwargs["since"] == datetime.datetime(
            1900, 1, 1, tzinfo=datetime.UTC,
        )


async def test_reddit_listings_are_processed_once_with_the_reddit_converter(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    settings = MagicMock()
    settings.sources = (
        SourceConfig("reddit.com", (), (
            "https://www.reddit.com/r/python/",
        )),
        SourceConfig("reddit.com", (), (
            "https://www.reddit.com/r/LocalLLaMA/",
        ), reddit=RedditSourceConfig()),
    )
    posts = AsyncMock(return_value=1)
    assets = AsyncMock(return_value=0)
    monkeypatch.setattr(runner, "process_source_posts", posts)
    monkeypatch.setattr(runner, "process_source_assets", assets)
    monkeypatch.setattr(runner, "ClickHouseClient", MagicMock())

    assert await runner.run_process(settings, age="all", reprocess=True) == 1
    posts.assert_awaited_once()
    assert posts.call_args.args[1:3] == (
        "reddit.com", runner.convert_raw_reddit_post,
    )
    assets.assert_awaited_once()


async def test_saved_pages_use_newest_match_and_source_identity() -> None:
    now = datetime.datetime(2026, 10, 9, tzinfo=datetime.UTC)
    post = Post(
        "a.test", "same", "https://a.test/latest", now, now, "Author",
        "Current title", (), "First. Second.", (), "rss",
    )
    other = post._replace(domain="b.test")
    formatted = post._replace(post_id="formatted", html="<p>Already HTML</p>")
    social = post._replace(domain="x.com", source="x")
    client = MagicMock()
    client.fetch_json = AsyncMock(return_value=_Result([
        {"domain": "a.test", "post_id": "same", "url": post.url,
         "payload": json.dumps({"page_html": "<head>Challenge</head>"})},
        {"domain": "a.test", "post_id": "same", "url": post.url,
         "payload": json.dumps({"feed_xml": "<item/>",
                                "page_html": '<article><p>First.</p>'
                                '<p><em>Second.</em></p></article>'})},
        {"domain": "a.test", "post_id": "same", "url": post.url,
         "payload": json.dumps({"feed_xml": "<item/>",
                                "page_html": '<article><p>First.</p>'
                                '<p>Second.</p></article>'})},
    ]))
    recovered = await _recover_saved_pages(
        client, [post, other, formatted, social],
    )

    assert recovered == [post._replace(
        html='<p>First.</p>\n<p><em>Second.</em></p>',
    ), other, formatted, social]
    query, params = client.fetch_json.call_args.args
    assert "ORDER BY scraped_at DESC" in query
    assert "since" not in query
    assert params == {"post_keys": [("a.test", "same"), ("b.test", "same")]}

    client.fetch_json.reset_mock()
    assert await _recover_saved_pages(client, [formatted, social]) == [
        formatted, social,
    ]
    client.fetch_json.assert_not_awaited()


async def test_insert_posts_writes_embedding_arrays() -> None:
    client = _InsertClient()

    class _TupleEmbedder:
        model_name = "test-model"

        def embed_texts(self, items: list[str]) -> list[tuple[float, ...]]:
            return [(float(index),) for index, _ in enumerate(items, start=1)]

    def _converter(_raw: Any) -> Post:
        return Post(
            domain="x.com",
            post_id="1",
            url="https://x.com/i/status/1",
            created_at=datetime.datetime(2026, 2, 1, tzinfo=datetime.UTC),
            updated_at=datetime.datetime(2026, 2, 1, tzinfo=datetime.UTC),
            author="alice",
            title="title",
            title_embedding=(),
            html="html",
            html_embedding=(),
            source="x",
        )

    converted = await process_source_posts(
        cast(Any, client),
        "x.com",
        _converter,
        since=datetime.datetime(2026, 2, 1, tzinfo=datetime.UTC),
        embedder=cast(Any, _TupleEmbedder()),
    )

    assert converted == 1
    assert client.table == "posts"
    assert client.rows
    assert client.rows[0]["title_embedding"] == [1.0]
    assert client.rows[0]["html_embedding"] == [1.0]


async def test_process_source_assets_only_selects_unprocessed_assets() -> None:
    client = _DummyAssetClient()
    since = datetime.datetime.now(datetime.UTC)

    converted = await process_source_assets(
        cast(Any, client),
        "x.com",
        lambda _raw: None,
        since=since,
    )

    assert converted == 0
    assert "LIMIT 1 BY post_id, url" in client.query
    assert "LEFT JOIN" in client.query
    assert "raw.scraped_at > processed.updated_at" in client.query


@pytest.mark.parametrize(("feed_author", "expected"), [
    ("", "Known Writer"),
    ("New Writer", "New Writer"),
])
async def test_reprocessing_retains_author_when_new_capture_omits_it(
    feed_author: str, expected: str,
) -> None:
    class Client(_InsertClient):
        async def fetch_json(
            self, query: str, params: dict[str, object] | None = None,
        ) -> _Result:
            result = await super().fetch_json(query, params)
            result.rows[0].update({
                "payload": f"<item><title>Article</title>"
                f"<author>{feed_author}</author>"
                "<description>Feed text.</description></item>",
                "previous_author": "Known Writer",
            })

            return result

    client = Client()
    await process_source_posts(
        cast(Any, client), "publisher.test", convert_raw_post,
        since=datetime.datetime(2026, 2, 1, tzinfo=datetime.UTC),
        embedder=cast(Any, _DummyEmbedder()), reprocess=True,
    )

    assert client.rows[0]["author"] == expected
    assert "FROM posts FINAL" in client.query
    assert client.rows[0]["updated_at"] == datetime.datetime(
        2026, 2, 1, tzinfo=datetime.UTC,
    )


async def test_embeddings_use_readable_text_instead_of_page_code() -> None:
    client = _InsertClient()
    calls: list[list[str]] = []

    class _SpyEmbedder:
        model_name = "test-model"

        def embed_texts(self, items: list[str]) -> list[tuple[float, ...]]:
            calls.append(items)

            return [(1.0,) for _ in items]

    def converter(_raw: Any) -> Post:
        now = datetime.datetime.now(datetime.UTC)

        return Post(
            domain="example.com", post_id="1", url="https://example.com/1",
            created_at=now, updated_at=now, author="author", title="Title",
            title_embedding=(), html_embedding=(), source="rss",
            html="<head><style>tracking css</style></head>"
                 "<p>Readable <strong>article</strong>.</p>",
        )

    await process_source_posts(
        cast(Any, client), "example.com", converter,
        since=datetime.datetime.now(datetime.UTC),
        embedder=cast(Any, _SpyEmbedder()),
    )
    assert calls == [["Title"], ["Readable article."]]


@pytest.mark.parametrize("domain, expected", [
    ("reddit.com", {"reddit-publisher.test"}),
    ("news.ycombinator.com", {"hn-publisher.test", "legacy.test"}),
    ("other.test", set[str]()),
])
async def test_source_processing_includes_its_linked_publishers_only(
    domain: str, expected: set[str],
) -> None:
    from dank.process.runner import (
        _with_linked_article_sources,  # pyright: ignore[reportPrivateUsage]
    )

    client = AsyncMock()
    client.fetch_json.return_value = _Result([
        {"domain": "reddit-publisher.test", "origin_domain": "reddit.com"},
        {"domain": "hn-publisher.test",
         "origin_domain": "news.ycombinator.com"},
        {"domain": "legacy.test", "origin_domain": ""},
    ])
    sources = await _with_linked_article_sources(
        client, (SourceConfig(domain, ()),),
        datetime.datetime(2026, 10, 19, tzinfo=datetime.UTC),
        re.compile("^" + re.escape(domain) + "$"),
    )
    assert {item.domain for item in sources} == {domain, *expected}


async def test_context_process_uses_collected_posts_and_linked_domains(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    settings = MagicMock()
    settings.sources = (SourceConfig("reddit.com", ()),)
    since = datetime.datetime(2026, 10, 22, tzinfo=datetime.UTC)
    posts = AsyncMock(return_value=1)
    assets = AsyncMock(return_value=0)
    linked_sources = AsyncMock()
    monkeypatch.setattr(runner, "process_source_posts", posts)
    monkeypatch.setattr(runner, "process_source_assets", assets)
    monkeypatch.setattr(runner, "_with_linked_article_sources", linked_sources)
    monkeypatch.setattr(runner, "ClickHouseClient", MagicMock())

    assert await runner.run_process(settings, since=since, post_keys={
        ("reddit.com", "selected-post"), ("publisher.test", "linked-story"),
    }) == 2
    assert {call.args[1]: call.kwargs["post_ids"]
            for call in posts.call_args_list} == {
        "reddit.com": ("selected-post",),
        "publisher.test": ("linked-story",),
    }

    for call in posts.call_args_list + assets.call_args_list:
        assert call.kwargs["since"] == since

    linked_sources.assert_not_awaited()


async def test_empty_context_process_never_loads_model_or_database(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    model = MagicMock()
    client = MagicMock()
    monkeypatch.setattr(runner, "get_embedding_model", model)
    monkeypatch.setattr(runner, "ClickHouseClient", client)

    assert await runner.run_process(MagicMock(), post_keys=set()) == 0
    model.assert_not_called()
    client.assert_not_called()


async def test_context_queries_filter_posts_and_assets_by_exact_ids() -> None:
    since = datetime.datetime(2026, 10, 22, tzinfo=datetime.UTC)
    client = _DummyClient()
    await process_source_posts(
        cast(Any, client), "publisher.test", convert_raw_post,
        since=since, embedder=cast(Any, _DummyEmbedder()), post_ids=("one",),
    )
    assert "AND post_id IN %(post_ids)s " in client.query
    assert client.params["post_ids"] == ["one"]

    await process_source_assets(
        cast(Any, client), "publisher.test", runner.convert_raw_asset,
        since=since, post_ids=("one",),
    )
    assert "AND post_id IN %(post_ids)s " in client.query
    assert client.params["post_ids"] == ["one"]