Anthias/tests/conftest.py

"""
Root-level pytest configuration.

This file enables the unit-test suite to run on a developer's host
without Docker, without a Redis server, and without the system PyGObject
stack the viewer service normally requires. Integration tests
(``-m integration``) opt back into real services by replacing these
fixtures themselves.

Three concerns are handled here, in order:

1. Force ``ENVIRONMENT=test`` so ``anthias_server.django_project/settings.py`` selects
   the SQLite test-DB branch (a repo-local path under ``BASE_DIR`` by
   default; CI overrides via ``ANTHIAS_TEST_DB_PATH``).

2. Stub ``gi`` / ``gi.repository`` / ``pydbus`` in ``sys.modules`` *before*
   any application module is imported. ``viewer/__init__.py`` does
   ``import pydbus`` at module load, and ``pydbus`` in turn imports
   ``gi.repository.Gio`` — which only resolves on hosts with the
   distribution's ``python3-gi`` package installed and wired into the
   active interpreter. The stubs let the import succeed; tests that
   exercise dbus paths mock the relevant calls themselves.

3. Replace ``anthias_common.utils.connect_to_redis`` with a dict-backed
   ``MagicMock`` factory before any test module imports it, then expose
   the same fake via an autouse fixture. Several modules call
   ``r = connect_to_redis()`` at import time
   (``anthias_server.celery_tasks``, ``anthias_server.lib.github``, ``anthias_server.lib.telemetry``, ...); patching
   the factory once at conftest load time means the module-level ``r``
   bindings hold a fake, not a client pointed at host ``redis``.

Browser-test concerns specific to the Playwright integration suite —
the shared ``browser_context_args`` / ``browser_type_launch_args``
overrides and the opt-in ``marketing_screenshot`` capture fixture —
live at the bottom of this file, after the unit-test setup.
"""

import importlib.util
import os
import sys
import types
from collections.abc import Callable, Iterator
from io import BytesIO
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock

import pytest

# ---------------------------------------------------------------------------
# 1. ENVIRONMENT=test (settings.py reads this at import time)
# ---------------------------------------------------------------------------

os.environ.setdefault('ENVIRONMENT', 'test')


# ---------------------------------------------------------------------------
# 2. Stub gi / gi.repository / pydbus before viewer modules are loaded
# ---------------------------------------------------------------------------


def _install_dbus_stubs() -> None:
    """
    Insert stand-in modules so ``import pydbus`` (and its transitive
    ``from gi.repository import Gio, GLib, GObject``) succeeds on
    hosts that don't have the full PyGObject + pydbus stack.

    Three host shapes show up in the wild:

    1. Both ``gi`` and ``pydbus`` available (target image): no-op.
    2. ``gi`` missing entirely (typical dev laptop without
       ``python3-gi`` apt package): stub both, because the real
       pydbus's module-load does ``from gi.repository import
       GLib, GObject`` and our minimal gi stub can't satisfy that.
    3. ``gi`` present but ``pydbus`` not pip-installed: stub only
       pydbus.
    """
    gi_missing = importlib.util.find_spec('gi') is None
    pydbus_missing = importlib.util.find_spec('pydbus') is None

    if gi_missing:
        gi_module = types.ModuleType('gi')
        gi_repository = types.ModuleType('gi.repository')
        # MagicMock for the GLib/Gio/GObject surface — pydbus only
        # touches these when a bus is actually constructed (e.g.
        # ``pydbus.SessionBus()`` inside a function body); tests that
        # hit those paths mock them themselves.
        setattr(gi_repository, 'Gio', MagicMock(name='gi.repository.Gio'))
        setattr(gi_repository, 'GLib', MagicMock(name='gi.repository.GLib'))
        setattr(
            gi_repository, 'GObject', MagicMock(name='gi.repository.GObject')
        )
        setattr(gi_module, 'repository', gi_repository)
        sys.modules['gi'] = gi_module
        sys.modules['gi.repository'] = gi_repository

    if pydbus_missing or gi_missing:
        pydbus_module = types.ModuleType('pydbus')
        setattr(
            pydbus_module, 'SessionBus', MagicMock(name='pydbus.SessionBus')
        )
        setattr(pydbus_module, 'SystemBus', MagicMock(name='pydbus.SystemBus'))
        sys.modules['pydbus'] = pydbus_module


_install_dbus_stubs()


# ---------------------------------------------------------------------------
# 3. Defensive Redis mock — replace connect_to_redis() everywhere
# ---------------------------------------------------------------------------


