From d49c54506d6572e1ab13e4801151e67bb8e81dc1 Mon Sep 17 00:00:00 2001 From: Divyam Talwar Date: Mon, 29 Jun 2026 21:45:17 +0530 Subject: [PATCH] Avoid ingestion service settings validation during import Importing IngestionService during pytest collection previously resolved settings immediately and could require POSTGRES_URI before tests ran. Move settings validation to service construction, pass the already-resolved startup/worker settings into production instances, keep module-level settings attribute access lazy, preserve direct proxy overrides through a validated settings reconstruction, and add regression coverage for the import and constructor contracts. Constraint: Default local config selects Postgres and validates POSTGRES_URI when get_settings() runs. Rejected: Test-only os.environ defaulting | mutates process-wide configuration during import. Rejected: Requiring test callers to replace the whole module settings object | breaks existing direct-attribute monkeypatch seams. Rejected: Returning the mutable settings proxy from construction | can bypass full settings validation when partial overrides exist. Rejected: model_copy(update=...) for override merges | Pydantic does not validate update values. Confidence: high Scope-risk: moderate Directive: Keep broader import-time settings cleanup out of this PR unless covered by targeted tests. Tested: uv run pre-commit run isort --files core/services/ingestion_service.py core/services_init.py core/workers/ingestion_worker.py core/tests/unit/test_ingestion_colpali_rendering.py core/tests/unit/test_ingestion_service_settings_import.py; uv run pre-commit run black --files core/services/ingestion_service.py core/services_init.py core/workers/ingestion_worker.py core/tests/unit/test_ingestion_colpali_rendering.py core/tests/unit/test_ingestion_service_settings_import.py; uv run pre-commit run ruff --files core/services/ingestion_service.py core/services_init.py core/workers/ingestion_worker.py core/tests/unit/test_ingestion_colpali_rendering.py core/tests/unit/test_ingestion_service_settings_import.py; env -u POSTGRES_URI uv run pytest -q core/tests/unit/test_ingestion_service_settings_import.py; env -u POSTGRES_URI uv run pytest --collect-only -q core/tests/unit/test_ingestion_service_settings_import.py core/tests/unit/test_ingestion_colpali_rendering.py; env -u POSTGRES_URI uv run pytest -q core/tests/unit/test_ingestion_service_settings_import.py core/tests/unit/test_ingestion_colpali_rendering.py; env -u POSTGRES_URI uv run pytest -q core/tests/unit/test_ingestion_service_settings_import.py core/tests/unit/test_ingestion_colpali_rendering.py core/tests/unit/test_ingestion_service_metadata_update.py; env -u POSTGRES_URI uv run pytest --collect-only -q Not-tested: Full test execution; includes integration/live-service tests that require external services. --- core/services/ingestion_service.py | 71 ++++- core/services_init.py | 1 + .../unit/test_ingestion_colpali_rendering.py | 35 ++- .../test_ingestion_service_settings_import.py | 261 ++++++++++++++++++ core/workers/ingestion_worker.py | 1 + 5 files changed, 350 insertions(+), 19 deletions(-) create mode 100644 core/tests/unit/test_ingestion_service_settings_import.py diff --git a/core/services/ingestion_service.py b/core/services/ingestion_service.py index 67abdcf6..842531f6 100644 --- a/core/services/ingestion_service.py +++ b/core/services/ingestion_service.py @@ -19,6 +19,7 @@ import os import tempfile import uuid +from copy import copy from datetime import UTC, datetime, timedelta from io import BytesIO from pathlib import Path @@ -30,7 +31,7 @@ from fastapi import HTTPException, UploadFile from PIL import Image as PILImage -from core.config import get_settings +from core import config as config_module from core.database.postgres_database import PostgresDatabase from core.embedding.base_embedding_model import BaseEmbeddingModel from core.limits_utils import check_and_increment_limits, estimate_pages_by_chars @@ -48,7 +49,58 @@ from core.vector_store.base_vector_store import BaseVectorStore logger = logging.getLogger(__name__) -settings = get_settings() + + +def get_settings(): + return config_module.get_settings() + + +class _SettingsProxy: + """Lazily resolve app settings while preserving direct test overrides.""" + + def __init__(self): + self._overrides: Dict[str, Any] = {} + + def __getattr__(self, name): + if name in self._overrides: + return self._overrides[name] + return getattr(get_settings(), name) + + def __setattr__(self, name, value): + if name.startswith("_"): + object.__setattr__(self, name, value) + return + self._overrides[name] = value + + def __delattr__(self, name): + if name in self._overrides: + del self._overrides[name] + return + raise AttributeError(name) + + def apply_overrides(self, resolved_settings): + if not self._overrides: + return resolved_settings + + overrides = dict(self._overrides) + if hasattr(resolved_settings, "model_dump"): + settings_data = resolved_settings.model_dump() + settings_data.update(overrides) + return type(resolved_settings)(**settings_data) + + settings_copy = copy(resolved_settings) + for name, value in overrides.items(): + setattr(settings_copy, name, value) + return settings_copy + + +settings = _SettingsProxy() + + +def _get_settings(): + if isinstance(settings, _SettingsProxy): + return settings.apply_overrides(get_settings()) + return settings class PdfConversionError(Exception): @@ -96,6 +148,7 @@ def __init__( parser: BaseParser, colpali_embedding_model: Optional[BaseEmbeddingModel] = None, colpali_vector_store: Optional[BaseVectorStore] = None, + settings: Optional[Any] = None, ): """ Initialize the IngestionService. @@ -108,6 +161,7 @@ def __init__( parser: Document parser for text extraction colpali_embedding_model: Optional ColPali embedding model (local or API) colpali_vector_store: Optional ColPali vector store for multi-vector embeddings + settings: Optional pre-resolved settings object """ self.db = database self.vector_store = vector_store @@ -116,6 +170,7 @@ def __init__( self.parser = parser self.colpali_embedding_model = colpali_embedding_model self.colpali_vector_store = colpali_vector_store + self.settings = settings if settings is not None else _get_settings() # ------------------------------------------------------------------------- # Validation helpers @@ -466,7 +521,7 @@ async def _verify_ingest_and_storage_limits( content_length: int, document_id: str, ) -> None: - if settings.MODE != "cloud" or not auth.user_id: + if self.settings.MODE != "cloud" or not auth.user_id: return num_pages = estimate_pages_by_chars(content_length) @@ -492,7 +547,7 @@ async def _verify_ingest_and_storage_limits( ) async def _record_storage_usage(self, auth: AuthContext, content_length: int, document_id: str) -> None: - if settings.MODE != "cloud" or not auth.user_id: + if self.settings.MODE != "cloud" or not auth.user_id: return try: @@ -1177,7 +1232,9 @@ async def _process_colpali_embeddings( """Process colpali multi-vector embeddings if enabled.""" chunk_objects_multivector = [] - if not (use_colpali and settings.ENABLE_COLPALI and self.colpali_embedding_model and self.colpali_vector_store): + if not ( + use_colpali and self.settings.ENABLE_COLPALI and self.colpali_embedding_model and self.colpali_vector_store + ): return chunk_objects_multivector mime_type = file_type if isinstance(file_type, str) else (file_type.mime if file_type is not None else None) @@ -1577,7 +1634,7 @@ def _process_pdf_for_colpali(self, file_content: bytes) -> List[Chunk]: logger.error("PDF file content is empty") raise PdfConversionError("PDF file content is empty") - dpi = settings.COLPALI_PDF_DPI + dpi = self.settings.COLPALI_PDF_DPI try: # Check document density to decide processing strategy @@ -1799,12 +1856,12 @@ def _convert_office_to_images( images_payload: List[Tuple[str, bytes]] = [] total_pages = len(pdf_document) render_failures = 0 + dpi = self.settings.COLPALI_PDF_DPI try: for page_num in range(total_pages): page = pdf_document[page_num] try: - dpi = settings.COLPALI_PDF_DPI mat = fitz.Matrix(dpi / 72, dpi / 72) pix = page.get_pixmap(matrix=mat) img_data = pix.tobytes("png") diff --git a/core/services_init.py b/core/services_init.py index 66bd2d98..4b2442a4 100644 --- a/core/services_init.py +++ b/core/services_init.py @@ -226,6 +226,7 @@ embedding_model=embedding_model, colpali_embedding_model=colpali_embedding_model, colpali_vector_store=colpali_vector_store, + settings=settings, ) logger.info("Ingestion service initialised") diff --git a/core/tests/unit/test_ingestion_colpali_rendering.py b/core/tests/unit/test_ingestion_colpali_rendering.py index 8ad2876c..cdd02f03 100644 --- a/core/tests/unit/test_ingestion_colpali_rendering.py +++ b/core/tests/unit/test_ingestion_colpali_rendering.py @@ -1,7 +1,9 @@ import subprocess from io import BytesIO from pathlib import Path +from types import SimpleNamespace +import pytest from PIL import Image from core.services import ingestion_service as ingestion_module @@ -61,8 +63,13 @@ def _non_blank_png_bytes() -> bytes: return output.getvalue() -def test_render_pdf_with_pymupdf_skips_blank_and_failed_pages(monkeypatch): - service = IngestionService(None, None, None, None, None) +@pytest.fixture +def colpali_rendering_settings(): + return SimpleNamespace(COLPALI_PDF_DPI=150, ENABLE_COLPALI=True, MODE="cloud") + + +def test_render_pdf_with_pymupdf_skips_blank_and_failed_pages(monkeypatch, colpali_rendering_settings): + service = IngestionService(None, None, None, None, None, settings=colpali_rendering_settings) fake_document = FakeDocument( [ FakePage(_non_blank_png_bytes()), @@ -81,8 +88,8 @@ def test_render_pdf_with_pymupdf_skips_blank_and_failed_pages(monkeypatch): assert fake_document.closed is True -def test_pdf_pdf2image_fallback_skips_blank_and_failed_pages(monkeypatch): - service = IngestionService(None, None, None, None, None) +def test_pdf_pdf2image_fallback_skips_blank_and_failed_pages(monkeypatch, colpali_rendering_settings): + service = IngestionService(None, None, None, None, None, settings=colpali_rendering_settings) good_page = Image.open(BytesIO(_non_blank_png_bytes())) blank_page = Image.open(BytesIO(_png_bytes((255, 255, 255)))) failing_page = Image.open(BytesIO(_non_blank_png_bytes())) @@ -97,12 +104,16 @@ def fake_img_to_base64_with_bytes(image): "open", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("force pdf2image fallback")), ) - monkeypatch.setattr(ingestion_module.pdf2image, "convert_from_bytes", lambda *args, **kwargs: [ - good_page, - blank_page, - failing_page, - good_page, - ]) + monkeypatch.setattr( + ingestion_module.pdf2image, + "convert_from_bytes", + lambda *args, **kwargs: [ + good_page, + blank_page, + failing_page, + good_page, + ], + ) monkeypatch.setattr(service, "img_to_base64_with_bytes", fake_img_to_base64_with_bytes) chunks = service._process_pdf_for_colpali(b"%PDF") @@ -112,8 +123,8 @@ def fake_img_to_base64_with_bytes(image): assert all(chunk.content.startswith("data:image/png;base64,") for chunk in chunks) -def test_office_conversion_skips_blank_and_failed_pages(monkeypatch): - service = IngestionService(None, None, None, None, None) +def test_office_conversion_skips_blank_and_failed_pages(monkeypatch, colpali_rendering_settings): + service = IngestionService(None, None, None, None, None, settings=colpali_rendering_settings) fake_document = FakeDocument( [ FakePage(_non_blank_png_bytes()), diff --git a/core/tests/unit/test_ingestion_service_settings_import.py b/core/tests/unit/test_ingestion_service_settings_import.py new file mode 100644 index 00000000..0caadad6 --- /dev/null +++ b/core/tests/unit/test_ingestion_service_settings_import.py @@ -0,0 +1,261 @@ +import importlib +import sys +from pathlib import Path +from types import ModuleType, SimpleNamespace + +import pytest +from pydantic import BaseModel, ValidationError + +INGESTION_MODULE_NAME = "core.services.ingestion_service" +_MISSING = object() + + +class PydanticSettingsStub(BaseModel): + MODE: str + ENABLE_COLPALI: bool + COLPALI_PDF_DPI: int + + +def _module_attr(module_name: str, attr_name: str): + module = sys.modules.get(module_name) + if module is None: + return _MISSING + return getattr(module, attr_name, _MISSING) + + +def _remove_module(module_name: str): + module = sys.modules.pop(module_name, _MISSING) + if module is _MISSING: + return + + parent_name, _, attr_name = module_name.rpartition(".") + parent_module = sys.modules.get(parent_name) + if parent_module is not None and getattr(parent_module, attr_name, _MISSING) is module: + delattr(parent_module, attr_name) + + +def _restore_module(module_name: str, previous_module, previous_parent_attr) -> None: + parent_name, _, attr_name = module_name.rpartition(".") + + if previous_module is _MISSING: + _remove_module(module_name) + else: + sys.modules[module_name] = previous_module + + parent_module = sys.modules.get(parent_name) + if parent_module is None: + return + + if previous_parent_attr is _MISSING: + current_attr = getattr(parent_module, attr_name, _MISSING) + current_module = sys.modules.get(module_name, _MISSING) + if current_attr is not _MISSING and current_attr is current_module: + delattr(parent_module, attr_name) + else: + setattr(parent_module, attr_name, previous_parent_attr) + + +def _remove_module_tree(root_module_name: str) -> None: + for module_name in sorted( + [name for name in sys.modules if name == root_module_name or name.startswith(f"{root_module_name}.")], + key=len, + reverse=True, + ): + _remove_module(module_name) + + +def _snapshot_module_tree(root_module_name: str): + snapshot = {} + for module_name, module in sys.modules.items(): + if module_name == root_module_name or module_name.startswith(f"{root_module_name}."): + parent_name, _, attr_name = module_name.rpartition(".") + snapshot[module_name] = (module, _module_attr(parent_name, attr_name)) + return snapshot + + +def _restore_module_tree(root_module_name: str, previous_modules, previous_root_attr) -> None: + _remove_module_tree(root_module_name) + for module_name, (previous_module, previous_parent_attr) in sorted( + previous_modules.items(), key=lambda item: len(item[0]) + ): + _restore_module(module_name, previous_module, previous_parent_attr) + + if root_module_name not in previous_modules: + parent_name, _, attr_name = root_module_name.rpartition(".") + parent_module = sys.modules.get(parent_name) + if parent_module is None: + return + if previous_root_attr is _MISSING: + if getattr(parent_module, attr_name, _MISSING) is not _MISSING: + delattr(parent_module, attr_name) + else: + setattr(parent_module, attr_name, previous_root_attr) + + +@pytest.fixture +def fresh_ingestion_service_import(): + modules_before = set(sys.modules) + previous_ingestion_module = sys.modules.get(INGESTION_MODULE_NAME, _MISSING) + previous_ingestion_attr = _module_attr("core.services", "ingestion_service") + previous_embedding_modules = _snapshot_module_tree("core.embedding") + previous_embedding_attr = _module_attr("core", "embedding") + + import core + + # Avoid unrelated embedding import side effects while testing ingestion_service imports. + embedding_stub = ModuleType("core.embedding") + embedding_stub.__path__ = [str(Path(__file__).resolve().parents[2] / "embedding")] + sys.modules["core.embedding"] = embedding_stub + setattr(core, "embedding", embedding_stub) + + _remove_module(INGESTION_MODULE_NAME) + + try: + yield lambda: importlib.import_module(INGESTION_MODULE_NAME) + finally: + for module_name in sorted(set(sys.modules) - modules_before, key=len, reverse=True): + if module_name == INGESTION_MODULE_NAME or module_name.startswith("core.embedding"): + _remove_module(module_name) + + _restore_module(INGESTION_MODULE_NAME, previous_ingestion_module, previous_ingestion_attr) + _restore_module_tree("core.embedding", previous_embedding_modules, previous_embedding_attr) + + +def test_importing_ingestion_service_does_not_resolve_settings(monkeypatch, fresh_ingestion_service_import): + from core import config as config_module + + calls = [] + + def fail_if_resolved(): + calls.append(True) + raise AssertionError("settings should not resolve during import") + + monkeypatch.setattr(config_module, "get_settings", fail_if_resolved) + + ingestion_module = fresh_ingestion_service_import() + + assert calls == [] + assert ingestion_module.IngestionService.__name__ == "IngestionService" + with pytest.raises(AssertionError, match="settings should not resolve during import"): + ingestion_module.settings.MODE + + +def test_injected_settings_bypass_constructor_resolution(monkeypatch): + from core.services import ingestion_service as ingestion_module + + sentinel_settings = SimpleNamespace(MODE="cloud", ENABLE_COLPALI=True, COLPALI_PDF_DPI=150) + + def fail_if_resolved(): + raise AssertionError("injected settings should bypass get_settings") + + monkeypatch.setattr(ingestion_module, "get_settings", fail_if_resolved) + + service = ingestion_module.IngestionService(None, None, None, None, None, settings=sentinel_settings) + + assert service.settings is sentinel_settings + + +def test_constructor_resolves_settings_when_not_injected(monkeypatch): + from core.services import ingestion_service as ingestion_module + + sentinel_settings = SimpleNamespace(MODE="self_hosted", ENABLE_COLPALI=False, COLPALI_PDF_DPI=96) + calls = [] + + def get_test_settings(): + calls.append(True) + return sentinel_settings + + monkeypatch.setattr(ingestion_module, "settings", ingestion_module._SettingsProxy()) + monkeypatch.setattr(ingestion_module, "get_settings", get_test_settings) + + service = ingestion_module.IngestionService(None, None, None, None, None) + + assert calls == [True] + assert service.settings is sentinel_settings + + +def test_constructor_uses_config_module_get_settings_after_import(monkeypatch): + from core import config as config_module + from core.services import ingestion_service as ingestion_module + + sentinel_settings = SimpleNamespace(MODE="self_hosted", ENABLE_COLPALI=False, COLPALI_PDF_DPI=96) + + monkeypatch.setattr(ingestion_module, "settings", ingestion_module._SettingsProxy()) + monkeypatch.setattr(config_module, "get_settings", lambda: sentinel_settings) + + service = ingestion_module.IngestionService(None, None, None, None, None) + + assert service.settings is sentinel_settings + + +def test_constructor_resolves_settings_before_applying_proxy_overrides(monkeypatch): + from core.services import ingestion_service as ingestion_module + + proxy = ingestion_module._SettingsProxy() + sentinel_settings = SimpleNamespace(MODE="self_hosted", ENABLE_COLPALI=False, COLPALI_PDF_DPI=96) + calls = [] + + def get_test_settings(): + calls.append(True) + return sentinel_settings + + monkeypatch.setattr(ingestion_module, "settings", proxy) + monkeypatch.setattr(ingestion_module, "get_settings", get_test_settings) + + ingestion_module.settings.MODE = "cloud" + service = ingestion_module.IngestionService(None, None, None, None, None) + + assert calls == [True] + assert service.settings is not proxy + assert service.settings is not sentinel_settings + assert service.settings.MODE == "cloud" + assert service.settings.ENABLE_COLPALI is False + + +def test_deleted_proxy_overrides_restore_constructor_fallback(monkeypatch): + from core.services import ingestion_service as ingestion_module + + proxy = ingestion_module._SettingsProxy() + sentinel_settings = SimpleNamespace(MODE="self_hosted", ENABLE_COLPALI=False, COLPALI_PDF_DPI=96) + calls = [] + + def get_test_settings(): + calls.append(True) + return sentinel_settings + + monkeypatch.setattr(ingestion_module, "settings", proxy) + monkeypatch.setattr(ingestion_module, "get_settings", get_test_settings) + + ingestion_module.settings.MODE = "cloud" + del ingestion_module.settings.MODE + service = ingestion_module.IngestionService(None, None, None, None, None) + + assert calls == [True] + assert service.settings is sentinel_settings + assert service.settings.MODE == "self_hosted" + + +def test_proxy_overrides_are_validated_during_constructor_resolution(monkeypatch): + from core.services import ingestion_service as ingestion_module + + proxy = ingestion_module._SettingsProxy() + sentinel_settings = PydanticSettingsStub(MODE="self_hosted", ENABLE_COLPALI=False, COLPALI_PDF_DPI=96) + + monkeypatch.setattr(ingestion_module, "settings", proxy) + monkeypatch.setattr(ingestion_module, "get_settings", lambda: sentinel_settings) + + ingestion_module.settings.COLPALI_PDF_DPI = "not-an-int" + + with pytest.raises(ValidationError): + ingestion_module.IngestionService(None, None, None, None, None) + + +def test_module_settings_proxy_resolves_lazily(monkeypatch): + from core.services import ingestion_service as ingestion_module + + sentinel_settings = SimpleNamespace(UNIQUE_TEST_VALUE="resolved") + + monkeypatch.setattr(ingestion_module, "settings", ingestion_module._SettingsProxy()) + monkeypatch.setattr(ingestion_module, "get_settings", lambda: sentinel_settings) + + assert ingestion_module.settings.UNIQUE_TEST_VALUE == "resolved" diff --git a/core/workers/ingestion_worker.py b/core/workers/ingestion_worker.py index 9dad0e3e..037548b6 100644 --- a/core/workers/ingestion_worker.py +++ b/core/workers/ingestion_worker.py @@ -509,6 +509,7 @@ def _meta_resolver(): # noqa: D401 parser=ctx["parser"], colpali_embedding_model=ctx.get("colpali_embedding_model"), colpali_vector_store=colpali_vector_store, + settings=settings, ) # 3. Download the file from storage