ECR: Implement ImageAlreadyExistsException (#6342)

moto/ecr/exceptions.py

@@ -105,6 +105,26 @@ class ImageNotFoundException(JsonRESTError):
         )
 
 
+class ImageAlreadyExistsException(JsonRESTError):
+    code = 400
+
+    def __init__(
+        self,
+        repository_name: str,
+        registry_id: str,
+        digest: str,
+        image_tag: str,
+    ):
+        super().__init__(
+            error_type="ImageAlreadyExistsException",
+            message=(
+                f"Image with digest '{digest}' and tag '{image_tag}' already exists "
+                f"in the repository with name '{repository_name}' "
+                f"in registry with id '{registry_id}'"
+            ),
+        )
+
+
 class InvalidParameterException(JsonRESTError):
     code = 400
 

moto/ecr/models.py

@@ -15,6 +15,7 @@ from moto.ecr.exceptions import (
     RepositoryAlreadyExistsException,
     RepositoryNotEmptyException,
     InvalidParameterException,
+    ImageAlreadyExistsException,
     RepositoryPolicyNotFoundException,
     LifecyclePolicyNotFoundException,
     RegistryPolicyNotFoundException,
@@ -566,19 +567,28 @@ class ECRBackend(BaseBackend):
             else:
                 image_manifest_mediatype = parsed_image_manifest["mediaType"]
 
-        # Tags are unique, so delete any existing image with this tag first
+        existing_images_with_matching_manifest = list(
-        self.batch_delete_image(
-            repository_name=repository_name, image_ids=[{"imageTag": image_tag}]
-        )
-
-        existing_images = list(
             filter(
                 lambda x: x.response_object["imageManifest"] == image_manifest,
                 repository.images,
             )
         )
 
-        if not existing_images:
+        # if an image with a matching manifest exists and it is tagged,
+        # trying to put the same image with the same tag will result in an
+        # ImageAlreadyExistsException
+
+        try:
+            existing_images_with_matching_tag = list(
+                filter(
+                    lambda x: x.response_object["imageTag"] == image_tag,
+                    repository.images,
+                )
+            )
+        except KeyError:
+            existing_images_with_matching_tag = []
+
+        if not existing_images_with_matching_manifest:
             # this image is not in ECR yet
             image = Image(
                 self.account_id,
@@ -588,11 +598,27 @@ class ECRBackend(BaseBackend):
                 image_manifest_mediatype,
             )
             repository.images.append(image)
+            if existing_images_with_matching_tag:
+                # 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}]
+                )
             return image
+        else:
+            # this image is in ECR
+            image = existing_images_with_matching_manifest[0]
+            if image.image_tag == image_tag:
+                raise ImageAlreadyExistsException(
+                    registry_id=repository.registry_id,
+                    image_tag=image_tag,
+                    digest=image.get_image_digest(),
+                    repository_name=repository_name,
+                )
             else:
                 # update existing image
-            existing_images[0].update_tag(image_tag)
+                image.update_tag(image_tag)
-            return existing_images[0]
+                return image
 
     def batch_get_image(
         self,

tests/test_ecr/test_ecr_boto3.py

@@ -506,6 +506,7 @@ def test_put_image_with_multiple_tags():
 def test_put_multiple_images_with_same_tag():
     repo_name = "testrepo"
     image_tag = "my-tag"
+    manifest = json.dumps(_create_image_manifest())
 
     client = boto3.client("ecr", "us-east-1")
     client.create_repository(repositoryName=repo_name)
@@ -513,10 +514,12 @@ def test_put_multiple_images_with_same_tag():
     image_1 = client.put_image(
         repositoryName=repo_name,
         imageTag=image_tag,
-        imageManifest=json.dumps(_create_image_manifest()),
+        imageManifest=manifest,
     )["image"]["imageId"]["imageDigest"]
 
-    # We should overwrite the first image
+    # We should overwrite the first image because the first image
+    # only has one tag
+
     image_2 = client.put_image(
         repositoryName=repo_name,
         imageTag=image_tag,
@@ -530,11 +533,11 @@ def test_put_multiple_images_with_same_tag():
     images.should.have.length_of(1)
     images[0]["imageDigest"].should.equal(image_2)
 
-    # Image with different tags are allowed
+    # Same image with different tags is allowed
     image_3 = client.put_image(
         repositoryName=repo_name,
         imageTag="different-tag",
-        imageManifest=json.dumps(_create_image_manifest()),
+        imageManifest=manifest,
     )["image"]["imageId"]["imageDigest"]
 
     images = client.describe_images(repositoryName=repo_name)["imageDetails"]
@@ -542,6 +545,42 @@ def test_put_multiple_images_with_same_tag():
     set([img["imageDigest"] for img in images]).should.equal({image_2, image_3})
 
 
+@mock_ecr
+def test_put_same_image_with_same_tag():
+    repo_name = "testrepo"
+    image_tag = "my-tag"
+    manifest = json.dumps(_create_image_manifest())
+
+    client = boto3.client("ecr", "us-east-1")
+    client.create_repository(repositoryName=repo_name)
+
+    image_1 = client.put_image(
+        repositoryName=repo_name,
+        imageTag=image_tag,
+        imageManifest=manifest,
+    )["image"]["imageId"]["imageDigest"]
+
+    with pytest.raises(ClientError) as e:
+        client.put_image(
+            repositoryName=repo_name,
+            imageTag=image_tag,
+            imageManifest=manifest,
+        )["image"]["imageId"]["imageDigest"]
+
+    ex = e.value
+    ex.operation_name.should.equal("PutImage")
+    ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
+    ex.response["Error"]["Code"].should.contain("ImageAlreadyExistsException")
+    ex.response["Error"]["Message"].should.equal(
+        f"Image with digest '{image_1}' and tag '{image_tag}' already exists "
+        f"in the repository with name '{repo_name}' in registry with id '{ACCOUNT_ID}'"
+    )
+
+    images = client.describe_images(repositoryName=repo_name)["imageDetails"]
+
+    images.should.have.length_of(1)
+
+
 @mock_ecr
 def test_list_images():
     client = boto3.client("ecr", region_name="us-east-1")