def _make_fake_redis() -> MagicMock:
    """
    A dict-backed Redis mock matching the surface our code uses.

    String ops (get/set/delete/expire/exists/flushdb/publish) and list
    ops (rpush/lpop/blpop) are modelled on the real Redis semantics so
    test paths that exercise both — notably ``ReplyCollector.recv_json``
    via BLPOP — see realistic behaviour rather than no-ops.
    """
    store: dict[str, Any] = {}

    fake = MagicMock(name='FakeRedis')
    fake.get.side_effect = store.get

    def _set(
        key: str,
        value: Any,
        *,
        nx: bool = False,
        ex: int | None = None,
        **_: Any,
    ) -> bool | None:
        # Match real Redis ``SET ... NX``: succeeds only if the key is
        # not already present, returns True on success / None on no-op.
        # Tests for SETNX-based gates (per-asset recheck cooldown,
        # sweep singleton lock, splash IP-refresh debounce) rely on
        # this — without it, every "is the lock held?" check would
        # spuriously believe the lock was free.
        if nx and key in store:
            return None
        store[key] = value
        return True

    fake.set.side_effect = _set
    fake.delete.side_effect = lambda *keys: sum(
        1 for k in keys if store.pop(k, None) is not None
    )
    fake.expire.side_effect = lambda key, _ttl: bool(key in store)
    fake.exists.side_effect = lambda *keys: sum(1 for k in keys if k in store)
    fake.flushdb.side_effect = lambda: store.clear()
    fake.publish.side_effect = lambda channel, msg: 0

    def _eval(script: str, numkeys: int, *args: Any) -> Any:
        # Compare-and-delete is the only ``EVAL`` script in the
        # codebase (sweep lock release in anthias_server.celery_tasks.py). Implement
        # that pattern directly; any other script becomes a no-op.
        if "redis.call('get', KEYS[1])" in script and "'del'" in script:
            keys = list(args[:numkeys])
            argv = list(args[numkeys:])
            if keys and argv and store.get(keys[0]) == argv[0]:
                store.pop(keys[0], None)
                return 1
            return 0
        return None

    fake.eval.side_effect = _eval

    def _rpush(key: str, *values: Any) -> int:
        bucket = store.setdefault(key, [])
        bucket.extend(values)
        return len(bucket)

    def _lpop(key: str) -> Any:
        bucket = store.get(key)
        if not bucket:
            return None
        value = bucket.pop(0)
        if not bucket:
            store.pop(key, None)
        return value

    def _blpop(keys: Any, timeout: float | None = None) -> Any:
        # Real BLPOP blocks until a value is available or the timeout
        # expires. Tests don't drive a writer thread, so block-then-
        # poll behaviour collapses to "return immediately": yield the
        # first available value, or None if every key is empty.
        if isinstance(keys, (str, bytes)):
            keys = [keys]
        for key in keys:
            value = _lpop(key)
            if value is not None:
                return (key, value)
        return None

    fake.rpush.side_effect = _rpush
    fake.lpop.side_effect = _lpop
    fake.blpop.side_effect = _blpop
    return fake


# Patch ``connect_to_redis`` at the source so the module-level
# ``r = connect_to_redis()`` bindings in anthias_server.celery_tasks / anthias_server.lib.github /
# anthias_server.lib.telemetry / api.views.mixins / viewer / etc. all resolve to the
# fake the moment those modules are first imported.
_SESSION_FAKE_REDIS = _make_fake_redis()


def _patch_connect_to_redis() -> None:
    import anthias_common.utils as _lib_utils

    _lib_utils.connect_to_redis = lambda: _SESSION_FAKE_REDIS


_patch_connect_to_redis()


@pytest.fixture(scope='session', autouse=True)
def _ensure_assetdir() -> None:
    """
    Some legacy fixtures (e.g. ``api/tests/test_v1_endpoints.py::
    cleanup_asset_dir``) iterate ``settings['assetdir']`` during
    teardown without creating it first. The Docker test image creates
    ``/data/anthias_assets`` in its build (see
    ``docker/Dockerfile.test.j2``); local hosts have no such guarantee.
    Materialise the path once per session so those fixtures don't
    ``FileNotFoundError`` out before the test even runs.
    """
    from anthias_server.settings import settings as _anthias_settings

    asset_dir = _anthias_settings.get('assetdir')
    if asset_dir:
        os.makedirs(asset_dir, exist_ok=True)


