fix: de pull fails for images with digest in URI

src/aap_eda/services/activation/engine/podman.py

@@ -25,6 +25,7 @@
 from aap_eda.core.enums import ActivationStatus
 from aap_eda.settings import features
 from aap_eda.utils import str_to_bool
+from aap_eda.utils.podman import parse_repository
 
 from . import exceptions, messages
 from .common import (
@@ -379,7 +380,12 @@ def _pull_image(
                     "username": request.credential.username,
                     "password": request.credential.secret,
                 }
-            image = self.client.images.pull(request.image_url, **kwargs)
+
+            image_repository, image_tag = parse_repository(request.image_url)
+
+            image = self.client.images.pull(
+                repository=image_repository, tag=image_tag, **kwargs
+            )
 
             LOGGER.info("Downloaded image")
             return image

src/aap_eda/utils/podman.py

@@ -0,0 +1,33 @@
+#  Copyright 2025 Red Hat, Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+from typing import Optional
+
+
+# temporary parser until bug is fixed in podman-py
+# https://github.com/containers/podman-py/pull/555
+def parse_repository(name: str) -> tuple[str, Optional[str]]:
+    """Parse repository image name from tag or digest.
+
+    Returns:
+        item 1: repository name
+        item 2: Either tag or None
+    """
+    # split repository and image name from tag
+    # tags need to be split from the right since
+    # a port number might increase the split list len by 1
+    elements = name.rsplit(":", 1)
+    if len(elements) == 2 and "/" not in elements[1]:
+        return elements[0], elements[1]
+
+    return name, None

tests/integration/services/activation/engine/test_podman.py

@@ -48,6 +48,7 @@
 from aap_eda.settings.default import (  # noqa: N811
     DYNACONF as orig_dynaconf_settings,
 )
+from aap_eda.utils.podman import parse_repository
 
 from .utils import InitData, get_ansible_rulebook_cmdline, get_request
 
@@ -262,16 +263,33 @@ def test_engine_start_with_none_image_url(
         engine.start(request, log_handler)
 
 
+@pytest.mark.parametrize(
+    "image_url",
+    [
+        "quay.io/ansible/ansible-rulebook",
+        "quay.io:5004/ansible/ansible-rulebook",
+        "quay.io:5004/ansible/ansible-rulebook:main",
+        "quay.io/ansible/ansible-rulebook@sha256:756dd6a40b62975ea10d83a577ed6b1cddd545de17d9fb56b3d4e80e166826df",  # noqa: E501
+        "quay.io:5004/ansible/ansible-rulebook@sha256:756dd6a40b62975ea10d83a577ed6b1cddd545de17d9fb56b3d4e80e166826df",  # noqa: E501
+        "quay.io:443/ansible/ansible-rulebook@sha512:756dd6a40b62975ea10d83a577ed6b1cddd545de17d9fb56b3d4e80e166826df",  # noqa: E501
+        "quay.io/ansible/ansible-rulebook:42@sha256:756dd6a40b62975ea10d83a577ed6b1cddd545de17d9fb56b3d4e80e166826df",  # noqa: E501
+    ],
+)
 @pytest.mark.django_db
 def test_engine_start_with_credential(
     init_podman_data,
     podman_engine,
     default_organization: models.Organization,
+    image_url,
 ):
     engine = podman_engine
     log_handler = DBLogger(init_podman_data.activation_instance.id)
     request = get_request(
-        init_podman_data, "me", default_organization, mounts=[{"/dev": "/opt"}]
+        init_podman_data,
+        "me",
+        default_organization,
+        image_url,
+        mounts=[{"/dev": "/opt"}],
     )
     credential = Credential(
         username="me",
@@ -283,8 +301,10 @@ def test_engine_start_with_credential(
 
     engine.start(request, log_handler)
 
+    image_repo, image_tag = parse_repository(request.image_url)
     engine.client.images.pull.assert_called_once_with(
-        request.image_url,
+        repository=image_repo,
+        tag=image_tag,
         tls_verify=bool(credential.ssl_verify),
         auth_config={
             "username": credential.username,

tests/integration/services/activation/engine/utils.py

@@ -46,11 +46,12 @@ def get_request(
     data: InitData,
     username: str,
     default_organization: models.Organization,
+    image_url: str = "quay.io/ansible/ansible-rulebook:main",
     **kwargs,
 ):
     return ContainerRequest(
         name="test-request",
-        image_url="quay.io/ansible/ansible-rulebook:main",
+        image_url=image_url,
         rulebook_process_id=data.activation_instance.id,
         process_parent_id=data.activation.id,
         cmdline=get_ansible_rulebook_cmdline(data),