[connectors-sdk] BaseConnectorSettings support for connectors as module (#5295)

Author: Powlinett

connectors-sdk/connectors_sdk/settings/base_settings.py

@@ -6,14 +6,13 @@
 These models can be extended to create specific configurations for different types of connectors.
 """
 
-import os
+import sys
 from abc import ABC
 from copy import deepcopy
 from datetime import timedelta
 from pathlib import Path
 from typing import Any, Literal, Self
 
-import __main__
 from connectors_sdk.settings.annotated_types import ListFromString
 from connectors_sdk.settings.exceptions import ConfigValidationError
 from pydantic import (
@@ -87,6 +86,47 @@ class _SettingsLoader(BaseSettings):
         enable_decoding=False,
     )
 
+    @classmethod
+    def _get_connector_main_path(cls) -> Path:
+        """Locate the main module of the running connector.
+        This method is used to locate configuration files relative to connector's entrypoint.
+
+        Notes:
+            - This method assumes that the connector is launched using a file-backed entrypoint
+            (i.e., `python -m <module>` or `python <file>`).
+            - At module import time, `__main__.__file__` might not be available yet,
+            thus this method should be called at runtime only.
+        """
+        main = sys.modules.get("__main__")
+        if main and getattr(main, "__file__", None):
+            return Path(main.__file__).resolve()  # type: ignore
+
+        raise RuntimeError(
+            "Cannot determine connector's location: __main__.__file__ is not available. "
+            "Ensure the connector is launched using `python -m <module>` or a file-backed entrypoint."
+        )
+
+    @classmethod
+    def _get_config_yml_file_path(cls) -> Path | None:
+        """Locate the `config.yml` file of the running connector."""
+        main_path = cls._get_connector_main_path()
+        config_yml_legacy_file_path = main_path.parent / "config.yml"
+        config_yml_file_path = main_path.parent.parent / "config.yml"
+
+        if config_yml_legacy_file_path.is_file():
+            return config_yml_legacy_file_path
+        elif config_yml_file_path.is_file():
+            return config_yml_file_path
+        return None
+
+    @classmethod
+    def _get_dot_env_file_path(cls) -> Path | None:
+        """Locate the `.env` file of the running connector."""
+        main_path = cls._get_connector_main_path()
+        dot_env_file_path = main_path.parent.parent / ".env"
+
+        return dot_env_file_path if dot_env_file_path.is_file() else None
+
     @classmethod
     def settings_customise_sources(
         cls,
@@ -109,26 +149,22 @@ def settings_customise_sources(
             1. If a config.yml file is found, the order will be: `ENV VAR` → config.yml → default value
             2. If a .env file is found, the order will be: `ENV VAR` → .env → default value
         """
-        _main_path = os.path.dirname(os.path.abspath(__main__.__file__))
-
-        settings_cls.model_config["env_file"] = f"{_main_path}/../.env"
-
-        if not settings_cls.model_config["yaml_file"]:
-            if Path(f"{_main_path}/config.yml").is_file():
-                settings_cls.model_config["yaml_file"] = f"{_main_path}/config.yml"
-            if Path(f"{_main_path}/../config.yml").is_file():
-                settings_cls.model_config["yaml_file"] = f"{_main_path}/../config.yml"
-
-        if Path(settings_cls.model_config["yaml_file"] or "").is_file():  # type: ignore
+        config_yml_file_path = cls._get_config_yml_file_path()
+        if config_yml_file_path:
+            settings_cls.model_config["yaml_file"] = config_yml_file_path
             return (
                 env_settings,
                 YamlConfigSettingsSource(settings_cls),
             )
-        if Path(settings_cls.model_config["env_file"] or "").is_file():  # type: ignore
+
+        dot_env_file_path = cls._get_dot_env_file_path()
+        if dot_env_file_path:
+            settings_cls.model_config["env_file"] = dot_env_file_path
             return (
                 env_settings,
                 DotEnvSettingsSource(settings_cls),
             )
+
         return (env_settings,)
 
     @classmethod

connectors-sdk/tests/test_models/__init__.py

@@ -1,8 +1,23 @@
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
 import pytest
 
 
 @pytest.fixture
-def mock_basic_environment(monkeypatch):
+def mock_main_path(monkeypatch):
+    """Mock the path of `__main__.__file__` for `_SettingsLoader._get_connector_main_path` calls."""
+
+    monkeypatch.setitem(
+        sys.modules, "__main__", SimpleNamespace(__file__="/app/src/main.py")
+    )
+
+
+@pytest.fixture
+def mock_environment(monkeypatch):
+    """Mock `os.environ` for `_SettingsLoader` and `BaseConnectorSettings` calls."""
+
     monkeypatch.setenv("OPENCTI_URL", "http://localhost:8080")
     monkeypatch.setenv("OPENCTI_TOKEN", "changeme")
     monkeypatch.setenv("CONNECTOR_ID", "connector-poc--uid")
@@ -13,60 +28,26 @@ def mock_basic_environment(monkeypatch):
 
 
 @pytest.fixture
-def mock_yaml_file_presence(monkeypatch):
-    def is_file(self):
-        if self.name == "config.yml":
-            return True
-        return False
-
-    monkeypatch.setattr("pathlib.Path.is_file", is_file)
+def mock_config_yml_file_presence(monkeypatch):
+    """Mock the path of `config.yml` for `_SettingsLoader` and `BaseConnectorSettings` calls."""
 
-
-@pytest.fixture
-def mock_env_file_presence(monkeypatch):
-    def is_file(self):
-        if self.name == ".env":
-            return True
-        return False
-
-    monkeypatch.setattr("pathlib.Path.is_file", is_file)
-
-
-@pytest.fixture
-def mock_yaml_config_settings_read_files(monkeypatch):
-    def read_files(_, __):
-        return {
-            "connector": {
-                "duration_period": "PT5M",
-                "id": "connector-poc--uid",
-                "log_level": "error",
-                "name": "Test Connector",
-                "scope": "test",
-            },
-            "opencti": {"token": "changeme", "url": "http://localhost:8080"},
-        }
+    def get_config_yml_file_path():
+        return Path(__file__).parent / "data" / "config.test.yml"
 
     monkeypatch.setattr(
-        "pydantic_settings.YamlConfigSettingsSource._read_files", read_files
+        "connectors_sdk.settings.base_settings._SettingsLoader._get_config_yml_file_path",
+        get_config_yml_file_path,
     )
 
 
 @pytest.fixture
-def mock_env_config_settings_read_env_files(monkeypatch):
-    def _read_env_files(self):
-        if self.settings_cls.__name__ == "SettingsLoader":
-            return {
-                "opencti": {"url": "http://localhost:8080", "token": "changeme"},
-                "connector": {
-                    "id": "connector-poc--uid",
-                    "name": "Test Connector",
-                    "duration_period": "PT5M",
-                    "log_level": "error",
-                    "scope": "test",
-                },
-            }
-        return {}
+def mock_dot_env_file_presence(monkeypatch):
+    """Mock the path of `.env` for `_SettingsLoader` and `BaseConnectorSettings` calls."""
+
+    def get_dot_env_file_path():
+        return Path(__file__).parent / "data" / ".env.test"
 
     monkeypatch.setattr(
-        "pydantic_settings.DotEnvSettingsSource._load_env_vars", _read_env_files
+        "connectors_sdk.settings.base_settings._SettingsLoader._get_dot_env_file_path",
+        get_dot_env_file_path,
     )

connectors-sdk/tests/test_settings/__init__.py

@@ -0,0 +1,7 @@
+OPENCTI_URL=http://localhost:8080
+OPENCTI_TOKEN=changeme
+CONNECTOR_ID=connector-poc--uid
+CONNECTOR_NAME=Test Connector
+CONNECTOR_SCOPE=test
+CONNECTOR_LOG_LEVEL=error
+CONNECTOR_DURATION_PERIOD=PT5M

connectors-sdk/tests/test_settings/conftest.py

@@ -0,0 +1,11 @@
+opencti:
+  url: http://localhost:8080
+  token: changeme
+
+connector:
+  id: connector-poc--uid
+  name: Test Connector
+  scope: test
+  log_level: error
+  duration_period: PT5M
+