Add ecr replication config (#4170)

moto/ecr/exceptions.py

@@ -125,3 +125,10 @@ class ScanNotFoundException(JsonRESTError):
                 f"in the registry with id '{registry_id}'"
             ),
         )
+
+
+class ValidationException(JsonRESTError):
+    code = 400
+
+    def __init__(self, message):
+        super().__init__(error_type=__class__.__name__, message=message)

moto/ecr/models.py

@@ -25,6 +25,7 @@ from moto.ecr.exceptions import (
     RegistryPolicyNotFoundException,
     LimitExceededException,
     ScanNotFoundException,
+    ValidationException,
 )
 from moto.ecr.policy_validation import EcrLifecyclePolicyValidator
 from moto.iam.exceptions import MalformedPolicyDocument
@@ -325,6 +326,7 @@ class ECRBackend(BaseBackend):
     def __init__(self, region_name):
         self.region_name = region_name
         self.registry_policy = None
+        self.replication_config = {"rules": []}
         self.repositories: Dict[str, Repository] = {}
         self.tagger = TaggingService(tagName="tags")
 
@@ -896,6 +898,32 @@ class ECRBackend(BaseBackend):
             },
         }
 
+    def put_replication_configuration(self, replication_config):
+        rules = replication_config["rules"]
+        if len(rules) > 1:
+            raise ValidationException("This feature is disabled")
+
+        if len(rules) == 1:
+            for dest in rules[0]["destinations"]:
+                if (
+                    dest["region"] == self.region_name
+                    and dest["registryId"] == DEFAULT_REGISTRY_ID
+                ):
+                    raise InvalidParameterException(
+                        "Invalid parameter at 'replicationConfiguration' failed to satisfy constraint: "
+                        "'Replication destination cannot be the same as the source registry'"
+                    )
+
+        self.replication_config = replication_config
+
+        return {"replicationConfiguration": replication_config}
+
+    def describe_registry(self):
+        return {
+            "registryId": DEFAULT_REGISTRY_ID,
+            "replicationConfiguration": self.replication_config,
+        }
+
 
 ecr_backends = {}
 for region, ec2_backend in ec2_backends.items():

moto/ecr/responses.py

@@ -301,3 +301,15 @@ class ECRResponse(BaseResponse):
                 image_id=image_id,
             )
         )
+
+    def put_replication_configuration(self):
+        replication_config = self._get_param("replicationConfiguration")
+
+        return json.dumps(
+            self.ecr_backend.put_replication_configuration(
+                replication_config=replication_config
+            )
+        )
+
+    def describe_registry(self):
+        return json.dumps(self.ecr_backend.describe_registry())

tests/terraform-tests.success.txt

@@ -46,6 +46,7 @@ TestAccAWSEc2TransitGatewayPeeringAttachment
 TestAccAWSEc2TransitGatewayPeeringAttachmentDataSource
 TestAccAWSEcrLifecyclePolicy
 TestAccAWSEcrRegistryPolicy
+TestAccAWSEcrReplicationConfiguration
 TestAccAWSEcrRepository
 TestAccAWSEcrRepositoryDataSource
 TestAccAWSEcrRepositoryPolicy

tests/test_ecr/test_ecr_boto3.py

@@ -2414,3 +2414,108 @@ def test_describe_image_scan_findings_error_scan_not_exists():
         f"in the repository with name '{repo_name}' "
         f"in the registry with id '{ACCOUNT_ID}'"
     )
+
+
+@mock_ecr
+def test_put_replication_configuration():
+    # given
+    client = boto3.client("ecr", region_name="eu-central-1")
+    config = {
+        "rules": [
+            {"destinations": [{"region": "eu-west-1", "registryId": ACCOUNT_ID},]},
+        ]
+    }
+
+    # when
+    response = client.put_replication_configuration(replicationConfiguration=config)
+
+    # then
+    response["replicationConfiguration"].should.equal(config)
+
+
+@mock_ecr
+def test_put_replication_configuration_error_feature_disabled():
+    # given
+    client = boto3.client("ecr", region_name="eu-central-1")
+    config = {
+        "rules": [
+            {
+                "destinations": [
+                    {"region": "eu-central-1", "registryId": "111111111111"},
+                ]
+            },
+            {
+                "destinations": [
+                    {"region": "eu-central-1", "registryId": "222222222222"},
+                ]
+            },
+        ]
+    }
+
+    # when
+    with pytest.raises(ClientError) as e:
+        client.put_replication_configuration(replicationConfiguration=config)
+
+    # then
+    ex = e.value
+    ex.operation_name.should.equal("PutReplicationConfiguration")
+    ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
+    ex.response["Error"]["Code"].should.contain("ValidationException")
+    ex.response["Error"]["Message"].should.equal("This feature is disabled")
+
+
+@mock_ecr
+def test_put_replication_configuration_error_same_source():
+    # given
+    region_name = "eu-central-1"
+    client = boto3.client("ecr", region_name=region_name)
+    config = {
+        "rules": [
+            {"destinations": [{"region": region_name, "registryId": ACCOUNT_ID}]},
+        ]
+    }
+
+    # when
+    with pytest.raises(ClientError) as e:
+        client.put_replication_configuration(replicationConfiguration=config)
+
+    # then
+    ex = e.value
+    ex.operation_name.should.equal("PutReplicationConfiguration")
+    ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
+    ex.response["Error"]["Code"].should.contain("InvalidParameterException")
+    ex.response["Error"]["Message"].should.equal(
+        "Invalid parameter at 'replicationConfiguration' failed to satisfy constraint: "
+        "'Replication destination cannot be the same as the source registry'"
+    )
+
+
+@mock_ecr
+def test_describe_registry():
+    # given
+    client = boto3.client("ecr", region_name="eu-central-1")
+
+    # when
+    response = client.describe_registry()
+
+    # then
+    response["registryId"].should.equal(ACCOUNT_ID)
+    response["replicationConfiguration"].should.equal({"rules": []})
+
+
+@mock_ecr
+def test_describe_registry_after_update():
+    # given
+    client = boto3.client("ecr", region_name="eu-central-1")
+    config = {
+        "rules": [
+            {"destinations": [{"region": "eu-west-1", "registryId": ACCOUNT_ID}]},
+        ]
+    }
+    client.put_replication_configuration(replicationConfiguration=config)
+
+    # when
+    response = client.describe_registry()
+
+    # then
+    response["replicationConfiguration"].should.equal(config)