conftest.py
Rama master · réplica de lectura del repositorio original.
"""
Repository-root pytest configuration.
This file holds the isolation that lets the unit-test suite run on a
developer's host without Docker, without a Redis server, and without the
system PyGObject stack the viewer service normally requires. It lives at
the **repo root** — not under ``tests/`` — on purpose: pytest only loads
a ``conftest.py`` for tests at or below its directory, and the two test
trees (``tests/`` and ``src/anthias_server/api/tests/``) have no common
ancestor other than the root. Putting the Redis/D-Bus/env isolation here
means ``pytest src/anthias_server/api/tests`` in isolation gets the same
fake-Redis wiring as a full run; previously that isolation lived only in
``tests/conftest.py`` and silently didn't apply to the API tree, so every
viewer-notifying view (asset create / update / delete → issue #2430)
``ConnectionError``'d against ``redis:6379`` whenever the API tests were
run on their own. The Playwright/marketing fixtures stay in
``tests/conftest.py`` since only the integration suite under ``tests/``
uses them.
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. ``src/anthias_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``.
"""
import importlib.util
import os
import sys
import types
from collections.abc import Iterator
from typing import Any
from unittest.mock import MagicMock
import pytest
# ---------------------------------------------------------------------------
# 0. App-availability probe
# ---------------------------------------------------------------------------
#
# This conftest lives at the repo root so the Redis/D-Bus/env isolation
# reaches BOTH ``tests/`` and ``src/anthias_server/api/tests/`` (their only
# common ancestor). But the root is also an ancestor of
# ``tools/raspberry_pi_imager/tests/``, which the ``run-python-linter`` CI
# job runs with ``-p no:django`` in a *minimal* venv that has neither
# Django nor pytz. Importing the app stack there (``anthias_common.utils``
# pulls in ``pytz``) would crash conftest collection for those tests.
#
# So gate every app-dependent hook on whether the app is importable at
# all. When it isn't (the rpi-imager lint env), those hooks (the
# connect_to_redis patch, the _mock_redis / _ensure_assetdir autouse
# fixtures) degrade to no-ops and the app-free tests run untouched. The
# cheap, dependency-free setup — ENVIRONMENT=test and the gi/pydbus
# stubs — still runs unconditionally; it's harmless for those tests and
# needed by everything else.
try:
import anthias_common.utils as _anthias_common_utils # noqa: F401
_APP_AVAILABLE = True
except ImportError:
# Only a missing dependency (the rpi-imager lint env lacks Django /
# pytz) should flip us to no-op. Catch ImportError specifically, not
# bare Exception: a genuine error inside the app package (SyntaxError,
# a broken import-time side effect) must surface loudly rather than
# silently disabling Redis isolation for the whole app test suite.
_APP_AVAILABLE = False
# ---------------------------------------------------------------------------
# 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.
gi_repository.Gio = MagicMock(name='gi.repository.Gio') # type: ignore[attr-defined]
gi_repository.GLib = MagicMock(name='gi.repository.GLib') # type: ignore[attr-defined]
gi_repository.GObject = MagicMock(name='gi.repository.GObject') # type: ignore[attr-defined]
gi_module.repository = gi_repository # type: ignore[attr-defined]
sys.modules['gi'] = gi_module
sys.modules['gi.repository'] = gi_repository
if pydbus_missing or gi_missing:
pydbus_module = types.ModuleType('pydbus')
pydbus_module.SessionBus = MagicMock(name='pydbus.SessionBus') # type: ignore[attr-defined]
pydbus_module.SystemBus = MagicMock(name='pydbus.SystemBus') # type: ignore[attr-defined]
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
# Swallows the connection overrides a caller may pass (see
# ``lib.system_logs._reply_connection``, which widens the socket
# for its blocking pop). The fake has no sockets to tune, but the
# stub still has to accept the call rather than TypeError on it.
def _fake_connect(**_overrides: Any) -> Any:
return _SESSION_FAKE_REDIS
_lib_utils.connect_to_redis = _fake_connect
if _APP_AVAILABLE:
_patch_connect_to_redis()
@pytest.fixture(scope='session', autouse=True)
def _isolated_assetdir(
tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[None]:
"""Point ``settings['assetdir']`` at a temp directory for the run.
Upload tests write real files. Unpatched, ``assetdir`` resolves to
``~/anthias_assets`` — a developer's actual asset directory on a
host run, and a directory the suite has no business writing into.
The path also has to exist before a test touches it: the Docker
test image creates ``/data/anthias_assets`` in its build (see
``docker/Dockerfile.test.j2``), local hosts guarantee nothing.
Redirecting the config file is what makes it hold, rather than
just setting the key. ``AnthiasSettings.load()`` rebuilds
``assetdir`` from whatever the config file says — and ``save()``
ends in a ``load()`` — so setting the key alone is undone by the
next settings-saving test in the same worker. Patching ``home``
does not help either: the config stores an absolute
``assetdir = /home/<user>/anthias_assets``, and ``path.join`` with
an absolute second argument discards the first.
So the singleton is pointed at a temp config that carries the temp
path, written from the values already in memory. Every later
``load()`` then resolves back to the temp directory, including the
one in ``_isolated_settings_conf``'s teardown, which restores
whatever ``conf_file`` it found — this one.
"""
if not _APP_AVAILABLE:
yield
return
from unittest import mock
from anthias_server.settings import settings as _anthias_settings
home = tmp_path_factory.mktemp('home')
asset_dir = home / 'anthias_assets'
asset_dir.mkdir()
conf_file = str(home / 'anthias.conf')
original_conf_file = _anthias_settings.conf_file
try:
with mock.patch.dict(_anthias_settings, {'assetdir': str(asset_dir)}):
_anthias_settings.conf_file = conf_file
# Writes the values already in memory — the developer's
# real settings — with `assetdir` redirected. `database`
# is redirected too: `_get` rebuilds both from the config
# the same way, so leaving one behind would point the
# suite at the real SQLite file. The secret key is blanked
# rather than copied into a world-readable-by-default
# temp tree.
_anthias_settings['database'] = str(home / 'anthias.db')
_anthias_settings['django_secret_key'] = ''
_anthias_settings.save()
yield
finally:
_anthias_settings.conf_file = original_conf_file
_anthias_settings.load()
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.
No-op (yields a bare fake) when the app stack isn't importable — the
app-free ``tools/raspberry_pi_imager/tests`` collected under this root
conftest must not drag in Django / redis / pytz.
"""
if not _APP_AVAILABLE:
yield _make_fake_redis()
return
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)
# ``ViewerPublisher`` / ``ReplyCollector`` are process-wide singletons
# (``INSTANCE`` classvar) that bind ``self._redis = connect_to_redis()``
# exactly once, in ``__init__``. Patching ``connect_to_redis`` above
# doesn't reach an instance an earlier test already built, so a
# singleton first constructed while pointed at the real ``redis`` host
# would leak that live client into every later test — any view that
# calls ``send_to_viewer`` (asset create / update / delete → issue
# #2430 viewer wake) would then ``ConnectionError`` on a host without
# Redis. Seed both singletons with this test's fake so the view's
# ``get_instance()`` returns a fake-backed instance regardless of what
# any prior test left behind, and clear them afterwards so nothing
# leaks forward.
from anthias_server.settings import ReplyCollector, ViewerPublisher
def _seed_singleton(cls: Any) -> None:
# Bypass __init__ (which both raises if an instance exists and
# calls the real connect_to_redis) and inject the fake directly.
instance = cls.__new__(cls)
instance._redis = fake
cls.INSTANCE = instance
_seed_singleton(ViewerPublisher)
_seed_singleton(ReplyCollector)
yield fake
ViewerPublisher.INSTANCE = None
ReplyCollector.INSTANCE = None
@pytest.fixture
def _isolated_settings_conf(tmp_path: Any) -> Any:
"""Redirect the settings singleton's config file to a per-test temp
path so ``settings.save()`` writes — whether direct or through the
``settings_save`` view — never touch the real
``~/.anthias/anthias.conf``.
Without this, a settings-mutating test leaks its posted values onto
the shared on-device config and silently breaks later tests that
read defaults back from it. Concretely, the integration suite's
``test_add_asset_via_url`` asserts a new asset inherits
``default_duration``; it fails with ``assert 10 == 0`` when a
sibling here has already persisted a different value to disk (only
when both suites run against the same ``/data`` volume — CI dodges
it by isolating the two jobs). Generalises the same conf_file
redirect already used by ``_reset_review_cta``.
"""
from anthias_server.settings import settings
original_conf_file = settings.conf_file
try:
# Inside the try so a failure in the reassignment or the initial
# save() can't leave conf_file pointed at the temp path for the
# rest of the session — the finally always restores it.
settings.conf_file = str(tmp_path / 'anthias.conf')
settings.save()
yield
finally:
# Point back at the real config and reload so the singleton's
# in-memory state is restored for any later test.
settings.conf_file = original_conf_file
settings.load()