changedetection.io/changedetectionio/tests/test_queue_ui.py

"""
Tests for the queue inspection UI (blueprint/ui/queue.py).

Strategy: shut down all workers, pause the scheduler, then push synthetic
UUIDs directly into update_q. Synthetic UUIDs are immune to scheduler
interference. The HTTP test client + live_server share the same process,
so update_q is a single instance.
"""

import time
from flask import url_for


def _quiesce(live_server, timeout=3.0):
    """Pause scheduler, shut down all workers, wait for in-flight async workers
    to fully drain, clear queue. Returns (update_q, worker_pool, queuedWatchMetaData).

    The "brutal" shutdown registers stop-flags but workers blocked in
    update_q.async_get() can still complete one more pop after shutdown is
    requested. We poll for a quiet period where consecutive clear()s show no
    new items being deposited.
    """
    from changedetectionio import worker_pool, queuedWatchMetaData
    from changedetectionio.flask_app import update_q

    live_server.app.config['DATASTORE'].data['settings']['application']['all_paused'] = True
    live_server.app.set_workers(0)

    deadline = time.time() + timeout
    while time.time() < deadline:
        if worker_pool.get_worker_count() == 0:
            break
        time.sleep(0.05)
    assert worker_pool.get_worker_count() == 0, "workers did not shut down in time"

    # Push a sentinel and watch for it to remain in the queue across consecutive reads.
    # If a lingering coroutine pops it, we requeue and wait. Once it survives a settle
    # window the queue is genuinely quiescent.
    sentinel_uuid = "00000000-0000-0000-0000-0000sent1nel0"
    sentinel = queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': sentinel_uuid})

    update_q.clear()
    settle_deadline = time.time() + timeout
    while time.time() < settle_deadline:
        worker_pool.queue_item_async_safe(update_q, sentinel)
        time.sleep(0.15)
        if sentinel_uuid in update_q.get_queued_uuids():
            break
        # Was popped by a lingering worker — try again
    else:
        raise AssertionError("Could not get queue to a quiescent state")

    update_q.clear()
    return update_q, worker_pool, queuedWatchMetaData


def _push(update_q, worker_pool, qmeta, uuid, priority=1):
    ok = worker_pool.queue_item_async_safe(
        update_q,
        qmeta.PrioritizedItem(priority=priority, item={'uuid': uuid}),
    )
    assert ok, f"queue_item_async_safe returned falsy for {uuid}"


def test_queue_page_renders_when_empty(client, live_server, measure_memory_usage, datastore_path):
    """Empty queue: HTML 200, JSON shape matches contract."""
    _quiesce(live_server)

    res = client.get(url_for("ui.ui_queue.queue_page"))
    assert res.status_code == 200
    assert b"Check queue" in res.data
    # Single combined "Workers & queue" table — replaces the old separate Running/Queue tables.
    assert b"Workers" in res.data and b"queue" in res.data
    assert b"Clear Queue" in res.data
    assert b"Re-check Errored" not in res.data

    data = client.get(url_for("ui.ui_queue.queue_json")).get_json()
    for key in ("worker_count", "running_count", "queued_count", "summary", "running", "queued"):
        assert key in data
    for sub in ("immediate", "clone", "scheduled", "priority_breakdown"):
        assert sub in data["summary"]
    assert data["queued_count"] == 0
    assert data["running_count"] == 0