# Browser-test failure artifacts are owned by pytest-playwright. The
# `--tracing retain-on-failure --screenshot only-on-failure
# --output test-artifacts` flags in pyproject.toml's addopts make it
# write `<output>/<test-id>/{trace.zip,test-failed-1.png}` for failed
# tests and nothing for passing ones. The GH Actions
# upload-artifact@v7 step in test-runner.yml uploads the directory on
# job failure, where the trace replays via `playwright show-trace`.


def pytest_collection_modifyitems(
    config: pytest.Config, items: list[pytest.Item]
) -> None:
    """Set ``DJANGO_ALLOW_ASYNC_UNSAFE=1`` only when the run actually
    contains integration tests.

    Playwright's sync API spins up an internal asyncio loop to talk
    to the browser over the CDP socket; Django's ORM detects that
    loop and refuses sync calls (``SynchronousOnlyOperation``) unless
    this flag is set. The single-threaded test process never invokes
    ORM and Playwright concurrently, so the safety net Django enforces
    isn't doing useful work for this suite.

    Setting it process-wide unconditionally would also disable the
    safety net for unit tests, where an accidental ORM call from
    inside an event loop is a real bug we want Django to flag. Hooking
    at collection time means a unit-only run (``pytest -m "not
    integration"``) leaves the variable unset and the check active;
    a run that includes integration tests sets it once before
    pytest-django's DB setup (which would otherwise hit the same
    check itself).
    """
    if any('integration' in item.keywords for item in items):
        os.environ['DJANGO_ALLOW_ASYNC_UNSAFE'] = '1'


@pytest.fixture(autouse=True)
def _mock_redis(monkeypatch: pytest.MonkeyPatch) -> Iterator[MagicMock]:
    """
    Replace ``anthias_common.utils.connect_to_redis`` with a dict-backed
    ``MagicMock`` for every test, including any module-level
    ``r = connect_to_redis()`` bindings that fixtures import indirectly.

    Tests that need their own Redis mock (e.g. ``tests/test_telemetry.py``,
    ``tests/test_messaging.py``) override this by patching ``module.r``
    directly — that takes precedence inside the per-test setup chain.
    """
    fake = _make_fake_redis()
    monkeypatch.setattr('anthias_common.utils.connect_to_redis', lambda: fake)

    # Replace already-bound ``r`` attributes on modules that called
    # connect_to_redis() at import time. Only modules already in
    # sys.modules are touched — others get the fake on first import via
    # the conftest-level patch above.
    for module_path in (
        'anthias_server.app.views',
        'anthias_server.api.views.mixins',
        'anthias_server.api.views.v2',
        'anthias_server.celery_tasks',
        'anthias_server.lib.github',
        'anthias_server.lib.telemetry',
        'anthias_viewer',
    ):
        mod = sys.modules.get(module_path)
        if mod is not None and hasattr(mod, 'r'):
            monkeypatch.setattr(f'{module_path}.r', fake)

    yield fake


# ---------------------------------------------------------------------------
# 4. Playwright browser overrides + opt-in marketing capture
# ---------------------------------------------------------------------------
#
# Consolidates the ``browser_context_args`` / ``browser_type_launch_args``
# overrides that test_app.py and test_migrate_to_screenly.py previously
# duplicated. The viewport (1400×900) matches the existing
# website/assets/images/overview*.png convention exactly, so captures
# slot into the Hugo image-set without rescaling.
#
# The ``marketing_screenshot`` fixture below is opt-in via
# MARKETING_SCREENSHOTS=1: when enabled it bumps the context to 3×
# device scale and saves a ``<name>.png`` plus retina ``<name>@2x.png``
# and ``<name>@3x.png`` siblings under test-artifacts/marketing/ —
# matching the website's existing overview*.png naming. Default
# integration runs stay at 1× and the fixture is a no-op so call sites
# in test bodies don't need to branch.

_MARKETING_ENABLED = os.environ.get('MARKETING_SCREENSHOTS') == '1'
_MARKETING_SCALE = 3
_MARKETING_OUT_DIR = Path('test-artifacts/marketing')


@pytest.fixture(scope='session', autouse=True)
def _reset_marketing_dir() -> None:
    """Clear ``test-artifacts/marketing/`` once per session when
    marketing capture is enabled. Runs inside the test container
    (where pytest executes), so the cleanup has the same root
    permissions as the container that originally wrote the files —
    a host-side ``rm -rf`` from a CI runner step would hit
    permission-denied on a retry attempt because the bind-mounted
    artefacts are owned by root.

    No-op when MARKETING_SCREENSHOTS is unset; ordinary integration
    runs leave any existing marketing/ tree alone."""
    if _MARKETING_ENABLED and _MARKETING_OUT_DIR.exists():
        import shutil

        shutil.rmtree(_MARKETING_OUT_DIR)


