S3: Multiple Transitions in lifecycle configuration

moto/s3/models.py

@@ -637,6 +637,47 @@ def to_config_dict(self) -> List[Dict[str, Any]]:
         return data
 
 
+class LifecycleTransition(BaseModel):
+    def __init__(
+        self,
+        date: Optional[str] = None,
+        days: Optional[int] = None,
+        storage_class: Optional[str] = None,
+    ):
+        self.date = date
+        self.days = days
+        self.storage_class = storage_class
+
+    def to_config_dict(self) -> Dict[str, Any]:
+        config: Dict[str, Any] = {}
+        if self.date is not None:
+            config["date"] = self.date
+        if self.days is not None:
+            config["days"] = self.days
+        if self.storage_class is not None:
+            config["storageClass"] = self.storage_class
+        return config
+
+
+class LifeCycleNoncurrentVersionTransition(BaseModel):
+    def __init__(
+        self, days: int, storage_class: str, newer_versions: Optional[int] = None
+    ):
+        self.newer_versions = newer_versions
+        self.days = days
+        self.storage_class = storage_class
+
+    def to_config_dict(self) -> Dict[str, Any]:
+        config: Dict[str, Any] = {}
+        if self.newer_versions is not None:
+            config["newerNoncurrentVersions"] = self.newer_versions
+        if self.days is not None:
+            config["noncurrentDays"] = self.days
+        if self.storage_class is not None:
+            config["storageClass"] = self.storage_class
+        return config
+
+
 class LifecycleRule(BaseModel):
     def __init__(
         self,
@@ -646,13 +687,12 @@ def __init__(
         status: Optional[str] = None,
         expiration_days: Optional[str] = None,
         expiration_date: Optional[str] = None,
-        transition_days: Optional[str] = None,
-        transition_date: Optional[str] = None,
-        storage_class: Optional[str] = None,
+        transitions: Optional[List[LifecycleTransition]] = None,
         expired_object_delete_marker: Optional[str] = None,
         nve_noncurrent_days: Optional[str] = None,
-        nvt_noncurrent_days: Optional[str] = None,
-        nvt_storage_class: Optional[str] = None,
+        noncurrent_version_transitions: Optional[
+            List[LifeCycleNoncurrentVersionTransition]
+        ] = None,
         aimu_days: Optional[str] = None,
     ):
         self.id = rule_id
@@ -661,22 +701,15 @@ def __init__(
         self.status = status
         self.expiration_days = expiration_days
         self.expiration_date = expiration_date
-        self.transition_days = transition_days
-        self.transition_date = transition_date
-        self.storage_class = storage_class
+        self.transitions = transitions
         self.expired_object_delete_marker = expired_object_delete_marker
         self.nve_noncurrent_days = nve_noncurrent_days
-        self.nvt_noncurrent_days = nvt_noncurrent_days
-        self.nvt_storage_class = nvt_storage_class
+        self.noncurrent_version_transitions = noncurrent_version_transitions
         self.aimu_days = aimu_days
 
     def to_config_dict(self) -> Dict[str, Any]:
         """Converts the object to the AWS Config data dict.
 
-        Note: The following are missing that should be added in the future:
-            - transitions (returns None for now)
-            - noncurrentVersionTransitions (returns None for now)
-
         :param kwargs:
         :return:
         """
@@ -691,10 +724,22 @@ def to_config_dict(self) -> Dict[str, Any]:
             "expiredObjectDeleteMarker": self.expired_object_delete_marker,
             "noncurrentVersionExpirationInDays": -1 or int(self.nve_noncurrent_days),  # type: ignore
             "expirationDate": self.expiration_date,
-            "transitions": None,  # Replace me with logic to fill in
-            "noncurrentVersionTransitions": None,  # Replace me with logic to fill in
         }
 
+        if self.transitions:
+            lifecycle_dict["transitions"] = [
+                t.to_config_dict() for t in self.transitions
+            ]
+        else:
+            lifecycle_dict["transitions"] = None
+
+        if self.noncurrent_version_transitions:
+            lifecycle_dict["noncurrentVersionTransitions"] = [
+                t.to_config_dict() for t in self.noncurrent_version_transitions
+            ]
+        else:
+            lifecycle_dict["noncurrentVersionTransitions"] = None
+
         if self.aimu_days:
             lifecycle_dict["abortIncompleteMultipartUpload"] = {
                 "daysAfterInitiation": self.aimu_days
@@ -965,7 +1010,19 @@ def set_lifecycle(self, rules: List[Dict[str, Any]]) -> None:
         for rule in rules:
             # Extract and validate actions from Lifecycle rule
             expiration = rule.get("Expiration")
-            transition = rule.get("Transition")
+
+            transitions_input = rule.get("Transition", [])
+            if transitions_input and not isinstance(transitions_input, list):
+                transitions_input = [rule.get("Transition")]
+
+            transitions = [
+                LifecycleTransition(
+                    date=transition.get("Date"),
+                    days=transition.get("Days"),
+                    storage_class=transition.get("StorageClass"),
+                )
+                for transition in transitions_input
+            ]
 
             try:
                 top_level_prefix = (
@@ -982,17 +1039,21 @@ def set_lifecycle(self, rules: List[Dict[str, Any]]) -> None:
                     "NoncurrentDays"
                 ]
 
-            nvt_noncurrent_days = None
-            nvt_storage_class = None
-            if rule.get("NoncurrentVersionTransition") is not None:
-                if rule["NoncurrentVersionTransition"].get("NoncurrentDays") is None:
-                    raise MalformedXML()
-                if rule["NoncurrentVersionTransition"].get("StorageClass") is None:
+            nv_transitions_input = rule.get("NoncurrentVersionTransition", [])
+            if nv_transitions_input and not isinstance(nv_transitions_input, list):
+                nv_transitions_input = [rule.get("NoncurrentVersionTransition")]
+
+            noncurrent_version_transitions = []
+            for nvt in nv_transitions_input:
+                if nvt.get("NoncurrentDays") is None or nvt.get("StorageClass") is None:
                     raise MalformedXML()
-                nvt_noncurrent_days = rule["NoncurrentVersionTransition"][
-                    "NoncurrentDays"
-                ]
-                nvt_storage_class = rule["NoncurrentVersionTransition"]["StorageClass"]
+
+                transition = LifeCycleNoncurrentVersionTransition(
+                    newer_versions=nvt.get("NewerNoncurrentVersions"),
+                    days=nvt.get("NoncurrentDays"),
+                    storage_class=nvt.get("StorageClass"),
+                )
+                noncurrent_version_transitions.append(transition)
 
             aimu_days = None
             if rule.get("AbortIncompleteMultipartUpload") is not None:
@@ -1085,15 +1146,10 @@ def set_lifecycle(self, rules: List[Dict[str, Any]]) -> None:
                     status=rule["Status"],
                     expiration_days=expiration.get("Days") if expiration else None,
                     expiration_date=expiration.get("Date") if expiration else None,
-                    transition_days=transition.get("Days") if transition else None,
-                    transition_date=transition.get("Date") if transition else None,
-                    storage_class=transition.get("StorageClass")
-                    if transition
-                    else None,
+                    transitions=transitions,
                     expired_object_delete_marker=eodm,
                     nve_noncurrent_days=nve_noncurrent_days,
-                    nvt_noncurrent_days=nvt_noncurrent_days,
-                    nvt_storage_class=nvt_storage_class,
+                    noncurrent_version_transitions=noncurrent_version_transitions,
                     aimu_days=aimu_days,
                 )
             )
