tests/test_analysis_queue.py

"""The bounded analysis pool: cap, ordering, position and ETA.

The pool is what turns a launch-day burst from "everything fell over" into
"you're 4th in line, ~10 min". These lock its behaviour.
"""
import os
import sys
import threading
import time

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from backend.analysis_queue import AnalysisPool


def _blocking_job(started, release, done):
    def run():
        started.set()
        release.wait(timeout=5)
        done.set()
    return run


def test_pool_caps_concurrency_at_worker_count():
    pool = AnalysisPool(workers=2)
    peak = {"n": 0}
    lock = threading.Lock()
    live = {"n": 0}
    release = threading.Event()

    def job():
        with lock:
            live["n"] += 1
            peak["n"] = max(peak["n"], live["n"])
        release.wait(timeout=5)
        with lock:
            live["n"] -= 1

    for i in range(6):
        pool.submit(f"job-{i}", job)

    time.sleep(0.4)                 # let the workers pick up
    assert peak["n"] <= 2, "never more than `workers` analyses run at once"
    assert peak["n"] == 2, "both workers should be busy under load"
    release.set()


def test_waiting_job_reports_its_place_in_line():
    pool = AnalysisPool(workers=1)
    release = threading.Event()
    first_started = threading.Event()

    def first():
        first_started.set()
        release.wait(timeout=5)

    def noop():
        release.wait(timeout=5)

    pool.submit("first", first)
    first_started.wait(timeout=5)   # occupies the single worker

    pool.submit("second", noop)
    pool.submit("third", noop)
    time.sleep(0.2)

    assert pool.position("first") == 0       # running
    assert pool.position("second") == 1      # next
    assert pool.position("third") == 2       # after that
    assert pool.position("unknown") is None
    release.set()


def test_finished_job_is_no_longer_in_the_pool():
    pool = AnalysisPool(workers=1)
    done = threading.Event()
    pool.submit("solo", lambda: done.set())
    assert done.wait(timeout=5)
    time.sleep(0.1)
    assert pool.position("solo") is None


def test_eta_grows_with_queue_depth():
    pool = AnalysisPool(workers=2)
    release = threading.Event()
    started = threading.Event()

    def hold():
        started.set()
        release.wait(timeout=5)

    for i in range(6):
        pool.submit(f"j{i}", hold)
    started.wait(timeout=5)
    time.sleep(0.2)

    etas = [pool.eta_seconds(f"j{i}") for i in range(6)]
    etas = [e for e in etas if e is not None]
    # A job deeper in the queue must never show a shorter wait than one ahead.
    assert etas == sorted(etas), f"ETA must be monotonic by depth, got {etas}"
    release.set()


def test_snapshot_counts_running_and_queued():
    pool = AnalysisPool(workers=2)
    release = threading.Event()
    started = threading.Semaphore(0)

    def hold():
        started.release()
        release.wait(timeout=5)

    for i in range(5):
        pool.submit(f"s{i}", hold)
    started.acquire(timeout=5)
    started.acquire(timeout=5)      # two are running
    time.sleep(0.2)

    snap = pool.snapshot()
    assert snap["workers"] == 2
    assert snap["running"] == 2
    assert snap["queued"] == 3
    release.set()


def test_a_crashing_job_frees_its_slot():
    pool = AnalysisPool(workers=1)
    ran = threading.Event()

    def boom():
        raise RuntimeError("analysis blew up")

    def after():
        ran.set()

    pool.submit("boom", boom)
    pool.submit("after", after)
    # If the crash leaked the worker, `after` would never run.
    assert ran.wait(timeout=5), "a crashed job must not wedge the worker"