fix(Script): change the way to manipulate docker information

sekoia_automation/scripts/sync_library.py

@@ -249,21 +249,20 @@ def load_connectors(self, module_path: Path) -> list:
 
         return connectors
 
-    def set_docker(self, manifests: list, module: dict) -> list:
+    def set_docker(self, manifests: list, module_docker_image: str) -> list:
         """Loops over the Docker name of objets linked to a module and adds the Docker
         version if missing
 
         Args:
             manifests (list): List of dict representing the objects to be checked
-            module (dict): Data dict of the parent module
+            module_docker_image (str): The docker image as defined in the module
 
         Returns:
             list: Modified version of the manifests received as parameter
         """
-        module_docker_name = self._get_module_docker_name(module)
         for manifest in manifests:
-            if "docker" not in manifest or manifest["docker"] == module_docker_name:
-                manifest["docker"] = f"{module_docker_name}:{module['version']}"
+            if "docker" not in manifest or manifest["docker"] != module_docker_image:
+                manifest["docker"] = module_docker_image
 
         return manifests
 
@@ -363,7 +362,6 @@ def load_module(self, module_path: Path):
             module_info = json.load(fd)
 
         docker_name = self._get_module_docker_name(module_info)
-        module_info["docker"] = f"{docker_name}:{module_info['version']}"
         if self.registry_check and not self.check_image_on_registry(
             docker_name, module_info["version"]
         ):
@@ -373,9 +371,13 @@ def load_module(self, module_path: Path):
             )
             raise typer.Exit(code=1)
 
-        triggers = self.set_docker(self.load_triggers(module_path), module_info)
-        connectors = self.set_docker(self.load_connectors(module_path), module_info)
-        actions = self.set_docker(self.load_actions(module_path), module_info)
+        module_docker_image = f"{docker_name}:{module_info['version']}"
+        module_info["docker"] = module_docker_image
+        triggers = self.set_docker(self.load_triggers(module_path), module_docker_image)
+        connectors = self.set_docker(
+            self.load_connectors(module_path), module_docker_image
+        )
+        actions = self.set_docker(self.load_actions(module_path), module_docker_image)
 
         module_uuid: str = module_info["uuid"]
         module_name: str = module_info["name"]
@@ -453,8 +455,6 @@ def execute(self):
             )
 
     def _get_module_docker_name(self, manifest: dict) -> str:
-        if docker := manifest.get("docker"):
-            return docker
         if slug := manifest.get("slug"):
             return f"{self.registry}/{self.namespace}/{self.DOCKER_PREFIX}-{slug}"
         raise ValueError("Impossible to generate image name")

tests/data/sample_module/manifest.json

@@ -2,11 +2,11 @@
     "name": "Test Module",
     "description": "TBD",
     "uuid": "7cf9ccf6-e20a-4e53-bd77-3c19cdc94154",
-    "docker": "sekoia-automation-module-sample",
+    "slug": "sample",
     "configuration": {},
     "version": "0.1",
     "categories": [
         "Endpoint",
         "Network"
     ]
-}
+}

tests/scripts/test_sync_library.py

@@ -204,52 +204,55 @@ def test_with_module_invalid_name():
 
 @requests_mock.Mocker(kw="m")
 def test_registry_check_default_success(module, **kwargs):
+    lib = SyncLibrary(SYMPOHNY_URL, API_KEY, Path("tests/data"), registry_pat=PAT)
     kwargs["m"].register_uri(
         "GET",
         re.compile("https://ghcr.io/.*"),
         status_code=200,
         json={"token": API_KEY},
     )
-    assert SyncLibrary(
-        SYMPOHNY_URL, API_KEY, Path("tests/data"), registry_pat=PAT
-    ).check_image_on_registry(module["docker"], module["version"])
+    assert lib.check_image_on_registry(
+        lib._get_module_docker_name(module), module["version"]
+    )
 
     history = kwargs["m"].request_history
     assert len(history) == 1
     assert history[0].method == "GET"
     assert (
         history[0].url
-        == f"https://ghcr.io/v2/sekoia-io/{module['docker']}/\
+        == f"https://ghcr.io/v2/sekoia-io/{lib.DOCKER_PREFIX}-{module['slug']}/\
 manifests/{module['version']}"
     )
 
 
 @requests_mock.Mocker(kw="m")
 def test_registry_check_default_fail(module, **kwargs):
