plugins/module_utils/_common.py

@@ -27,7 +27,6 @@ from ansible_collections.community.docker.plugins.module_utils._util import (
     DOCKER_COMMON_ARGS,
     DOCKER_MUTUALLY_EXCLUSIVE,
     DOCKER_REQUIRED_TOGETHER,
-    build_pull_arguments,
     filter_images_by_tag,
     sanitize_result,
     update_tls_hostname,
@@ -573,22 +572,17 @@ class AnsibleDockerClientBase(Client):
         """
         Pull an image
         """
-        kwargs: dict[str, t.Any] = {
+        kwargs = {
+            "tag": tag,
             "stream": True,
             "decode": True,
         }
         if image_platform is not None:
             kwargs["platform"] = image_platform
-
-        # Build pull arguments - handles combined tag@digest format
-        pull_name, pull_tag = build_pull_arguments(name, tag)
-        if pull_tag is not None:
-            kwargs["tag"] = pull_tag
-
         self.log(f"Pulling image {name}:{tag}")
         old_tag = self.find_image(name, tag)
         try:
-            for line in self.pull(pull_name, **kwargs):
+            for line in self.pull(name, **kwargs):
                 self.log(line, pretty_print=True)
                 if line.get("error"):
                     if line.get("errorDetail"):

plugins/module_utils/_common_api.py

@@ -56,7 +56,6 @@ from ansible_collections.community.docker.plugins.module_utils._util import (
     DOCKER_COMMON_ARGS,
     DOCKER_MUTUALLY_EXCLUSIVE,
     DOCKER_REQUIRED_TOGETHER,
-    build_pull_arguments,
     filter_images_by_tag,
     sanitize_result,
     update_tls_hostname,
@@ -532,15 +531,10 @@ class AnsibleDockerClientBase(Client):
         try:
             repository, image_tag = parse_repository_tag(name)
             registry, dummy_repo_name = auth.resolve_repository_name(repository)
-
-            # Build pull arguments - handles combined tag@digest format
-            effective_tag = tag or image_tag or "latest"
-            pull_name, pull_tag = build_pull_arguments(repository, effective_tag)
-
-            params: dict[str, t.Any] = {"fromImage": pull_name}
-            if pull_tag is not None:
-                params["tag"] = pull_tag
-
+            params = {
+                "tag": tag or image_tag or "latest",
+                "fromImage": repository,
+            }
             if image_platform is not None:
                 params["platform"] = image_platform
 

plugins/module_utils/_util.py

@@ -125,13 +125,13 @@ def filter_images_by_tag(
     # The @ is for digest, but if it doesn't start with sha256:, it's combined format
     if "@" in tag and not tag.startswith("sha256:"):
         tag_part, digest_part = tag.split("@", 1)
+        lookup_tag = f"{name}:{tag_part}"
         lookup_digest = f"{name}@{digest_part}"
         for image in images:
+            repo_tags = image.get("RepoTags") or []
             repo_digests = image.get("RepoDigests") or []
-            # When digest is specified, match by digest only.
-            # Docker doesn't preserve the tag in RepoTags when pulling by digest,
-            # so we can't require both to match.
-            if lookup_digest in repo_digests:
+            # For combined format, match BOTH tag AND digest
+            if lookup_tag in repo_tags and lookup_digest in repo_digests:
                 return [image]
         return []
 
@@ -146,36 +146,6 @@ def filter_images_by_tag(
     return []
 
 
-def build_pull_arguments(
-    name: str, tag: str
-) -> tuple[str, str | None]:
-    """
-    Build the correct arguments for Docker pull operations.
-
-    Docker SDK and API don't accept combined "tag@digest" in the tag parameter.
-    This function handles the three formats:
-    - Tag-only: "v1.0" -> (name, "v1.0")
-    - Digest-only: "sha256:abc..." -> (name, "sha256:abc...")
-    - Combined tag@digest: "v1.0@sha256:abc..." -> ("name:v1.0@sha256:abc...", None)
-
-    Args:
-        name: Repository name (e.g., "nginx", "ghcr.io/user/repo")
-        tag: Tag, digest, or combined tag@digest
-
-    Returns:
-        Tuple of (pull_name, pull_tag) where:
-        - For combined format: pull_name is full reference, pull_tag is None
-        - For other formats: pull_name is just the name, pull_tag is the tag/digest
-    """
-    # Handle combined tag@digest format (e.g., "v1.0@sha256:abc123")
-    # The @ indicates a digest, but if it doesn't START with sha256:, it's combined
-    if "@" in tag and not tag.startswith("sha256:"):
-        tag_part, digest = tag.split("@", 1)
-        return f"{name}:{tag_part}@{digest}", None
-    else:
-        return name, tag
-
-
 def is_valid_tag(tag: str, allow_empty: bool = False) -> bool:
     """Check whether the given string is a valid docker tag name."""
     if not tag:

plugins/modules/docker_image.py

@@ -405,7 +405,6 @@ from ansible_collections.community.docker.plugins.module_utils._image_archive im
 )
 from ansible_collections.community.docker.plugins.module_utils._util import (
     DockerBaseClass,
-    build_pull_arguments,
     clean_dict_booleans_for_docker_api,
     is_image_name_id,
     is_valid_tag,
@@ -785,10 +784,6 @@ class ImageManager(DockerBaseClass):
                         push_repository, push_tag = parse_repository_tag(
                             push_repository
                         )
-
-                    # Handle combined tag@digest format - Docker API doesn't accept it in tag param
-                    push_name, push_tag_param = build_pull_arguments(push_repository, push_tag)
-
                     push_registry, dummy = resolve_repository_name(push_repository)
                     headers = {}
                     header = get_config_header(self.client, push_registry)
@@ -797,17 +792,12 @@ class ImageManager(DockerBaseClass):
                         # See https://github.com/moby/moby/issues/50614.
                         header = base64.urlsafe_b64encode(b"{}")
                     headers["X-Registry-Auth"] = header
-
-                    params: dict[str, t.Any] = {}
-                    if push_tag_param is not None:
-                        params["tag"] = push_tag_param
-
                     response = self.client._post_json(
-                        self.client._url("/images/{0}/push", push_name),
+                        self.client._url("/images/{0}/push", push_repository),
                         data=None,
                         headers=headers,
                         stream=True,
-                        params=params,
+                        params={"tag": push_tag},
                     )
                     self.client._raise_for_status(response)
                     for line in self.client._stream_helper(response, decode=True):

tests/unit/plugins/module_utils/test__util.py

@@ -544,11 +544,9 @@ class TestFilterImagesByTag:
         assert len(result) == 0
 
     # ========== Combined tag@digest tests ==========
-    # When a digest is specified, we match by digest ONLY because Docker doesn't
-    # preserve tags in RepoTags when pulling by digest. The tag is informational.
 
-    def test_combined_tag_digest_matches_by_digest(self) -> None:
-        """Combined tag@digest should match when digest matches (tag is informational)."""
+    def test_combined_tag_digest_matches_both(self) -> None:
+        """Combined tag@digest should match when BOTH tag AND digest match."""
         image = _create_mock_image(
             repo_tags=["nginx:1.21"],
             repo_digests=["nginx@sha256:" + SHA],
@@ -568,18 +566,14 @@ class TestFilterImagesByTag:
         )
         assert len(result) == 1
 
-    def test_combined_tag_digest_matches_even_if_tag_differs(self) -> None:
-        """Combined tag@digest SHOULD match even if tag differs - digest is authoritative.
-
-        Docker doesn't store tags when pulling by digest, so requiring tag match
-        would cause lookups to always fail. The digest is the authoritative identifier.
-        """
+    def test_combined_tag_digest_fails_if_tag_mismatch(self) -> None:
+        """Combined tag@digest should NOT match if tag doesn't match."""
         image = _create_mock_image(
-            repo_tags=["nginx:1.20"],  # Different tag - doesn't matter
+            repo_tags=["nginx:1.20"],  # Different tag
             repo_digests=["nginx@sha256:" + SHA],
         )
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
-        assert len(result) == 1  # Should match by digest
+        assert len(result) == 0
 
     def test_combined_tag_digest_fails_if_digest_mismatch(self) -> None:
         """Combined tag@digest should NOT match if digest doesn't match."""
@@ -591,40 +585,27 @@ class TestFilterImagesByTag:
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
         assert len(result) == 0
 
-    def test_combined_tag_digest_fails_if_digest_mismatch_tag_irrelevant(self) -> None:
-        """Combined tag@digest should NOT match if digest doesn't match, regardless of tag."""
+    def test_combined_tag_digest_fails_if_both_mismatch(self) -> None:
+        """Combined tag@digest should NOT match if both tag and digest don't match."""
         other_sha = "a" * 64
         image = _create_mock_image(
-            repo_tags=["nginx:1.20"],  # Different tag - irrelevant
-            repo_digests=["nginx@sha256:" + other_sha],  # Different digest - this matters
+            repo_tags=["nginx:1.20"],  # Different tag
+            repo_digests=["nginx@sha256:" + other_sha],  # Different digest
         )
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
         assert len(result) == 0
 
     # ========== Edge cases ==========
 
-    def test_empty_repo_tags_matches_by_digest(self) -> None:
-        """Handle images where RepoTags is empty or None - should still match by digest.
-
-        This is the critical real-world case: Docker doesn't store tags when pulling
-        by digest, so RepoTags is often empty. We must match by digest only.
-        """
+    def test_empty_repo_tags(self) -> None:
+        """Handle images where RepoTags is empty or None."""
         image = _create_mock_image(
-            repo_tags=None,  # Empty - typical when pulled by digest
+            repo_tags=None,
             repo_digests=["nginx@sha256:" + SHA],
         )
-        # Combined format SHOULD match because digest matches
+        # Combined format should not match if RepoTags is empty
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
-        assert len(result) == 1
-
-    def test_empty_repo_tags_list_matches_by_digest(self) -> None:
-        """Empty list RepoTags should still match by digest."""
-        image = _create_mock_image(
-            repo_tags=[],  # Empty list - typical when pulled by digest
-            repo_digests=["nginx@sha256:" + SHA],
-        )
-        result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
-        assert len(result) == 1
+        assert len(result) == 0
 
     def test_empty_repo_digests(self) -> None:
         """Handle images where RepoDigests is empty or None."""
@@ -632,24 +613,22 @@ class TestFilterImagesByTag:
             repo_tags=["nginx:1.21"],
             repo_digests=None,
         )
-        # Combined format should not match if RepoDigests is empty (digest is authoritative)
+        # Combined format should not match if RepoDigests is empty
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
         assert len(result) == 0
 
-    def test_multiple_tags_irrelevant_for_combined(self) -> None:
-        """For combined tag@digest, tags are irrelevant - digest is authoritative."""
+    def test_multiple_tags(self) -> None:
+        """Image with multiple tags should match on any of them."""
         image = _create_mock_image(
             repo_tags=["nginx:1.21", "nginx:latest", "nginx:stable"],
             repo_digests=["nginx@sha256:" + SHA],
         )
-        # All of these should match because the digest matches
+        # Should match on 1.21
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
         assert len(result) == 1
+        # Should also match on latest
         result = filter_images_by_tag("nginx", f"latest@sha256:{SHA}", [image])
         assert len(result) == 1
-        # Even a non-existent tag should match because digest matches
-        result = filter_images_by_tag("nginx", f"nonexistent@sha256:{SHA}", [image])
-        assert len(result) == 1
 
     def test_multiple_digests(self) -> None:
         """Image with multiple digests should match on any of them."""
@@ -658,15 +637,12 @@ class TestFilterImagesByTag:
             repo_tags=["nginx:1.21"],
             repo_digests=["nginx@sha256:" + SHA, "nginx@sha256:" + other_sha],
         )
-        # Should match on first digest (tag is irrelevant)
+        # Should match on first digest
         result = filter_images_by_tag("nginx", f"1.21@sha256:{SHA}", [image])
         assert len(result) == 1
         # Should also match on second digest
         result = filter_images_by_tag("nginx", f"1.21@sha256:{other_sha}", [image])
         assert len(result) == 1
-        # Should match even with different tag
-        result = filter_images_by_tag("nginx", f"different@sha256:{SHA}", [image])
-        assert len(result) == 1
 
     def test_port_in_registry_name(self) -> None:
         """Registry with port number should not be confused with tag."""

tests/unit/plugins/module_utils/test_pull_image.py

@@ -1,218 +0,0 @@
-# Copyright (c) Ansible Project
-# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
-# SPDX-License-Identifier: GPL-3.0-or-later
-
-"""
-Tests for build_pull_arguments handling of tag@digest format.
-
-When users specify an image with both tag and digest (e.g., nginx:1.21@sha256:abc...),
-the parse_repository_tag function returns ("nginx", "1.21@sha256:abc...").
-
-The Docker SDK and API don't accept "tag@digest" in the tag parameter.
-build_pull_arguments handles this by:
-- Returning full reference as name when tag contains @
-- Returning None for tag in that case
-
-This function is used by:
-- pull_image() in _common.py and _common_api.py (for pulling images)
-- push_image() in docker_image.py (for pushing images)
-"""
-
-from __future__ import annotations
-
-import unittest
-
-from ansible_collections.community.docker.plugins.module_utils._util import (
-    build_pull_arguments,
-)
-
-
-SHA = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
-
-
-class TestBuildPullArguments(unittest.TestCase):
-    """Test cases for build_pull_arguments function."""
-
-    # ========== Tag-only tests ==========
-
-    def test_tag_only_simple(self) -> None:
-        """Tag-only should return name and tag separately."""
-        assert build_pull_arguments("nginx", "1.21") == ("nginx", "1.21")
-
-    def test_tag_only_latest(self) -> None:
-        """Latest tag should return name and tag separately."""
-        assert build_pull_arguments("nginx", "latest") == ("nginx", "latest")
-
-    def test_tag_only_with_registry(self) -> None:
-        """Tag-only with registry should return name and tag separately."""
-        assert build_pull_arguments("ghcr.io/user/repo", "v1.0") == (
-            "ghcr.io/user/repo",
-            "v1.0",
-        )
-
-    def test_tag_only_with_registry_port(self) -> None:
-        """Tag-only with registry port should return name and tag separately."""
-        assert build_pull_arguments("localhost:5000/myapp", "v2.0") == (
-            "localhost:5000/myapp",
-            "v2.0",
-        )
-
-    # ========== Digest-only tests ==========
-
-    def test_digest_only(self) -> None:
-        """Digest-only (sha256:...) should return name and digest separately."""
-        assert build_pull_arguments("nginx", f"sha256:{SHA}") == (
-            "nginx",
-            f"sha256:{SHA}",
-        )
-
-    def test_digest_only_with_registry(self) -> None:
-        """Digest-only with registry should return name and digest separately."""
-        assert build_pull_arguments("ghcr.io/user/repo", f"sha256:{SHA}") == (
-            "ghcr.io/user/repo",
-            f"sha256:{SHA}",
-        )
-
-    # ========== Combined tag@digest tests ==========
-
-    def test_combined_tag_digest_simple(self) -> None:
-        """Combined tag@digest should return full reference with None tag."""
-        assert build_pull_arguments("nginx", f"1.21@sha256:{SHA}") == (
-            f"nginx:1.21@sha256:{SHA}",
-            None,
-        )
-
-    def test_combined_tag_digest_with_user_repo(self) -> None:
-        """Combined tag@digest with user/repo should return full reference."""
-        assert build_pull_arguments("portainer/portainer-ee", f"2.35.0-alpine@sha256:{SHA}") == (
-            f"portainer/portainer-ee:2.35.0-alpine@sha256:{SHA}",
-            None,
-        )
-
-    def test_combined_tag_digest_with_ghcr(self) -> None:
-        """Combined tag@digest with GHCR should return full reference."""
-        assert build_pull_arguments("ghcr.io/gethomepage/homepage", f"v1.7@sha256:{SHA}") == (
-            f"ghcr.io/gethomepage/homepage:v1.7@sha256:{SHA}",
-            None,
-        )
-
-    def test_combined_tag_digest_with_registry_port(self) -> None:
-        """Combined tag@digest with registry port should return full reference."""
-        assert build_pull_arguments("localhost:5000/myapp", f"v2.0@sha256:{SHA}") == (
-            f"localhost:5000/myapp:v2.0@sha256:{SHA}",
-            None,
-        )
-
-    # ========== Edge cases ==========
-
-    def test_empty_tag_part_before_digest(self) -> None:
-        """Handle edge case where tag part is empty like '@sha256:...'."""
-        # This is an unusual case but should still work
-        assert build_pull_arguments("nginx", f"@sha256:{SHA}") == (
-            f"nginx:@sha256:{SHA}",
-            None,
-        )
-
-    def test_tag_with_hyphen_and_digest(self) -> None:
-        """Tag with hyphens and digest should work."""
-        assert build_pull_arguments("nginx", f"1.21-alpine@sha256:{SHA}") == (
-            f"nginx:1.21-alpine@sha256:{SHA}",
-            None,
-        )
-
-    def test_tag_with_dots_and_digest(self) -> None:
-        """Tag with dots and digest should work."""
-        assert build_pull_arguments("nginx", f"1.21.0@sha256:{SHA}") == (
-            f"nginx:1.21.0@sha256:{SHA}",
-            None,
-        )
-
-
-class TestBuildPullArgumentsIntegration(unittest.TestCase):
-    """Integration tests with parse_repository_tag."""
-
-    def test_full_flow_docker_hub(self) -> None:
-        """Test parse_repository_tag -> build_pull_arguments for Docker Hub."""
-        from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
-            parse_repository_tag,
-        )
-
-        # User specifies: portainer/portainer-ee:2.35.0-alpine@sha256:abc...
-        image_ref = f"portainer/portainer-ee:2.35.0-alpine@sha256:{SHA}"
-        repo, tag = parse_repository_tag(image_ref)
-
-        # parse_repository_tag returns repo and combined tag@digest
-        assert repo == "portainer/portainer-ee"
-        assert tag == f"2.35.0-alpine@sha256:{SHA}"
-
-        # build_pull_arguments converts to full reference
-        pull_name, pull_tag = build_pull_arguments(repo, tag)
-        assert pull_name == f"portainer/portainer-ee:2.35.0-alpine@sha256:{SHA}"
-        assert pull_tag is None
-
-    def test_full_flow_ghcr(self) -> None:
-        """Test parse_repository_tag -> build_pull_arguments for GHCR."""
-        from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
-            parse_repository_tag,
-        )
-
-        # User specifies: ghcr.io/gethomepage/homepage:v1.7@sha256:abc...
-        image_ref = f"ghcr.io/gethomepage/homepage:v1.7@sha256:{SHA}"
-        repo, tag = parse_repository_tag(image_ref)
-
-        assert repo == "ghcr.io/gethomepage/homepage"
-        assert tag == f"v1.7@sha256:{SHA}"
-
-        pull_name, pull_tag = build_pull_arguments(repo, tag)
-        assert pull_name == f"ghcr.io/gethomepage/homepage:v1.7@sha256:{SHA}"
-        assert pull_tag is None
-
-    def test_full_flow_private_registry_with_port(self) -> None:
-        """Test full flow for private registry with port number."""
-        from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
-            parse_repository_tag,
-        )
-
-        # User specifies: localhost:5000/myapp:v2.0@sha256:abc...
-        image_ref = f"localhost:5000/myapp:v2.0@sha256:{SHA}"
-        repo, tag = parse_repository_tag(image_ref)
-
-        # Port should be part of repo, not confused with tag
-        assert repo == "localhost:5000/myapp"
-        assert tag == f"v2.0@sha256:{SHA}"
-
-        pull_name, pull_tag = build_pull_arguments(repo, tag)
-        assert pull_name == f"localhost:5000/myapp:v2.0@sha256:{SHA}"
-        assert pull_tag is None
-
-    def test_full_flow_tag_only(self) -> None:
-        """Test full flow for tag-only (no digest)."""
-        from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
-            parse_repository_tag,
-        )
-
-        image_ref = "nginx:1.21"
-        repo, tag = parse_repository_tag(image_ref)
-
-        assert repo == "nginx"
-        assert tag == "1.21"
-
-        pull_name, pull_tag = build_pull_arguments(repo, tag)
-        assert pull_name == "nginx"
-        assert pull_tag == "1.21"
-
-    def test_full_flow_digest_only(self) -> None:
-        """Test full flow for digest-only (no tag)."""
-        from ansible_collections.community.docker.plugins.module_utils._api.utils.utils import (
-            parse_repository_tag,
-        )
-
-        image_ref = f"nginx@sha256:{SHA}"
-        repo, tag = parse_repository_tag(image_ref)
-
-        assert repo == "nginx"
-        assert tag == f"sha256:{SHA}"
-
-        pull_name, pull_tag = build_pull_arguments(repo, tag)
-        assert pull_name == "nginx"
-        assert pull_tag == f"sha256:{SHA}"