changedetection.io/changedetectionio/tests/test_llm_summary_async.py

#!/usr/bin/env python3
"""
The /llm-summary route is start-then-wait, not a blocking request.

Generating a summary is an LLM round-trip, so the route no longer holds a request thread for it:
POST starts the work on a small pool and answers 202, and a read-only GET picks the result up out
of the watch's summary cache - driven either by the llm_summary_ready event (realtime) or by
polling (no realtime). These tests pin the properties that make that safe:

  - POST answers immediately instead of waiting for the LLM
  - a second click (or a poll) never starts a duplicate generation - that would bill twice
  - the GET is genuinely read-only, so it is safe to leave it as a GET
  - completion announces itself, since a realtime client makes exactly one request per event
  - every 'pending' reply carries the deadline the server itself is working to, because a client
    that gives up earlier reports a failure for a job that is still running
  - that deadline belongs to one job, not to the page: the watchlist has a [Summary] button per
    row and each one is its own request, started at its own moment

plus the CSRF protection on the POST, which is the reason starting generation is not a GET in the
first place: a bare <img src="/diff/<uuid>/llm-summary"> on any page would otherwise spend the
operator's tokens and mark the watch as viewed.
"""

import threading
import time

from unittest.mock import MagicMock, patch

from flask import url_for

from changedetectionio.tests.util import delete_all_watches, fetch_llm_summary


def _configure_llm(client, api_base=None):
    ds = client.application.config.get('DATASTORE')
    existing = ds.data['settings']['application'].get('llm') or {}
    existing.update({'model': 'gpt-4o-mini', 'api_key': 'sk-test'})
    if api_base is not None:
        existing['api_base'] = api_base
    ds.data['settings']['application']['llm'] = existing


def _watch_with_history(client, from_ts, to_ts):
    ds = client.application.config.get('DATASTORE')
    test_url = url_for('test_endpoint', content_type='text/html', content='v1', _external=True)
    uuid = ds.add_watch(url=test_url)
    watch = ds.data['watching'][uuid]
    watch.save_history_blob('old content\n', from_ts, 'snap-old')
    watch.save_history_blob('new content\n', to_ts, 'snap-new')
    return uuid, watch


def _llm_response(text='Content changed from old to new.'):
    mock_response = MagicMock()
    mock_response.choices = [MagicMock()]
    mock_response.choices[0].message.content = text
    mock_response.usage = MagicMock(total_tokens=50, prompt_tokens=40, completion_tokens=10)
    return mock_response


def test_post_returns_pending_then_poll_returns_summary(
        client, live_server, measure_memory_usage, datastore_path):
    """The POST must not block on the LLM: it answers 202, the summary arrives via the poll."""
    _configure_llm(client)
    uuid, watch = _watch_with_history(client, '5000000000', '5000000001')
    url = url_for('ui.ui_diff.diff_llm_summary', uuid=uuid,
                  from_version='5000000000', to_version='5000000001')

    released = threading.Event()

    def slow_completion(*args, **kwargs):
        released.wait(10)
        return _llm_response()

    with patch('litellm.completion', side_effect=slow_completion):
        start = client.post(url)
        assert start.status_code == 202, "POST should hand off to the background pool"
        assert start.get_json()['status'] == 'pending'
        assert start.get_json()['summary'] is None

        # Still generating — the poll reports pending without starting anything of its own
        pending = client.get(url)
        assert pending.status_code == 202
        assert pending.get_json()['status'] == 'pending'

        released.set()

        deadline = time.time() + 15
        while time.time() < deadline:
            res = client.get(url)
            if res.status_code != 202:
                break
            time.sleep(0.05)

    assert res.status_code == 200
    data = res.get_json()
    assert data['status'] == 'done'
    assert data['summary'] == 'Content changed from old to new.'

    delete_all_watches(client)


