agentic idp test case

Author: kazmer97

lib/idp_common_pkg/idp_common/extraction/service.py

@@ -1342,8 +1342,8 @@ def process_document_section(self, document: Document, section_id: str) -> Docum
                 else:
                     message_prompt = content
                 logger.info("Using Agentic extraction")
-                logger.info(f"Using input: {str(message_prompt)}")
-                structured_data, response_with_metering = structured_output(
+                logger.debug(f"Using input: {str(message_prompt)}")
+                structured_data, response_with_metering = structured_output(  # pyright: ignore[reportPossiblyUnboundVariable]
                     model_id=model_id,
                     data_format=dynamic_model,
                     prompt=message_prompt,  # pyright: ignore[reportArgumentType]

lib/idp_common_pkg/pyproject.toml

@@ -150,5 +150,6 @@ agentic-extraction = [
 
 [dependency-groups]
 dev = [
+    "moto>=5.1.8",
     "pytest-asyncio>=1.1.0",
 ]

lib/idp_common_pkg/tests/unit/extraction/agentic_idp/conftest.py

@@ -0,0 +1,104 @@
+"""
+Pytest configuration for agentic_idp extraction tests.
+These tests require real strands/pyarrow packages and are automatically skipped when unavailable.
+
+To run these tests locally:
+    pytest -m agentic tests/unit/extraction/agentic_idp/
+"""
+
+import os
+import sys
+
+
+def _should_skip_collection(config):
+    """Determine if we should skip collection based on environment and pytest marks."""
+    # Check if user explicitly requested agentic tests via -m agentic
+    if config.option.markexpr and "agentic" in config.option.markexpr:
+        return False
+
+    # Skip in CI
+    if os.getenv("CI"):
+        return True
+
+    # Check if strands is available and not mocked
+    try:
+        from unittest.mock import MagicMock
+
+        import strands
+
+        if isinstance(strands, MagicMock):
+            return True
+    except ImportError:
+        return True
+
+    return False
+
+
+def pytest_ignore_collect(collection_path, config):
+    """Skip collection of test files in this directory if strands unavailable."""
+    if _should_skip_collection(config) and "agentic_idp" in str(collection_path):
+        return True
+    return False
+
+
+def pytest_configure(config):
+    """Remove mocked modules to allow real imports."""
+    config.addinivalue_line(
+        "markers",
+        "agentic: mark test as requiring real strands package (run with -m agentic)",
+    )
+
+    if _should_skip_collection(config):
+        return
+
+    # Only remove modules that are actually MagicMock instances
+    from unittest.mock import MagicMock
+
+    modules_to_unmock = [
+        "strands",
+        "strands.models",
+        "strands.models.bedrock",
+        "strands.types",
+        "strands.types.content",
+        "strands.types.media",
+        "strands.hooks",
+        "strands.hooks.events",
+        "pyarrow",
+    ]
+
+    # Remove mocked modules
+    for module_name in modules_to_unmock:
+        if module_name in sys.modules and isinstance(
+            sys.modules[module_name], MagicMock
+        ):
+            sys.modules.pop(module_name, None)
+
+    # Remove any modules that imported the mocked modules so they get re-imported fresh
+    modules_to_reload = [
+        "idp_common.extraction.agentic_idp",
+        "idp_common.extraction.service",
+        "PIL",
+        "PIL.Image",
+        "PIL.ImageEnhance",
+        "PIL.ImageOps",
+    ]
+
+    for module_name in modules_to_reload:
+        sys.modules.pop(module_name, None)
+
+    # Remove any modules that imported the mocked modules so they get re-imported fresh
+    modules_to_reload = [
+        "idp_common.extraction.agentic_idp",
+        "idp_common.extraction.service",
+    ]
+
+    for module_name in modules_to_reload:
+        sys.modules.pop(module_name, None)
+
+    # Remove modules that depend on the unmocked modules so they get re-imported fresh
+    for module_name in modules_to_reload:
+        sys.modules.pop(module_name, None)
+
+    # Remove modules that depend on the unmocked modules
+    for module_name in modules_to_reload:
+        sys.modules.pop(module_name, None)