basicmachines-co-basic-memory/tests/db/test_uvloop_policy.py

"""Unit tests for the uvloop event-loop policy installer.

Covers the structural fix for issue #831 / #877: the asyncpg engine-dispose
race ("IndexError: pop from an empty deque") is avoided by running the Postgres
backend on uvloop, whose C scheduler has no self._ready.popleft() codepath.

These tests verify the gating logic of ``maybe_install_uvloop`` without touching
a real database. The global event-loop policy is saved and restored around each
test so installing uvloop here cannot leak into the rest of the suite.
"""

import asyncio
import sys

import pytest

from basic_memory.config import BasicMemoryConfig, DatabaseBackend
from basic_memory.db import maybe_install_uvloop


@pytest.fixture
def restore_event_loop_policy():
    """Save/restore the global event-loop policy around a test."""
    original = asyncio.get_event_loop_policy()
    try:
        yield
    finally:
        asyncio.set_event_loop_policy(original)


def _postgres_config() -> BasicMemoryConfig:
    return BasicMemoryConfig(
        env="test",
        database_backend=DatabaseBackend.POSTGRES,
        database_url="postgresql+asyncpg://user:pass@localhost/db",
    )


def _sqlite_config() -> BasicMemoryConfig:
    return BasicMemoryConfig(env="test", database_backend=DatabaseBackend.SQLITE)


@pytest.mark.skipif(sys.platform == "win32", reason="uvloop is not available on Windows")
def test_installs_uvloop_for_postgres_backend(restore_event_loop_policy):
    """uvloop policy is installed when backend is Postgres and uvloop is available."""
    import uvloop

    installed = maybe_install_uvloop(_postgres_config())

    assert installed is True
    assert isinstance(asyncio.get_event_loop_policy(), uvloop.EventLoopPolicy)


def test_no_uvloop_for_sqlite_backend(restore_event_loop_policy):
    """SQLite users keep the default loop - the helper is a no-op."""
    before = asyncio.get_event_loop_policy()

    installed = maybe_install_uvloop(_sqlite_config())

    assert installed is False
    # Policy must be unchanged for the default (SQLite) path.
    assert asyncio.get_event_loop_policy() is before


@pytest.mark.skipif(sys.platform == "win32", reason="uvloop is not available on Windows")
def test_uvloop_unavailable_is_a_safe_noop(restore_event_loop_policy, monkeypatch):
    """When uvloop cannot be imported the helper returns False without raising."""
    import builtins

    real_import = builtins.__import__

    def _fail_uvloop_import(name, *args, **kwargs):
        if name == "uvloop":
            raise ImportError("simulated missing uvloop")
        return real_import(name, *args, **kwargs)

    monkeypatch.setattr(builtins, "__import__", _fail_uvloop_import)

    before = asyncio.get_event_loop_policy()
    installed = maybe_install_uvloop(_postgres_config())

    assert installed is False
    assert asyncio.get_event_loop_policy() is before