def test_repeated_clicks_generate_one_summary(
        client, live_server, measure_memory_usage, datastore_path):
    """Two tabs / an impatient double-click must not pay for the same summary twice."""
    _configure_llm(client)
    uuid, watch = _watch_with_history(client, '5100000000', '5100000001')
    url = url_for('ui.ui_diff.diff_llm_summary', uuid=uuid,
                  from_version='5100000000', to_version='5100000001')

    calls = {'n': 0}
    released = threading.Event()
    entered = threading.Event()

    def slow_completion(*args, **kwargs):
        calls['n'] += 1
        entered.set()
        released.wait(10)
        return _llm_response()

    try:
        with patch('litellm.completion', side_effect=slow_completion):
            first = client.post(url)
            assert first.status_code == 202
            assert entered.wait(10), "background job never reached the LLM call"

            # Extra clicks while the first generation is still running
            for _ in range(3):
                again = client.post(url)
                assert again.status_code == 202
                assert again.get_json()['status'] == 'pending'

            released.set()

            deadline = time.time() + 15
            while time.time() < deadline:
                res = client.get(url)
                if res.status_code != 202:
                    break
                time.sleep(0.05)
    finally:
        released.set()

    assert res.status_code == 200
    assert res.get_json()['summary'] == 'Content changed from old to new.'
    assert calls['n'] == 1, f"LLM was called {calls['n']} times for one summary"

    # And once cached, a further click is answered straight from disk — still no new call
    with patch('litellm.completion', side_effect=slow_completion):
        cached = client.post(url)
    assert cached.status_code == 200
    assert cached.get_json()['cached'] is True
    assert calls['n'] == 1

    delete_all_watches(client)


def test_poll_alone_never_starts_generation(
        client, live_server, measure_memory_usage, datastore_path):
    """The GET is read-only — that is what makes it safe to leave CSRF-exempt.

    With nothing cached and no job running it reports 'idle' so the client re-POSTs, rather than
    quietly spending tokens on a request that carried no CSRF token.
    """
    _configure_llm(client)
    uuid, watch = _watch_with_history(client, '5200000000', '5200000001')
    url = url_for('ui.ui_diff.diff_llm_summary', uuid=uuid,
                  from_version='5200000000', to_version='5200000001')

    calls = {'n': 0}

    def counted(*args, **kwargs):
        calls['n'] += 1
        return _llm_response()

    with patch('litellm.completion', side_effect=counted):
        for _ in range(3):
            res = client.get(url)
            assert res.status_code == 200
            assert res.get_json()['status'] == 'idle'
            assert res.get_json()['summary'] is None

    assert calls['n'] == 0, "a GET poll must never trigger an LLM call"
    assert watch['last_viewed'] == 0, "a GET poll must not mark the watch as viewed either"

    delete_all_watches(client)


def test_starting_generation_requires_a_csrf_token(
        client, live_server, measure_memory_usage, datastore_path):
    """Starting a generation spends tokens, so it must be CSRF-protected.

    conftest disables CSRF for the test client, so switch it back on for this one check to prove
    the route is actually covered by CSRFProtect (i.e. it was not exempted the way the JSON API is
    via `csrf.exempt`).
    """
    _configure_llm(client)
    uuid, watch = _watch_with_history(client, '5300000000', '5300000001')
    url = url_for('ui.ui_diff.diff_llm_summary', uuid=uuid,
                  from_version='5300000000', to_version='5300000001')

    calls = {'n': 0}

    def counted(*args, **kwargs):
        calls['n'] += 1
        return _llm_response()

    app = client.application
    app.config['WTF_CSRF_ENABLED'] = True
    try:
        with patch('litellm.completion', side_effect=counted):
            res = client.post(url)          # no token supplied
            assert res.status_code in (400, 403), \
                f"POST without a CSRF token was accepted ({res.status_code})"

            # The read-only poll is unaffected
            poll = client.get(url)
            assert poll.status_code == 200
            assert poll.get_json()['status'] == 'idle'
    finally:
        app.config['WTF_CSRF_ENABLED'] = False

    assert calls['n'] == 0, "a request rejected for CSRF must not have reached the LLM"

    delete_all_watches(client)