@@ -2317,7 +2373,6 @@ def delete_object(
 
                     for key in bucket.keys.getlist(key_name):
                         if str(key.version_id) == str(version_id):
-
                             if (
                                 hasattr(key, "is_locked")
                                 and key.is_locked

moto/s3/responses.py

@@ -2465,17 +2465,19 @@ def _invalid_headers(self, url: str, headers: Dict[str, str]) -> bool:
             {% endif %}
         {% endif %}
         <Status>{{ rule.status }}</Status>
-        {% if rule.storage_class %}
-        <Transition>
-            {% if rule.transition_days %}
-               <Days>{{ rule.transition_days }}</Days>
-            {% endif %}
-            {% if rule.transition_date %}
-               <Date>{{ rule.transition_date }}</Date>
-            {% endif %}
-           <StorageClass>{{ rule.storage_class }}</StorageClass>
-        </Transition>
-        {% endif %}
+        {% for transition in rule.transitions %}
+            <Transition>
+                {% if transition.days %}
+                <Days>{{ transition.days }}</Days>
+                {% endif %}
+                {% if transition.date %}
+                <Date>{{ transition.date }}</Date>
+                {% endif %}
+                {% if transition.storage_class %}
+                <StorageClass>{{ transition.storage_class }}</StorageClass>
+                {% endif %}
+            </Transition>
+        {% endfor %}
         {% if rule.expiration_days or rule.expiration_date or rule.expired_object_delete_marker %}
         <Expiration>
             {% if rule.expiration_days %}
@@ -2489,12 +2491,19 @@ def _invalid_headers(self, url: str, headers: Dict[str, str]) -> bool:
             {% endif %}
         </Expiration>
         {% endif %}
-        {% if rule.nvt_noncurrent_days and rule.nvt_storage_class %}
-        <NoncurrentVersionTransition>
-           <NoncurrentDays>{{ rule.nvt_noncurrent_days }}</NoncurrentDays>
-           <StorageClass>{{ rule.nvt_storage_class }}</StorageClass>
-        </NoncurrentVersionTransition>
-        {% endif %}
+        {% for nvt in rule.noncurrent_version_transitions %}
+            <NoncurrentVersionTransition>
+                {% if nvt.newer_versions %}
+                <NewerNoncurrentVersions>{{ nvt.newer_versions }}</NewerNoncurrentVersions>
+                {% endif %}
+                {% if nvt.days %}
+                <NoncurrentDays>{{ nvt.days }}</NoncurrentDays>
+                {% endif %}
+                {% if nvt.storage_class %}
+                <StorageClass>{{ nvt.storage_class }}</StorageClass>
+                {% endif %}
+            </NoncurrentVersionTransition>
+        {% endfor %}
         {% if rule.nve_noncurrent_days %}
         <NoncurrentVersionExpiration>
            <NoncurrentDays>{{ rule.nve_noncurrent_days }}</NoncurrentDays>

tests/test_s3/test_s3_config.py

@@ -12,7 +12,6 @@
 
 @mock_s3
 def test_s3_public_access_block_to_config_dict():
-
     # With 1 bucket in us-west-2:
     s3_config_query_backend.create_bucket("bucket1", "us-west-2")
 
@@ -48,7 +47,6 @@ def test_s3_public_access_block_to_config_dict():
 
 @mock_s3
 def test_list_config_discovered_resources():
-
     # Without any buckets:
     assert s3_config_query.list_config_service_resources(
         "global", "global", None, None, 100, None
@@ -173,6 +171,18 @@ def test_s3_lifecycle_config_dict():
             "Filter": {"Prefix": ""},
             "AbortIncompleteMultipartUpload": {"DaysAfterInitiation": 1},
         },
+        {
+            "ID": "rule5",
+            "Status": "Enabled",
+            "Filter": {"Prefix": ""},
+            "Transition": [{"Days": 10, "StorageClass": "GLACIER"}],
+            "NoncurrentVersionTransition": [
+                {
+                    "NoncurrentDays": 10,
+                    "StorageClass": "GLACIER",
+                }
+            ],
+        },
     ]
     s3_config_query_backend.put_bucket_lifecycle("bucket1", lifecycle)
 
@@ -253,6 +263,22 @@ def test_s3_lifecycle_config_dict():
         "filter": {"predicate": {"type": "LifecyclePrefixPredicate", "prefix": ""}},
     }
 
