conjurer/tests/integration/test_librarian_ping.py

"""Integration: the librarian health check is a full comm round-trip, not a GET.

``librarian_ping`` injects a pseudo-query into the SAME machinery a real search
uses - it rides ``OUT_COMM_Q`` -> ``scan_queue`` -> ``awaiting_q``, the librarian
is expected to pull it off its own queue and pong it back over ``/conjurer`` ->
``incoming_q`` -> ``scan_incoming``, which matches it by uuid and wakes the
waiter. These tests stand in for the librarian with a stubbed ``requests.post``
and assert the loop closes (and, crucially, that a pong never leaks into
``IN_COMM_Q`` where the librarian cog would mistake it for a real result).
"""
import threading
import time

import pytest

import communication_subroutine as cs


def _drain(queue):
    while not queue.empty():
        queue.get()


@pytest.fixture
def comm_threads():
    """Run scan_queue + scan_incoming (the two workers librarian_ping relies on)
    for the duration of a test, on cleared shared state."""
    cs.awaiting_q.clear()
    _drain(cs.incoming_q)
    _drain(cs.OUT_COMM_Q)
    _drain(cs.IN_COMM_Q)
    stop = threading.Event()
    workers = [
        threading.Thread(target=cs.scan_queue, kwargs={"stop_event": stop}, daemon=True),
        threading.Thread(target=cs.scan_incoming, kwargs={"stop_event": stop}, daemon=True),
    ]
    for worker in workers:
        worker.start()
    yield
    stop.set()
    for worker in workers:
        worker.join(timeout=3)


def _await_in_awaiting(ping_uuid, timeout=2):
    """Block until scan_queue has moved the ping into awaiting_q."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        if any(getattr(r, "uuid", None) == ping_uuid for r in list(cs.awaiting_q)):
            return True
        time.sleep(0.01)
    return False


class _Resp:
    def __init__(self, status_code=200):
        self.status_code = status_code


def test_ping_round_trip_ok(comm_threads, monkeypatch):
    # Stub librarian: only pong AFTER the record is in awaiting_q, mirroring the
    # real network latency that always lets scan_queue win.
    def fake_post(url, json=None, headers=None, timeout=None):
        ping_uuid = json["UUID"]
        _await_in_awaiting(ping_uuid)
        cs.incoming_q.put({"__pong__": ping_uuid})
        return _Resp(200)

    monkeypatch.setattr(cs.requests, "post", fake_post)

    assert cs.librarian_ping("http://lib", "/ping", {}, timeout=3.0) is True
    # A pong must NEVER reach the cog's inbound queue...
    assert cs.IN_COMM_Q.empty()
    # ...and the ping record must be cleaned out of awaiting_q.
    assert not any(getattr(r, "is_ping", False) for r in list(cs.awaiting_q))


def test_ping_times_out_when_librarian_accepts_but_never_pongs(comm_threads, monkeypatch):
    monkeypatch.setattr(cs.requests, "post", lambda *a, **k: _Resp(200))
    start = time.monotonic()
    assert cs.librarian_ping("http://lib", "/ping", {}, timeout=0.3) is False
    # Bounded: it must not block much beyond the timeout.
    assert time.monotonic() - start < 2.0


def test_ping_false_when_librarian_unreachable(comm_threads, monkeypatch):
    def boom(*a, **k):
        raise cs.requests.exceptions.RequestException("no route to host")

    monkeypatch.setattr(cs.requests, "post", boom)
    assert cs.librarian_ping("http://lib", "/ping", {}, timeout=0.3) is False


def test_ping_false_on_non_200(comm_threads, monkeypatch):
    monkeypatch.setattr(cs.requests, "post", lambda *a, **k: _Resp(503))
    assert cs.librarian_ping("http://lib", "/ping", {}, timeout=0.3) is False


def test_orphan_pong_is_dropped_not_enqueued(comm_threads):
    # A pong with no matching waiter (e.g. after the ping already timed out) must
    # be silently dropped - never turned into an "Orphaned" IN_COMM_Q record that
    # makes the librarian cog post a bogus "no results" message.
    cs.incoming_q.put({"__pong__": "no-such-uuid"})
    deadline = time.time() + 3
    while time.time() < deadline and not cs.incoming_q.empty():
        time.sleep(0.02)
    time.sleep(0.2)  # give scan_incoming a beat to (not) enqueue anything
    assert cs.IN_COMM_Q.empty()


def test_real_result_still_reaches_in_comm_q(comm_threads):
    # Guard the existing path: a normal {uuid: {...}} result must still match its
    # QueryControl and land in IN_COMM_Q for the cog to render.
    query = cs.QueryControl("user", "real-uuid", "jakieś zapytanie", None)
    cs.OUT_COMM_Q.put(query)
    assert _await_in_awaiting("real-uuid")
    cs.incoming_q.put({"real-uuid": {"10.1/x": {"Title": ["Tytuł"], "type": "article"}}})
    got = cs.IN_COMM_Q.get(timeout=3)
    assert got.uuid == "real-uuid"
    assert got.stop is True
    assert "10.1/x" in got.entries