ECR: put_image() - ensure that we can overwrite tags on image with multiple tags (#6408)

moto/ecr/models.py

@@ -581,7 +581,7 @@ class ECRBackend(BaseBackend):
         try:
             existing_images_with_matching_tag = list(
                 filter(
-                    lambda x: x.response_object["imageTag"] == image_tag,
+                    lambda x: image_tag in x.image_tags,
                     repository.images,
                 )
             )
@@ -616,6 +616,11 @@ class ECRBackend(BaseBackend):
                     repository_name=repository_name,
                 )
             else:
+                # Tags are unique, so delete any existing image with this tag first
+                # (or remove the tag if the image has more than one tag)
+                self.batch_delete_image(
+                    repository_name=repository_name, image_ids=[{"imageTag": image_tag}]
+                )
                 # update existing image
                 image.update_tag(image_tag)
                 return image

tests/test_ecr/test_ecr_boto3.py

@@ -581,6 +581,55 @@ def test_put_same_image_with_same_tag():
     images.should.have.length_of(1)
 
 
+@mock_ecr
+def test_multiple_tags__ensure_tags_exist_only_on_one_image():
+    repo_name = "testrepo"
+    tag_to_move = "mock-tag"
+    image_manifests = {
+        "image_001": json.dumps(_create_image_manifest()),
+        "image_002": json.dumps(_create_image_manifest()),
+    }
+
+    client = boto3.client("ecr", "us-east-1")
+    client.create_repository(repositoryName=repo_name)
+
+    # Create image with unique tag
+    for name, manifest in image_manifests.items():
+        client.put_image(
+            repositoryName=repo_name,
+            imageTag=name,
+            imageManifest=manifest,
+        )
+
+    # Tag first image with shared tag
+    client.put_image(
+        repositoryName=repo_name,
+        imageTag=tag_to_move,
+        imageManifest=image_manifests["image_001"],
+    )["image"]["imageId"]["imageDigest"]
+
+    # Image can be found
+    initial_image, *_ = client.batch_get_image(
+        repositoryName=repo_name,
+        imageIds=[{"imageTag": tag_to_move}],
+    )["images"]
+    assert initial_image["imageManifest"] == image_manifests["image_001"]
+
+    # Tag second image with shared tag
+    client.put_image(
+        repositoryName=repo_name,
+        imageTag=tag_to_move,
+        imageManifest=image_manifests["image_002"],
+    )["image"]["imageId"]["imageDigest"]
+
+    # We should find the second image now - the shared tag should be removed from the first image
+    new_image, *_ = client.batch_get_image(
+        repositoryName=repo_name,
+        imageIds=[{"imageTag": tag_to_move}],
+    )["images"]
+    assert new_image["imageManifest"] == image_manifests["image_002"]
+
+
 @mock_ecr
 def test_list_images():
     client = boto3.client("ecr", region_name="us-east-1")