+    assert lifecycles[4] == {
+        "id": "rule5",
+        "prefix": None,
+        "status": "Enabled",
+        "expirationInDays": None,
+        "expiredObjectDeleteMarker": None,
+        "noncurrentVersionExpirationInDays": -1,
+        "expirationDate": None,
+        "abortIncompleteMultipartUpload": None,
+        "filter": {"predicate": {"type": "LifecyclePrefixPredicate", "prefix": ""}},
+        "transitions": [{"days": 10, "storageClass": "GLACIER"}],
+        "noncurrentVersionTransitions": [
+            {"noncurrentDays": 10, "storageClass": "GLACIER"}
+        ],
+    }
+
 
 @mock_s3
 def test_s3_notification_config_dict():

tests/test_s3/test_s3_lifecycle.py

@@ -376,6 +376,76 @@ def test_lifecycle_with_nvt():
     assert err.value.response["Error"]["Code"] == "MalformedXML"
 
 
+@mock_s3
+def test_lifecycle_with_multiple_nvt():
+    client = boto3.client("s3")
+    client.create_bucket(
+        Bucket="bucket", CreateBucketConfiguration={"LocationConstraint": "us-west-1"}
+    )
+
+    lfc = {
+        "Rules": [
+            {
+                "NoncurrentVersionTransitions": [
+                    {"NoncurrentDays": 30, "StorageClass": "ONEZONE_IA"},
+                    {"NoncurrentDays": 50, "StorageClass": "GLACIER"},
+                ],
+                "ID": "wholebucket",
+                "Filter": {"Prefix": ""},
+                "Status": "Enabled",
+            }
+        ]
+    }
+    client.put_bucket_lifecycle_configuration(
+        Bucket="bucket", LifecycleConfiguration=lfc
+    )
+    result = client.get_bucket_lifecycle_configuration(Bucket="bucket")
+    assert len(result["Rules"]) == 1
+    assert result["Rules"][0]["NoncurrentVersionTransitions"][0] == {
+        "NoncurrentDays": 30,
+        "StorageClass": "ONEZONE_IA",
+    }
+    assert result["Rules"][0]["NoncurrentVersionTransitions"][1] == {
+        "NoncurrentDays": 50,
+        "StorageClass": "GLACIER",
+    }
+
+
+@mock_s3
+def test_lifecycle_with_multiple_transitions():
+    client = boto3.client("s3")
+    client.create_bucket(
+        Bucket="bucket", CreateBucketConfiguration={"LocationConstraint": "us-west-1"}
+    )
+
+    lfc = {
+        "Rules": [
+            {
+                "Transitions": [
+                    {"Days": 30, "StorageClass": "ONEZONE_IA"},
+                    {"Days": 50, "StorageClass": "GLACIER"},
+                ],
+                "ID": "wholebucket",
+                "Filter": {"Prefix": ""},
+                "Status": "Enabled",
+            }
+        ]
+    }
+    client.put_bucket_lifecycle_configuration(
+        Bucket="bucket", LifecycleConfiguration=lfc
+    )
+    result = client.get_bucket_lifecycle_configuration(Bucket="bucket")
+    assert len(result["Rules"]) == 1
+    assert result["Rules"][0]["Transitions"][0] == {
+        "Days": 30,
+        "StorageClass": "ONEZONE_IA",
+    }
+    assert result["Rules"][0]["Transitions"][1] == {
+        "Days": 50,
+        "StorageClass": "GLACIER",
+    }
+
+
 @mock_s3
 def test_lifecycle_with_aimu():
     client = boto3.client("s3")