tests/test_qdrant_keepalive.py

import asyncio

import pytest

from app.main import _qdrant_keepalive_loop


class _FakeQdrant:
    def __init__(self) -> None:
        self.calls = 0

    def get_collections(self) -> None:
        self.calls += 1


@pytest.mark.asyncio
async def test_keepalive_loop_pings_qdrant() -> None:
    qdrant = _FakeQdrant()
    stop_event = asyncio.Event()

    task = asyncio.create_task(
        _qdrant_keepalive_loop(qdrant=qdrant, interval_seconds=1, stop_event=stop_event)
    )

    await asyncio.sleep(1.2)
    stop_event.set()
    await asyncio.wait_for(task, timeout=1)

    assert qdrant.calls >= 1


@pytest.mark.asyncio
async def test_keepalive_loop_disabled_when_interval_non_positive() -> None:
    qdrant = _FakeQdrant()
    stop_event = asyncio.Event()

    await _qdrant_keepalive_loop(qdrant=qdrant, interval_seconds=0, stop_event=stop_event)

    assert qdrant.calls == 0