fix(connector) Use intake URL from file

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.12.2] - 2024-03-26
+
+### Fixed
+
+- Use file for batch url computation
+
 ## [1.12.1] - 2024-03-22
 
 ### Fixed

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry]
 name = "sekoia-automation-sdk"
 
-version = "1.12.1"
+version = "1.12.2"
 description = "SDK to create Sekoia.io playbook modules"
 license = "MIT"
 readme = "README.md"

sekoia_automation/aio/connector.py

@@ -1,6 +1,5 @@
 """Contains connector with async version."""
 
-import os
 from abc import ABC
 from asyncio import AbstractEventLoop, get_event_loop
 from collections.abc import AsyncGenerator
@@ -154,8 +153,10 @@ async def push_data_to_intakes(
             list[str]:
         """
         self._last_events_time = datetime.utcnow()
-        intake_host = os.getenv("INTAKE_URL", self.configuration.intake_server)
-        batch_api = urljoin(intake_host, "batch")
+        if intake_server := self.configuration.intake_server:
+            batch_api = urljoin(intake_server, "batch")
+        else:
+            batch_api = urljoin(self.intake_url, "batch")
 
         result_ids = []
 

sekoia_automation/connector/__init__.py

@@ -1,4 +1,3 @@
-import os
 import uuid
 from abc import ABC
 from collections.abc import Generator, Sequence
@@ -36,7 +35,7 @@
 
 
 class DefaultConnectorConfiguration(BaseModel):
-    intake_server: str = "https://intake.sekoia.io"
+    intake_server: str | None = None
     intake_key: str
 
 
@@ -171,8 +170,10 @@ def push_events_to_intakes(
         # Reset the consecutive error count
         self._error_count = 0
         self._last_events_time = datetime.utcnow()
-        intake_host = os.getenv("INTAKE_URL", self.configuration.intake_server)
-        batch_api = urljoin(intake_host, "batch")
+        if intake_server := self.configuration.intake_server:
+            batch_api = urljoin(intake_server, "batch")
+        else:
+            batch_api = urljoin(self.intake_url, "batch")
 
         # Dict to collect event_ids for the API
         collect_ids: dict[int, list] = {}

sekoia_automation/module.py

@@ -322,6 +322,7 @@ class ModuleItem(ABC):
     CALLBACK_URL_FILE_NAME = "url_callback"
     SECRETS_URL_FILE_NAME = "url_secrets"
     LOGS_URL_FILE_NAME = "url_logs"
+    INTAKE_URL_FILE_NAME = "intake_url"
 
     name: str | None = None
     description: str | None = None
@@ -419,6 +420,10 @@ def secrets_url(self) -> str:
         except FileNotFoundError:
             return self.callback_url.replace("/callback", "/secrets")
 
+    @cached_property
+    def intake_url(self) -> str:
+        return self.module.load_config(self.INTAKE_URL_FILE_NAME)
+
     @property
     def _headers(self) -> dict[str, str]:
         return {"Authorization": f"Bearer {self.token}"}

tests/aio/test_connector.py

@@ -1,6 +1,7 @@
 """Test async connector."""
 
-from unittest.mock import Mock, patch
+import json
+from unittest.mock import Mock, mock_open, patch
 from urllib.parse import urljoin
 
 import pytest

tests/connectors/test_connector.py

@@ -1,4 +1,3 @@
-import os
 from unittest.mock import Mock, PropertyMock, patch
 
 import pytest
@@ -14,6 +13,12 @@
 EVENTS = ["foo", "bar"]
 
 
+@pytest.fixture(autouse=True)
+def configure_intake_url(config_storage):
+    with (config_storage / "intake_url").open("w") as f:
+        f.write("https://intake.sekoia.io")
+
+
 class DummyConnector(Connector):
     def run(self):
         raise NotImplementedError
@@ -136,25 +141,42 @@ def test_push_event_to_intake_with_chunks(test_connector, mocked_trigger_logs):
     assert result == ["001", "002", "003", "004"]
 
 
-def test_push_event_to_intake_custom_url(test_connector, mocked_trigger_logs):
+def test_push_event_to_intake_custom_url(
+    test_connector, mocked_trigger_logs, config_storage
+):
+    assert test_connector.configuration.intake_server is None
+
     url = "https://fra2.app.sekoia.io/v1/intake-http/batch"
     batch_mock = mocked_trigger_logs.post(
         url, json={"event_ids": ["001"]}, additional_matcher=match_events("foo")
     )
     # With trailing slash
-    with patch.dict(
-        os.environ, {"INTAKE_URL": "https://fra2.app.sekoia.io/v1/intake-http/"}
-    ):
-        test_connector.push_events_to_intakes(["foo"])
-        assert batch_mock.call_count == 1
+    with (config_storage / "intake_url").open("w") as f:
+        f.write("https://fra2.app.sekoia.io/v1/intake-http/")
+    test_connector.push_events_to_intakes(["foo"])
+    assert batch_mock.call_count == 1
 
     # Without trailing slash
+    with (config_storage / "intake_url").open("w") as f:
+        f.write("https://fra2.app.sekoia.io/v1/intake-http")
     mocked_trigger_logs.reset_mock()
-    with patch.dict(
-        os.environ, {"INTAKE_URL": "https://fra2.app.sekoia.io/v1/intake-http"}
-    ):
-        test_connector.push_events_to_intakes(["foo"])
-        assert batch_mock.call_count == 1
+    test_connector.push_events_to_intakes(["foo"])
+    assert batch_mock.call_count == 1
+
+
+def test_push_event_to_intake_custom_url_configuration(
+    test_connector, mocked_trigger_logs
+):
+    url = "https://fra2.app.sekoia.io/v1/intake-http/batch"
+    batch_mock = mocked_trigger_logs.post(
+        url, json={"event_ids": ["001"]}, additional_matcher=match_events("foo")
+    )
+
+    test_connector.configuration.intake_server = (
+        "https://fra2.app.sekoia.io/v1/intake-http"
+    )
+    test_connector.push_events_to_intakes(["foo"])
+    assert batch_mock.call_count == 1
 
 
 def test_push_event_to_intake_with_chunks_executor_stopped(

tests/expectations/sample_module/connector_sample_connector.json

@@ -9,7 +9,6 @@
     "properties": {
       "intake_server": {
         "title": "Intake Server",
-        "default": "https://intake.sekoia.io",
         "type": "string"
       },
       "intake_key": {
@@ -22,4 +21,4 @@
     ]
   },
   "results": {}
-}
+}

tests/test_trigger.py

@@ -61,6 +61,13 @@ def test_secrets_url():
     mock.assert_called_with(trigger.SECRETS_URL_FILE_NAME)
 
 
+def test_intake_url():
+    trigger = DummyTrigger()
+    with patch.object(Module, "load_config", return_value="intake") as mock:
+        assert trigger.intake_url == "intake"
+    mock.assert_called_with(trigger.INTAKE_URL_FILE_NAME)
+
+
 def test_logs_url():
     trigger = DummyTrigger()
     with patch.object(Module, "load_config", return_value="logs") as mock: