""" 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//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()