def test_a_job_finishing_mid_request_is_not_reported_as_idle(
        client, live_server, measure_memory_usage, datastore_path):
    """Regression: the poll must re-read the cache before it declares nothing is running.

    A job writes the summary and *then* clears its pending flag, so a request that reads the cache
    (miss), then checks the flag a moment later (clear), would announce 'idle' for a summary that
    is already on disk - and send the client back to POST, which is the request that spends money.
    Simulated by making the first cache read miss and the second one hit.
    """
    _configure_llm(client)
    uuid, watch = _watch_with_history(client, '5600000000', '5600000001')
    url = url_for('ui.ui_diff.diff_llm_summary', uuid=uuid,
                  from_version='5600000000', to_version='5600000001')

    with patch.object(type(watch), 'get_llm_diff_summary',
                      side_effect=['', 'Summary that landed mid-request.']):
        res = client.get(url)

    assert res.status_code == 200
    data = res.get_json()
    assert data['status'] == 'done', f"poll reported {data['status']!r} for a cached summary"
    assert data['summary'] == 'Summary that landed mid-request.'

    delete_all_watches(client)


def test_pending_reply_carries_the_servers_own_deadline(
        client, live_server, measure_memory_usage, datastore_path):
    """A 'pending' reply must say when the summary stops being worth waiting for.

    The browser cannot derive this: it knows neither the configured per-call timeout nor when the
    job started (another tab may have started it minutes ago). It used to guess 180s, which is
    shorter than every configured timeout here - so a slow model produced "AI summary timed out"
    on screen while the job was still running and still about to write its summary.
    """
    from changedetectionio.llm import client as llm_client

    _configure_llm(client)
    uuid, watch = _watch_with_history(client, '5600000000', '5600000001')
    url = url_for('ui.ui_diff.diff_llm_summary', uuid=uuid,
                  from_version='5600000000', to_version='5600000001')

    released = threading.Event()
    try:
        with patch('litellm.completion', side_effect=lambda *a, **k: (released.wait(10),
                                                                      _llm_response())[1]):
            start = client.post(url)
            assert start.status_code == 202
            body = start.get_json()

            assert body['expires_in'] >= llm_client.DEFAULT_TIMEOUT, \
                f"client told to give up after {body['expires_in']}s, before the server would"
            assert body['timeout_at'] > time.time()

            # The poll reports the same deadline, not one restarted from when it was asked -
            # otherwise the wait could be extended indefinitely by polling.
            poll = client.get(url)
            assert poll.status_code == 202
            assert abs(poll.get_json()['timeout_at'] - body['timeout_at']) <= 1, \
                "each poll must report the same instant, anchored to the job's start"
    finally:
        released.set()

    delete_all_watches(client)


def test_local_endpoint_gets_the_long_deadline(
        client, live_server, measure_memory_usage, datastore_path):
    """A local Ollama/vLLM box gets 30 minutes server-side - the browser must be told so.

    This is the case that produced the bug report: LLM_LOCAL_TIMEOUT is 1800s, so a client capped
    at 180s gave up ten times too early while Ollama was still churning on CPU.
    """
    from changedetectionio.llm import client as llm_client

    _configure_llm(client, api_base='http://127.0.0.1:11434')
    uuid, watch = _watch_with_history(client, '5700000000', '5700000001')
    url = url_for('ui.ui_diff.diff_llm_summary', uuid=uuid,
                  from_version='5700000000', to_version='5700000001')

    released = threading.Event()
    try:
        with patch('litellm.completion', side_effect=lambda *a, **k: (released.wait(10),
                                                                      _llm_response())[1]):
            body = client.post(url).get_json()
            assert body['expires_in'] >= llm_client.DEFAULT_LOCAL_TIMEOUT, \
                f"local endpoint deadline was only {body['expires_in']}s"
    finally:
        released.set()
        ds = client.application.config.get('DATASTORE')
        ds.data['settings']['application']['llm'].pop('api_base', None)

    delete_all_watches(client)


