Fix pytest plugin relative import

Author: LucasG0

changelog/166.fixed.md

@@ -0,0 +1 @@
+Fix relative imports for the pytest plugin, note that the relative imports can't be at the top level of the repository alongside .infrahub.yml. They have to be located within a subfolder.

infrahub_sdk/_importer.py

@@ -16,6 +16,15 @@
 def import_module(
     module_path: Path, import_root: Optional[str] = None, relative_path: Optional[str] = None
 ) -> ModuleType:
+    """Imports a python module.
+
+    Attributes:
+        module_path (Path): Absolute path of the module to import.
+        import_root (Optional[str]): Absolute string path to the folder.
+        relative_path (Optional[str]): Relative string path between module_path and import_root.
+                                       TODO Compute `relative_path` here instead of having it as a parameter?
+    """
+
     import_root = import_root or str(module_path.parent)
 
     file_on_disk = module_path
@@ -35,6 +44,8 @@ def import_module(
         module_name = relative_path.replace("/", ".") + f".{module_name}"
 
     try:
+        # We hold a mapping of imported modules. If a module is already loaded and does not have recent changes,
+        # then we do not reload/import this module.
         if module_name in sys.modules:
             module = sys.modules[module_name]
             current_mtime = file_on_disk.stat().st_mtime

infrahub_sdk/checks.py

@@ -11,13 +11,10 @@
 from git.repo import Repo
 from pydantic import BaseModel, Field
 
-from .exceptions import InfrahubCheckNotFoundError, UninitializedError
+from .exceptions import UninitializedError
 
 if TYPE_CHECKING:
-    from pathlib import Path
-
     from . import InfrahubClient
-    from .schema.repository import InfrahubCheckDefinitionConfig
 
 INFRAHUB_CHECK_VARIABLE_TO_IMPORT = "INFRAHUB_CHECKS"
 
@@ -176,27 +173,3 @@ async def run(self, data: Optional[dict] = None) -> bool:
             self.log_info("Check succesfully completed")
 
         return self.passed
-
-
-def get_check_class_instance(
-    check_config: InfrahubCheckDefinitionConfig, search_path: Optional[Path] = None
-) -> InfrahubCheck:
-    if check_config.file_path.is_absolute() or search_path is None:
-        search_location = check_config.file_path
-    else:
-        search_location = search_path / check_config.file_path
-
-    try:
-        spec = importlib.util.spec_from_file_location(check_config.class_name, search_location)
-        module = importlib.util.module_from_spec(spec)  # type: ignore[arg-type]
-        spec.loader.exec_module(module)  # type: ignore[union-attr]
-
-        # Get the specified class from the module
-        check_class = getattr(module, check_config.class_name)
-
-        # Create an instance of the class
-        check_instance = check_class()
-    except (FileNotFoundError, AttributeError) as exc:
-        raise InfrahubCheckNotFoundError(name=check_config.name) from exc
-
-    return check_instance

infrahub_sdk/import_utils.py

@@ -0,0 +1,36 @@
+import importlib
+from pathlib import Path
+from typing import Optional
+
+from .exceptions import InfrahubCheckNotFoundError, InfrahubTransformNotFoundError
+from .schema.repository import InfrahubCheckDefinitionConfig, InfrahubPythonTransformConfig
+
+
+def get_check_or_transform_class(
+    config: InfrahubCheckDefinitionConfig | InfrahubPythonTransformConfig, search_path: Optional[Path] = None
+) -> type:
+    if config.file_path.is_absolute() or search_path is None:
+        search_location = config.file_path
+    else:
+        search_location = search_path / config.file_path
+
+    try:
+        spec = importlib.util.spec_from_file_location(config.class_name, search_location)
+        module = importlib.util.module_from_spec(spec)  # type: ignore[arg-type]
+
+        # Set base module for relative import. See https://github.com/opsmill/infrahub-sdk-python/issues/166.
+        # NOTE 1: When pytest plugin runs through proposed change pipeline, it is invoked within infrahub folder,
+        # ie outside of the imported repository. Thus, we cannot rely on `importlib.import_module` as other components,
+        # so we need to use `importlib.util.spec_from_file_location` in order to import desired module.
+        # NOTE 2: Using `__package__` logs a `DeprecationWarning: __package__ != __spec__.parent`
+        module.__package__ = str(search_location.parent.name)
+
+        spec.loader.exec_module(module)  # type: ignore[union-attr]
+
+        # Get the specified class from the module
+        return getattr(module, config.class_name)
+
+    except (FileNotFoundError, AttributeError) as exc:
+        if isinstance(config, InfrahubPythonTransformConfig):
+            raise InfrahubTransformNotFoundError(name=config.name) from exc
+        raise InfrahubCheckNotFoundError(name=config.name) from exc

infrahub_sdk/pytest_plugin/items/check.py

@@ -6,7 +6,7 @@
 import ujson
 from httpx import HTTPStatusError
 
-from ...checks import get_check_class_instance
+from ...import_utils import get_check_or_transform_class
 from ..exceptions import CheckDefinitionError, CheckResultError
 from ..models import InfrahubTestExpectedResult
 from .base import InfrahubItem
@@ -33,10 +33,11 @@ def __init__(
         self.check_instance: InfrahubCheck
 
     def instantiate_check(self) -> None:
-        self.check_instance = get_check_class_instance(
-            check_config=self.resource_config,  # type: ignore[arg-type]
+        check_class = get_check_or_transform_class(
+            config=self.resource_config,  # type: ignore[arg-type]
             search_path=self.session.infrahub_config_path.parent,  # type: ignore[attr-defined]
         )
+        self.check_instance = check_class()
 
     def run_check(self, variables: dict[str, Any]) -> Any:
         self.instantiate_check()

infrahub_sdk/pytest_plugin/items/python_transform.py

@@ -6,7 +6,7 @@
 import ujson
 from httpx import HTTPStatusError
 
-from ...transforms import get_transform_class_instance
+from ...import_utils import get_check_or_transform_class
 from ..exceptions import OutputMatchError, PythonTransformDefinitionError
 from ..models import InfrahubTestExpectedResult
 from .base import InfrahubItem
@@ -33,10 +33,11 @@ def __init__(
         self.transform_instance: InfrahubTransform
 
     def instantiate_transform(self) -> None:
-        self.transform_instance = get_transform_class_instance(
-            transform_config=self.resource_config,  # type: ignore[arg-type]
+        transform_class = get_check_or_transform_class(
+            config=self.resource_config,  # type: ignore[arg-type]
             search_path=self.session.infrahub_config_path.parent,  # type: ignore[attr-defined]
         )
+        self.transform_instance = transform_class(branch="", client=None)
 
     def run_transform(self, variables: dict[str, Any]) -> Any:
         self.instantiate_transform()

infrahub_sdk/transforms.py

@@ -1,20 +1,16 @@
 from __future__ import annotations
 
 import asyncio
-import importlib
 import os
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Any, Optional
 
 from git import Repo
 
-from .exceptions import InfrahubTransformNotFoundError, UninitializedError
+from .exceptions import UninitializedError
 
 if TYPE_CHECKING:
-    from pathlib import Path
-
     from . import InfrahubClient
-    from .schema.repository import InfrahubPythonTransformConfig
 
 INFRAHUB_TRANSFORM_VARIABLE_TO_IMPORT = "INFRAHUB_TRANSFORMS"
 
@@ -95,40 +91,3 @@ async def run(self, data: Optional[dict] = None) -> Any:
             return await self.transform(data=unpacked)
 
         return self.transform(data=unpacked)
-
-
-def get_transform_class_instance(
-    transform_config: InfrahubPythonTransformConfig,
-    search_path: Optional[Path] = None,
-    branch: str = "",
-    client: Optional[InfrahubClient] = None,
-) -> InfrahubTransform:
-    """Gets an instance of the InfrahubTransform class.
-
-    Args:
-        transform_config: A config object with information required to find and load the transform.
-        search_path: The path in which to search for a python file containing the transform. The current directory is
-            assumed if not speicifed.
-        branch: Infrahub branch which will be targeted in graphql query used to acquire data for transformation.
-        client: InfrahubClient used to interact with infrahub API.
-    """
-    if transform_config.file_path.is_absolute() or search_path is None:
-        search_location = transform_config.file_path
-    else:
-        search_location = search_path / transform_config.file_path
-
-    try:
-        spec = importlib.util.spec_from_file_location(transform_config.class_name, search_location)
-        module = importlib.util.module_from_spec(spec)  # type: ignore[arg-type]
-        spec.loader.exec_module(module)  # type: ignore[union-attr]
-
-        # Get the specified class from the module
-        transform_class = getattr(module, transform_config.class_name)
-
-        # Create an instance of the class
-        transform_instance = transform_class(branch=branch, client=client)
-
-    except (FileNotFoundError, AttributeError) as exc:
-        raise InfrahubTransformNotFoundError(name=transform_config.name) from exc
-
-    return transform_instance