DealDocumentScreening/tests/conftest.py
2026-09-02 23:32:47 +03:00

71 lines
2.7 KiB
Python

"""Shared pytest fixtures.
Unit tests stay fast and dependency-free (no DB/MQ/S3). Integration tests use
the running Docker Compose infrastructure and are marked `@pytest.mark.integration`
(deselected by the default `pytest -q` run).
"""
from __future__ import annotations
import importlib.abc
import importlib.util
import os
import sys
import types
# Default test env: prevent Settings() from failing on required fields when no
# .env is present. Individual tests that build Settings override as needed.
os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://x:x@localhost:5432/x")
os.environ.setdefault("RABBITMQ_URL", "amqp://x:x@localhost:5672//")
os.environ.setdefault("S3_ENDPOINT_URL", "http://localhost:9000")
os.environ.setdefault("S3_ACCESS_KEY", "test")
os.environ.setdefault("S3_SECRET_KEY", "test")
os.environ.setdefault("JWT_SECRET", "test")
# `src/` is on sys.path for production imports (`src.contract_check.*`), but
# tests use the installed package path (`contract_check.*`). When both import
# paths are used in the same process, Python creates two module objects for
# every shared module. Metrics use the global Prometheus registry and crash
# with DuplicateTimeseries. This import hook redirects `src.contract_check.*`
# to the already-loaded `contract_check.*` modules when they exist.
class _SrcRedirectFinder(importlib.abc.MetaPathFinder):
def find_spec(
self,
fullname: str,
path: object = None,
target: object = None,
) -> importlib.util.ModuleSpec | None:
if not fullname.startswith("src.contract_check"):
return None
target_name = fullname[4:] # strip leading "src."
if target_name in sys.modules:
return importlib.util.spec_from_loader(fullname, self)
# The `contract_check.*` package is already loaded (tests use that
# path), but a production source file references the `src.contract_check.*`
# path. Eagerly load the target through the `contract_check.*` path and
# redirect so both module names point to the same object.
if "contract_check" in sys.modules:
try:
__import__(target_name)
except Exception:
return None
if target_name in sys.modules:
return importlib.util.spec_from_loader(fullname, self)
return None
def create_module(
self,
spec: importlib.util.ModuleSpec, # noqa: ARG002
) -> types.ModuleType | None:
target_name = spec.name[4:]
return sys.modules.get(target_name)
def exec_module(
self,
module: types.ModuleType, # noqa: ARG002
) -> None:
return None
sys.meta_path.insert(0, _SrcRedirectFinder())