+    lib = SyncLibrary(SYMPOHNY_URL, API_KEY, Path("tests/data"), PAT)
     kwargs["m"].register_uri(
         "GET",
         re.compile("https://ghcr.io/v2/sekoia-io.*"),
         status_code=404,
     )
-    assert not SyncLibrary(
-        SYMPOHNY_URL, API_KEY, Path("tests/data"), PAT
-    ).check_image_on_registry(module["docker"], module["version"])
+    assert not lib.check_image_on_registry(
+        lib._get_module_docker_name(module), module["version"]
+    )
 
     history = kwargs["m"].request_history
     assert len(history) == 1
     assert history[0].method == "GET"
     assert (
         history[0].url
-        == f"https://ghcr.io/v2/sekoia-io/{module['docker']}/\
+        == f"https://ghcr.io/v2/sekoia-io/{lib.DOCKER_PREFIX}-{module['slug']}/\
 manifests/{module['version']}"
     )
 
 
 @requests_mock.Mocker(kw="m")
 def test_registry_check_custom_success(module, **kwargs):
+    lib = SyncLibrary(SYMPOHNY_URL, API_KEY, Path("tests/data"), PAT)
     custom_path = "foo.bar"
     custom_pathinfo = "sekoia-io"
-    image_name = module["docker"]
+    image_name = lib._get_module_docker_name(module)
     module["docker"] = f"{custom_path}/{custom_pathinfo}/{image_name}"
 
     kwargs["m"].register_uri(
@@ -258,9 +261,7 @@ def test_registry_check_custom_success(module, **kwargs):
         status_code=200,
         json={"token": API_KEY},
     )
-    assert SyncLibrary(
-        SYMPOHNY_URL, API_KEY, Path("tests/data"), PAT
-    ).check_image_on_registry(module["docker"], module["version"])
+    assert lib.check_image_on_registry(module["docker"], module["version"])
 
     history = kwargs["m"].request_history
     assert len(history) == 1
@@ -276,20 +277,21 @@ def test_registry_check_custom_success(module, **kwargs):
 def test_registry_check_not_found(module, **kwargs):
     custom_path = "foo.bar"
     custom_pathinfo = "sekoia-io"
-    image_name = module["docker"]
-    module["docker"] = f"{custom_path}/{custom_pathinfo}/{image_name}"
+    lib = SyncLibrary(
+        SYMPOHNY_URL,
+        API_KEY,
+        Path("tests/data"),
+        registry=custom_path,
+        namespace=custom_pathinfo,
+    )
+    module["docker"] = lib._get_module_docker_name(module)
 
     kwargs["m"].register_uri(
         "GET",
-        f"https://{custom_path}/v2/sekoia-io/sekoia-automation-module-sample/manifests/0.1",
+        f"https://{custom_path}/v2/sekoia-io/automation-module-sample/manifests/0.1",
         status_code=404,
     )
-    assert (
-        SyncLibrary(
-            SYMPOHNY_URL, API_KEY, Path("tests/data"), PAT, USER
-        ).check_image_on_registry(module["docker"], module["version"])
-        is False
-    )
+    assert lib.check_image_on_registry(module["docker"], module["version"]) is False
 
     history = kwargs["m"].request_history
     assert len(history) == 1
@@ -325,9 +327,6 @@ def test_get_module_docker_name():
     lib = SyncLibrary(SYMPOHNY_URL, API_KEY, Path("tests/data"))
 
     manifest = {"docker": "foo", "slug": "bar"}
-    assert lib._get_module_docker_name(manifest) == "foo"
-
-    manifest.pop("docker")
     assert (
         lib._get_module_docker_name(manifest)
         == "ghcr.io/sekoia-io/automation-module-bar"