def test_queue_state_full_lifecycle(client, live_server, measure_memory_usage, datastore_path):
    """One consolidated test of pushed items, priority labels, deleted-watch handling, and Clear Queue.

    Combined into one test deliberately — having separate tests revealed cross-test state leakage
    (workers occasionally surviving teardown) that this single test avoids.
    """
    update_q, worker_pool, qmeta = _quiesce(live_server)

    running_uuid = "11111111-1111-1111-1111-111111111111"
    immediate_uuid = "22222222-2222-2222-2222-222222222222"
    clone_uuid = "33333333-3333-3333-3333-333333333333"
    scheduled_uuid = "44444444-4444-4444-4444-444444444444"

    # Mark one synthetic UUID as currently running.
    assert worker_pool.claim_uuid_for_processing(running_uuid, worker_id=999) is True

    # Push at three different priority classes.
    _push(update_q, worker_pool, qmeta, immediate_uuid, priority=1)
    _push(update_q, worker_pool, qmeta, clone_uuid, priority=5)
    _push(update_q, worker_pool, qmeta, scheduled_uuid, priority=99999)

    # Local sanity check: items are actually in the queue object before the HTTP read.
    queued_local = update_q.get_queued_uuids()
    assert set(queued_local) == {immediate_uuid, clone_uuid, scheduled_uuid}, \
        f"items missing from update_q after push: {queued_local}"

    try:
        # --- JSON view shows pushed state ---
        data = client.get(url_for("ui.ui_queue.queue_json")).get_json()
        assert data["worker_count"] == 0
        assert data["running_count"] == 1
        assert data["queued_count"] == 3, f"unexpected queue state: {data!r}"
        assert data["running"][0]["uuid"] == running_uuid
        # Synthetic UUIDs aren't in the datastore — should be rendered as 'gone'
        assert data["running"][0]["gone"] is True

        # --- Priority labels map correctly ---
        labels = {entry["uuid"]: entry["priority_label"] for entry in data["queued"]}
        assert labels[immediate_uuid] == "immediate"
        assert labels[clone_uuid] == "clone"
        assert labels[scheduled_uuid] == "scheduled"
        assert data["summary"]["immediate"] == 1
        assert data["summary"]["clone"] == 1
        assert data["summary"]["scheduled"] == 1

        # --- Deleted-watch placeholder ---
        for entry in data["queued"]:
            assert entry["gone"] is True
            assert entry.get("title") is None
            assert entry.get("url") is None

        # --- HTML view also surfaces the queued UUIDs ---
        html = client.get(url_for("ui.ui_queue.queue_page")).data
        for uuid in (immediate_uuid, clone_uuid, scheduled_uuid):
            assert uuid.encode("utf-8") in html

        # --- POST /queue/clear empties pending ---
        res = client.post(
            url_for("ui.ui_queue.queue_clear"),
            data={"csrf_token": "test"},
            follow_redirects=True,
        )
        assert res.status_code == 200
        assert b"Queue cleared" in res.data

        after = client.get(url_for("ui.ui_queue.queue_json")).get_json()
        assert after["queued_count"] == 0, f"queue should be empty after clear, got {after!r}"
        # The synthetic 'running' claim is untouched by Clear Queue — running checks aren't queue items.
        assert after["running_count"] == 1

    finally:
        worker_pool.release_uuid_from_processing(running_uuid, worker_id=999)
        update_q.clear()


def test_queue_includes_timestamps(client, live_server, measure_memory_usage, datastore_path):
    """Queue items carry enqueued_at; running items carry started_at."""
    update_q, worker_pool, qmeta = _quiesce(live_server)

    # Push something — enqueued_at should be ~now
    before = time.time()
    queued_uuid = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
    _push(update_q, worker_pool, qmeta, queued_uuid, priority=1)
    after = time.time()

    snap = client.get(url_for("ui.ui_queue.queue_json")).get_json()
    assert len(snap["queued"]) == 1
    ts_q = snap["queued"][0]["enqueued_at"]
    assert isinstance(ts_q, float)
    assert before <= ts_q <= after + 0.5, f"enqueued_at out of range: {ts_q!r}"

    # Claim a synthetic running uuid — started_at should be ~now
    running_uuid = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
    before = time.time()
    assert worker_pool.claim_uuid_for_processing(running_uuid, worker_id=999)
    after = time.time()

    snap = client.get(url_for("ui.ui_queue.queue_json")).get_json()
    running_entry = next((r for r in snap["running"] if r["uuid"] == running_uuid), None)
    assert running_entry is not None
    ts_r = running_entry["started_at"]
    assert isinstance(ts_r, float)
    assert before <= ts_r <= after + 0.5, f"started_at out of range: {ts_r!r}"

    # Release clears the timestamp
    worker_pool.release_uuid_from_processing(running_uuid, worker_id=999)
    assert worker_pool.get_uuid_started_at(running_uuid) is None

    update_q.clear()


def test_queue_pagination(client, live_server, measure_memory_usage, datastore_path):
    """Page math: per_page from settings.pager_size, total_pages, page slicing."""
    update_q, worker_pool, qmeta = _quiesce(live_server)

    # Override pager_size for this test so we don't have to push 50+ items
    live_server.app.config['DATASTORE'].data['settings']['application']['pager_size'] = 2

    uuids = [f"{i:08d}-2222-3333-4444-555555555555" for i in range(5)]
    for u in uuids:
        _push(update_q, worker_pool, qmeta, u, priority=1)

    snap1 = client.get(url_for("ui.ui_queue.queue_json")).get_json()
    assert snap1["per_page"] == 2
    assert snap1["queued_count"] == 5
    assert snap1["total_pages"] == 3
    assert snap1["page"] == 1
    assert len(snap1["queued"]) == 2

    snap2 = client.get(url_for("ui.ui_queue.queue_json") + "?page=2").get_json()
    assert snap2["page"] == 2
    assert len(snap2["queued"]) == 2

    snap3 = client.get(url_for("ui.ui_queue.queue_json") + "?page=3").get_json()
    assert snap3["page"] == 3
    assert len(snap3["queued"]) == 1, f"last page should be partial: {snap3!r}"

    # Pages cover every uuid exactly once
    all_uuids = [w["uuid"] for w in snap1["queued"] + snap2["queued"] + snap3["queued"]]
    assert sorted(all_uuids) == sorted(uuids)

    # Restore + drain
    live_server.app.config['DATASTORE'].data['settings']['application']['pager_size'] = 50
    update_q.clear()


def test_cancel_running_uuid_helper(client, live_server, measure_memory_usage, datastore_path):
    """worker_pool.cancel_running_uuid() drops tracking and returns cancelled flag."""
    update_q, worker_pool, qmeta = _quiesce(live_server)

    synth = "cccccccc-cccc-cccc-cccc-cccccccccccc"
    assert worker_pool.claim_uuid_for_processing(synth, worker_id=999) is True
    assert worker_pool.get_uuid_started_at(synth) is not None

    # No worker_id=999 actually exists in worker_threads — the stop/remove path is
    # a no-op — but tracking still gets dropped.
    result = worker_pool.cancel_running_uuid(synth)
    assert result["cancelled"] is True
    assert result["worker_id"] == 999
    assert result["replaced"] is False  # no replacement params provided

    assert synth not in worker_pool.get_running_uuids()
    assert worker_pool.get_uuid_started_at(synth) is None

    # Cancelling something that was never running returns cancelled=False
    result2 = worker_pool.cancel_running_uuid("never-claimed")
    assert result2["cancelled"] is False
    assert result2["worker_id"] is None


def test_cancel_running_endpoint(client, live_server, measure_memory_usage, datastore_path):
    """POST /queue/cancel-running: 200 on success, 400 on empty uuid, 404 on unknown uuid."""
    update_q, worker_pool, qmeta = _quiesce(live_server)

    synth = "dddddddd-dddd-dddd-dddd-dddddddddddd"
    assert worker_pool.claim_uuid_for_processing(synth, worker_id=999) is True

    # Happy path
    res = client.post(
        url_for("ui.ui_queue.queue_cancel_running"),
        data={"uuid": synth, "csrf_token": "test"},
    )
    assert res.status_code == 200
    body = res.get_json()
    assert body["ok"] is True
    assert body["cancelled"] is True
    assert body["worker_id"] == 999

    # No longer in running per the snapshot
    snap = client.get(url_for("ui.ui_queue.queue_json")).get_json()
    assert all(r["uuid"] != synth for r in snap["running"])

    # Empty uuid → 400
    res = client.post(url_for("ui.ui_queue.queue_cancel_running"), data={"uuid": ""})
    assert res.status_code == 400

    # Unknown uuid → 404
    res = client.post(url_for("ui.ui_queue.queue_cancel_running"), data={"uuid": "never-claimed-anywhere"})
    assert res.status_code == 404

    # cancel_running_uuid() spawned a replacement worker via the route (update_q
    # etc. were available). Tear it down so the next test starts at workers=0.
    live_server.app.set_workers(0)