@pytest.fixture(scope='session')
def browser_context_args(
    browser_context_args: dict[str, Any],
) -> dict[str, Any]:
    args = {
        **browser_context_args,
        'viewport': {'width': 1400, 'height': 900},
    }
    if _MARKETING_ENABLED:
        # Driving the whole session at 3× costs ~25% more CPU per
        # test (framebuffer scales with DPR²) — the price of being
        # able to drop call sites into existing tests rather than
        # fork a parallel marketing suite.
        args['device_scale_factor'] = _MARKETING_SCALE
    return args


@pytest.fixture(scope='session')
def browser_type_launch_args(
    browser_type_launch_args: dict[str, Any],
) -> dict[str, Any]:
    # ``--no-sandbox`` because the test container runs as root;
    # Chromium's setuid sandbox refuses to come up in that
    # configuration and the user-namespace sandbox would need extra
    # capabilities at compose-up time.
    return {
        **browser_type_launch_args,
        'args': [*browser_type_launch_args.get('args', []), '--no-sandbox'],
    }


MarketingShotFn = Callable[..., None]


@pytest.fixture
def marketing_screenshot(request: pytest.FixtureRequest) -> MarketingShotFn:
    """Capture ``page`` at the context's scale factor and emit a
    ``<name>.png`` plus retina ``<name>@2x.png`` and ``<name>@3x.png``
    siblings under ``test-artifacts/marketing/``. Filename convention
    matches the existing website ``overview.png`` / ``overview@2x.png``
    / ``overview@3x.png`` set so Hugo's image-set picker resolves the
    new files without additional config — the base 1× is the canonical
    URL and the ``@Nx`` siblings are retina sources.

    Call as ``marketing_screenshot('home')`` for a viewport-only
    capture (default) at 1400×900 — the size the website's home-page
    slider expects every slide to be. Pass ``full_page=True`` to
    capture the entire scrolled document instead; useful for asset
    pages that aren't sliced into the slider.

    Viewport-only is the default for two reasons. First, the home-
    page slider in ``website/`` lays every slide into a uniform
    1400×900 frame, and full-page captures (variable height, growing
    with document content) crop unpredictably in that frame. Second,
    any capture that includes a ``position: fixed`` overlay needs
    viewport-only: Playwright's full-page mode artificially extends
    the viewport height to fit the document, and fixed-position
    elements anchored to the (now much taller) viewport drift out of
    frame or get hidden under the dimming backdrop.

    No-op (still callable) when ``MARKETING_SCREENSHOTS`` is unset, so
    test bodies can sprinkle calls unconditionally — non-marketing
    integration runs pay zero capture cost beyond the function call.

    The 3× capture is the source of truth; 2× and 1× variants come
    from a LANCZOS downscale via Pillow — same algorithm professional
    design tooling uses for retina export. Re-rendering at lower DPRs
    produces visible moiré on the asset-table grid lines, so we don't
    do that.
    """
    if not _MARKETING_ENABLED:

        def _noop(name: str, *, full_page: bool = False) -> None:
            pass

        return _noop

    # Only resolve ``page`` when capture is actually active — this
    # fixture lives in the root conftest and the ``page`` fixture only
    # exists when pytest-playwright is collected, which is only the
    # case for integration tests. Lazy fixture lookup keeps unit-test
    # collection from importing playwright when MARKETING_SCREENSHOTS
    # is unset (the common case).
    page = request.getfixturevalue('page')

    # Pillow ships with the server image (HEIC normalisation pipeline),
    # so the test container has it. Import is local so the non-marketing
    # path stays clean if Pillow ever moves out of the server group.
    from PIL import Image as _PILImage

    _MARKETING_OUT_DIR.mkdir(parents=True, exist_ok=True)

    def _capture(name: str, *, full_page: bool = False) -> None:
        png_bytes = page.screenshot(full_page=full_page)

        # Write the 3× original verbatim — re-encoding through PIL
        # would needlessly re-compress.
        (_MARKETING_OUT_DIR / f'{name}@3x.png').write_bytes(png_bytes)

        src = _PILImage.open(BytesIO(png_bytes))
        w, h = src.size
        for scale, suffix in ((2, '@2x'), (1, '')):
            target = (
                int(w * scale / _MARKETING_SCALE),
                int(h * scale / _MARKETING_SCALE),
            )
            resized = src.resize(target, _PILImage.Resampling.LANCZOS)
            resized.save(_MARKETING_OUT_DIR / f'{name}{suffix}.png')

    return _capture