CloudFront: Initial implementation of list_invalidations() (#5575)

moto/cloudfront/models.py

@@ -234,6 +234,7 @@ class CloudFrontBackend(BaseBackend):
     def __init__(self, region_name, account_id):
         super().__init__(region_name, account_id)
         self.distributions = dict()
+        self.invalidations = dict()
         self.tagger = TaggingService()
 
         state_manager.register_default_transition(
@@ -315,9 +316,19 @@ class CloudFrontBackend(BaseBackend):
     def create_invalidation(self, dist_id, paths, caller_ref):
         dist, _ = self.get_distribution(dist_id)
         invalidation = Invalidation(dist, paths, caller_ref)
+        try:
+            self.invalidations[dist_id].append(invalidation)
+        except KeyError:
+            self.invalidations[dist_id] = [invalidation]
 
         return invalidation
 
+    def list_invalidations(self, dist_id):
+        """
+        Pagination is not yet implemented
+        """
+        return self.invalidations.get(dist_id) or {}
+
     def list_tags_for_resource(self, resource):
         return self.tagger.list_tags_for_resource(resource)
 

moto/cloudfront/responses.py

@@ -30,6 +30,8 @@ class CloudFrontResponse(BaseResponse):
         self.setup_class(request, full_url, headers)
         if request.method == "POST":
             return self.create_invalidation()
+        if request.method == "GET":
+            return self.list_invalidations()
 
     def tags(self, request, full_url, headers):
         self.setup_class(request, full_url, headers)
@@ -104,6 +106,14 @@ class CloudFrontResponse(BaseResponse):
 
         return 200, {"Location": invalidation.location}, response
 
+    def list_invalidations(self):
+        dist_id = self.path.split("/")[-2]
+        invalidations = self.backend.list_invalidations(dist_id)
+        template = self.response_template(INVALIDATIONS_TEMPLATE)
+        response = template.render(invalidations=invalidations, xmlns=XMLNS)
+
+        return 200, {}, response
+
     def list_tags_for_resource(self):
         resource = unquote(self._get_param("Resource"))
         tags = self.backend.list_tags_for_resource(resource=resource)["Tags"]
@@ -590,6 +600,24 @@ CREATE_INVALIDATION_TEMPLATE = """<?xml version="1.0"?>
 </Invalidation>
 """
 
+INVALIDATIONS_TEMPLATE = """<?xml version="1.0" encoding="UTF-8"?>
+<InvalidationList>
+   <IsTruncated>false</IsTruncated>
+   <Items>
+      {% for invalidation in invalidations %}
+      <InvalidationSummary>
+         <CreateTime>{{ invalidation.create_time }}</CreateTime>
+         <Id>{{ invalidation.invalidation_id }}</Id>
+         <Status>{{ invalidation.status }}</Status>
+      </InvalidationSummary>
+      {% endfor %}
+   </Items>
+   <Marker></Marker>
+   <MaxItems>100</MaxItems>
+   <Quantity>{{ invalidations|length }}</Quantity>
+</InvalidationList>
+"""
+
 TAGS_TEMPLATE = """<?xml version="1.0"?>
 <Tags>
   <Items>

tests/test_cloudfront/test_cloudfront_invalidation.py

@@ -31,3 +31,49 @@ def test_create_invalidation():
             "CallerReference": "ref2",
         }
     )
+
+
+@mock_cloudfront
+def test_list_invalidations():
+    client = boto3.client("cloudfront", region_name="us-west-1")
+    config = scaffold.example_distribution_config("ref")
+    resp = client.create_distribution(DistributionConfig=config)
+    dist_id = resp["Distribution"]["Id"]
+    resp = client.create_invalidation(
+        DistributionId=dist_id,
+        InvalidationBatch={
+            "Paths": {"Quantity": 2, "Items": ["/path1", "/path2"]},
+            "CallerReference": "ref2",
+        },
+    )
+    invalidation_id = resp["Invalidation"]["Id"]
+
+    resp = client.list_invalidations(
+        DistributionId=dist_id,
+    )
+
+    resp.should.have.key("InvalidationList")
+    resp["InvalidationList"].shouldnt.have.key("NextMarker")
+    resp["InvalidationList"].should.have.key("MaxItems").equal(100)
+    resp["InvalidationList"].should.have.key("IsTruncated").equal(False)
+    resp["InvalidationList"].should.have.key("Quantity").equal(1)
+    resp["InvalidationList"].should.have.key("Items").length_of(1)
+    resp["InvalidationList"]["Items"][0].should.have.key("Id").equal(invalidation_id)
+    resp["InvalidationList"]["Items"][0].should.have.key("CreateTime")
+    resp["InvalidationList"]["Items"][0].should.have.key("Status").equal("COMPLETED")
+
+
+@mock_cloudfront
+def test_list_invalidations__no_entries():
+    client = boto3.client("cloudfront", region_name="us-west-1")
+
+    resp = client.list_invalidations(
+        DistributionId="SPAM",
+    )
+
+    resp.should.have.key("InvalidationList")
+    resp["InvalidationList"].shouldnt.have.key("NextMarker")
+    resp["InvalidationList"].should.have.key("MaxItems").equal(100)
+    resp["InvalidationList"].should.have.key("IsTruncated").equal(False)
+    resp["InvalidationList"].should.have.key("Quantity").equal(0)
+    resp["InvalidationList"].should.have.key("Items").equals([])