def test_each_watch_gets_its_own_deadline(
        client, live_server, measure_memory_usage, datastore_path):
    """The watchlist has a [Summary] button per row, so a deadline is per job, never per page.

    Rows are clicked at different moments and the pool is small, so at any instant one row can be
    generating while another is still queued behind it. Sharing a deadline across the UI would
    expire a row that had only just been submitted.
    """
    _configure_llm(client)
    first_uuid, _ = _watch_with_history(client, '5800000000', '5800000001')
    second_uuid, _ = _watch_with_history(client, '5800000000', '5800000001')
    first_url = url_for('ui.ui_diff.diff_llm_summary', uuid=first_uuid,
                        from_version='5800000000', to_version='5800000001')
    second_url = url_for('ui.ui_diff.diff_llm_summary', uuid=second_uuid,
                         from_version='5800000000', to_version='5800000001')

    released = threading.Event()
    try:
        with patch('litellm.completion', side_effect=lambda *a, **k: (released.wait(20),
                                                                      _llm_response())[1]):
            first = client.post(first_url).get_json()
            time.sleep(2)
            second = client.post(second_url).get_json()

            assert second['timeout_at'] > first['timeout_at'], \
                "the row clicked later must not inherit the earlier row's deadline"

            # ...and the earlier row keeps its own as the later one runs on.
            assert client.get(first_url).get_json()['timeout_at'] == first['timeout_at']
    finally:
        released.set()

    delete_all_watches(client)


def test_completion_emits_llm_summary_ready_signal(
        client, live_server, measure_memory_usage, datastore_path):
    """Realtime clients do not poll, so the signal the socket server re-broadcasts must fire.

    Asserted at the blinker signal rather than through a Socket.IO client: that is the seam the
    socket server subscribes to, and it stays a no-op when realtime is disabled.
    """
    from blinker import signal

    _configure_llm(client)
    uuid, watch = _watch_with_history(client, '5500000000', '5500000001')
    url = url_for('ui.ui_diff.diff_llm_summary', uuid=uuid,
                  from_version='5500000000', to_version='5500000001')

    received = []
    fired = threading.Event()

    def listener(sender, **kwargs):
        received.append(kwargs)
        fired.set()

    sig = signal('llm_summary_ready')
    sig.connect(listener, weak=False)
    try:
        with patch('litellm.completion', return_value=_llm_response()):
            assert client.post(url).status_code == 202
            assert fired.wait(15), "job finished without announcing llm_summary_ready"
    finally:
        sig.disconnect(listener)

    assert received[0]['watch_uuid'] == uuid
    assert received[0]['from_version'] == '5500000000'
    assert received[0]['to_version'] == '5500000001'

    # The announcement means "go look" - the summary must already be readable by then, because a
    # realtime client makes exactly one request in response to it.
    res = client.get(url)
    assert res.status_code == 200
    assert res.get_json()['summary'] == 'Content changed from old to new.'

    delete_all_watches(client)


def test_generation_failure_reaches_the_poller_once(
        client, live_server, measure_memory_usage, datastore_path):
    """A failure in the background thread must surface, not leave a permanent spinner."""
    import litellm as _real_litellm

    _configure_llm(client)
    uuid, watch = _watch_with_history(client, '5400000000', '5400000001')
    url = url_for('ui.ui_diff.diff_llm_summary', uuid=uuid,
                  from_version='5400000000', to_version='5400000001')

    exc = _real_litellm.AuthenticationError(
        'litellm.AuthenticationError: Invalid API key.',
        llm_provider='openai', model='gpt-4o-mini',
    )

    with patch('litellm.completion', side_effect=exc):
        res = fetch_llm_summary(client, url)

    assert res.status_code == 500
    data = res.get_json()
    assert data['summary'] is None
    assert data['error']
    assert data['status'] == 'error'

    # Delivered once: the next poll is back to 'idle' rather than repeating a stale failure
    followup = client.get(url)
    assert followup.status_code == 200
    assert followup.get_json()['status'] == 'idle'

    delete_all_watches(client)