Add ASG Scheduled Scaling Action batch APIs (#7323)

moto/autoscaling/models.py

@@ -321,11 +321,11 @@ class FakeScheduledAction(CloudFormationModel):
         max_size: Optional[int],
         min_size: Optional[int],
         scheduled_action_name: str,
-        start_time: str,
+        start_time: Optional[str],
-        end_time: str,
+        end_time: Optional[str],
-        recurrence: str,
+        recurrence: Optional[str],
+        timezone: Optional[str],
     ):
-
         self.name = name
         self.desired_capacity = desired_capacity
         self.max_size = max_size
@@ -334,6 +334,7 @@ class FakeScheduledAction(CloudFormationModel):
         self.end_time = end_time
         self.recurrence = recurrence
         self.scheduled_action_name = scheduled_action_name
+        self.timezone = timezone
 
     @staticmethod
     def cloudformation_name_type() -> str:
@@ -353,7 +354,6 @@ class FakeScheduledAction(CloudFormationModel):
         region_name: str,
         **kwargs: Any,
     ) -> "FakeScheduledAction":
-
         properties = cloudformation_json["Properties"]
 
         backend = autoscaling_backends[account_id][region_name]
@@ -373,10 +373,24 @@ class FakeScheduledAction(CloudFormationModel):
             start_time=properties.get("StartTime"),
             end_time=properties.get("EndTime"),
             recurrence=properties.get("Recurrence"),
+            timezone=properties.get("TimeZone"),
         )
         return scheduled_action
 
 
+class FailedScheduledUpdateGroupActionRequest:
+    def __init__(
+        self,
+        *,
+        scheduled_action_name: str,
+        error_code: Optional[str] = None,
+        error_message: Optional[str] = None,
+    ) -> None:
+        self.scheduled_action_name = scheduled_action_name
+        self.error_code = error_code
+        self.error_message = error_message
+
+
 def set_string_propagate_at_launch_booleans_on_tags(
     tags: List[Dict[str, Any]]
 ) -> List[Dict[str, Any]]:
@@ -953,9 +967,10 @@ class AutoScalingBackend(BaseBackend):
         max_size: Union[None, str, int],
         min_size: Union[None, str, int],
         scheduled_action_name: str,
-        start_time: str,
+        start_time: Optional[str],
-        end_time: str,
+        end_time: Optional[str],
-        recurrence: str,
+        recurrence: Optional[str],
+        timezone: Optional[str],
     ) -> FakeScheduledAction:
         max_size = make_int(max_size)
         min_size = make_int(min_size)
@@ -970,11 +985,39 @@ class AutoScalingBackend(BaseBackend):
             start_time=start_time,
             end_time=end_time,
             recurrence=recurrence,
+            timezone=timezone,
         )
 
         self.scheduled_actions[scheduled_action_name] = scheduled_action
         return scheduled_action
 
+    def batch_put_scheduled_update_group_action(
+        self, name: str, actions: List[Dict[str, Any]]
+    ) -> List[FailedScheduledUpdateGroupActionRequest]:
+        result = []
+        for action in actions:
+            try:
+                self.put_scheduled_update_group_action(
+                    name=name,
+                    desired_capacity=action.get("DesiredCapacity"),
+                    max_size=action.get("MaxSize"),
+                    min_size=action.get("MinSize"),
+                    scheduled_action_name=action["ScheduledActionName"],
+                    start_time=action.get("StartTime"),
+                    end_time=action.get("EndTime"),
+                    recurrence=action.get("Recurrence"),
+                    timezone=action.get("TimeZone"),
+                )
+            except AutoscalingClientError as err:
+                result.append(
+                    FailedScheduledUpdateGroupActionRequest(
+                        scheduled_action_name=action["ScheduledActionName"],
+                        error_code=err.error_type,
+                        error_message=err.message,
+                    )
+                )
+        return result
+
     def describe_scheduled_actions(
         self,
         autoscaling_group_name: Optional[str] = None,
@@ -1002,6 +1045,27 @@ class AutoScalingBackend(BaseBackend):
         )
         if scheduled_action:
             self.scheduled_actions.pop(scheduled_action_name, None)
+        else:
+            raise ValidationError("Scheduled action name not found")
+
+    def batch_delete_scheduled_action(
+        self, auto_scaling_group_name: str, scheduled_action_names: List[str]
+    ) -> List[FailedScheduledUpdateGroupActionRequest]:
+        result = []
+        for scheduled_action_name in scheduled_action_names:
+            try:
+                self.delete_scheduled_action(
+                    auto_scaling_group_name, scheduled_action_name
+                )
+            except AutoscalingClientError as err:
+                result.append(
+                    FailedScheduledUpdateGroupActionRequest(
+                        scheduled_action_name=scheduled_action_name,
+                        error_code=err.error_type,
+                        error_message=err.message,
+                    )
+                )
+        return result
 
     def create_auto_scaling_group(
         self,

moto/autoscaling/responses.py

@@ -118,10 +118,23 @@ class AutoScalingResponse(BaseResponse):
             start_time=self._get_param("StartTime"),
             end_time=self._get_param("EndTime"),
             recurrence=self._get_param("Recurrence"),
+            timezone=self._get_param("TimeZone"),
         )
         template = self.response_template(PUT_SCHEDULED_UPDATE_GROUP_ACTION_TEMPLATE)
         return template.render()
 
+    def batch_put_scheduled_update_group_action(self) -> str:
+        failed_actions = (
+            self.autoscaling_backend.batch_put_scheduled_update_group_action(
+                name=self._get_param("AutoScalingGroupName"),
+                actions=self._get_multi_param("ScheduledUpdateGroupActions.member"),
+            )
+        )
+        template = self.response_template(
+            BATCH_PUT_SCHEDULED_UPDATE_GROUP_ACTION_TEMPLATE
+        )
+        return template.render(failed_actions=failed_actions)
+
     def describe_scheduled_actions(self) -> str:
         scheduled_actions = self.autoscaling_backend.describe_scheduled_actions(
             autoscaling_group_name=self._get_param("AutoScalingGroupName"),
@@ -140,6 +153,16 @@ class AutoScalingResponse(BaseResponse):
         template = self.response_template(DELETE_SCHEDULED_ACTION_TEMPLATE)
         return template.render()
 
+    def batch_delete_scheduled_action(self) -> str:
+        auto_scaling_group_name = self._get_param("AutoScalingGroupName")
+        scheduled_action_names = self._get_multi_param("ScheduledActionNames.member")
+        failed_actions = self.autoscaling_backend.batch_delete_scheduled_action(
+            auto_scaling_group_name=auto_scaling_group_name,
+            scheduled_action_names=scheduled_action_names,
+        )
+        template = self.response_template(BATCH_DELETE_SCHEDULED_ACTION_TEMPLATE)
+        return template.render(failed_actions=failed_actions)
+
     def describe_scaling_activities(self) -> str:
         template = self.response_template(DESCRIBE_SCALING_ACTIVITIES_TEMPLATE)
         return template.render()
@@ -631,13 +654,35 @@ PUT_SCHEDULED_UPDATE_GROUP_ACTION_TEMPLATE = """<PutScheduledUpdateGroupActionRe
 </ResponseMetadata>
 </PutScheduledUpdateGroupActionResponse>"""
 
+
+BATCH_PUT_SCHEDULED_UPDATE_GROUP_ACTION_TEMPLATE = """<BatchPutScheduledUpdateGroupActionResponse xmlns="http://autoscaling.amazonaws.com/doc/2011-01-01/">
+  <BatchPutScheduledUpdateGroupActionResult>
+    <FailedScheduledUpdateGroupActions>
+      {% for failed_action in failed_actions %}
+      <member>
+        <ScheduledActionName>{{ failed_action.scheduled_action_name }}</ScheduledActionName>
+        {% if failed_action.error_code %}
+        <ErrorCode>{{ failed_action.error_code }}</ErrorCode>
+        {% endif %}
+        {% if failed_action.error_message %}
+        <ErrorMessage>{{ failed_action.error_message }}</ErrorMessage>
+        {% endif %}
+      </member>
+      {% endfor %}
+    </FailedScheduledUpdateGroupActions>
+  </BatchPutScheduledUpdateGroupActionResult>
+  <ResponseMetadata>
+    <RequestId></RequestId>
+  </ResponseMetadata>
+</BatchPutScheduledUpdateGroupActionResponse>"""
+
 DESCRIBE_SCHEDULED_ACTIONS = """<DescribeScheduledActionsResponse xmlns="http://autoscaling.amazonaws.com/doc/2011-01-01/">
   <DescribeScheduledActionsResult>
     <ScheduledUpdateGroupActions>
       {% for scheduled_action in scheduled_actions %}
       <member>
         <AutoScalingGroupName>{{ scheduled_action.name }}</AutoScalingGroupName>
-        <ScheduledActionName> {{ scheduled_action.scheduled_action_name }}</ScheduledActionName>
+        <ScheduledActionName>{{ scheduled_action.scheduled_action_name }}</ScheduledActionName>
         {% if scheduled_action.start_time %}
         <StartTime>{{ scheduled_action.start_time }}</StartTime>
         {% endif %}
@@ -647,9 +692,18 @@ DESCRIBE_SCHEDULED_ACTIONS = """<DescribeScheduledActionsResponse xmlns="http://
         {% if scheduled_action.recurrence %}
         <Recurrence>{{ scheduled_action.recurrence }}</Recurrence>
         {% endif %}
+        {% if scheduled_action.min_size is not none %}
         <MinSize>{{ scheduled_action.min_size }}</MinSize>
+        {% endif %}
+        {% if scheduled_action.max_size is not none %}
         <MaxSize>{{ scheduled_action.max_size }}</MaxSize>
+        {% endif %}
+        {% if scheduled_action.desired_capacity is not none %}
         <DesiredCapacity>{{ scheduled_action.desired_capacity }}</DesiredCapacity>
+        {% endif %}
+        {% if scheduled_action.timezone %}
+        <TimeZone>{{ scheduled_action.timezone }}</TimeZone>
+        {% endif %}
       </member>
       {% endfor %}
     </ScheduledUpdateGroupActions>
@@ -663,6 +717,27 @@ DELETE_SCHEDULED_ACTION_TEMPLATE = """<DeleteScheduledActionResponse xmlns="http
   </ResponseMetadata>
 </DeleteScheduledActionResponse>"""
 
+BATCH_DELETE_SCHEDULED_ACTION_TEMPLATE = """<BatchDeleteScheduledActionResponse xmlns="http://autoscaling.amazonaws.com/doc/2011-01-01/">
+  <BatchDeleteScheduledActionResult>
+    <FailedScheduledActions>
+      {% for failed_action in failed_actions %}
+      <member>
+        <ScheduledActionName>{{ failed_action.scheduled_action_name }}</ScheduledActionName>
+        {% if failed_action.error_code %}
+        <ErrorCode>{{ failed_action.error_code }}</ErrorCode>
+        {% endif %}
+        {% if failed_action.error_message %}
+        <ErrorMessage>{{ failed_action.error_message }}</ErrorMessage>
+        {% endif %}
+      </member>
+      {% endfor %}
+    </FailedScheduledActions>
+  </BatchDeleteScheduledActionResult>
+  <ResponseMetadata>
+    <RequestId></RequestId>
+  </ResponseMetadata>
+</BatchDeleteScheduledActionResponse>"""
+
 ATTACH_LOAD_BALANCER_TARGET_GROUPS_TEMPLATE = """<AttachLoadBalancerTargetGroupsResponse xmlns="http://autoscaling.amazonaws.com/doc/2011-01-01/">
 <AttachLoadBalancerTargetGroupsResult>
 </AttachLoadBalancerTargetGroupsResult>

tests/test_autoscaling/test_autoscaling_cloudformation.py

@@ -305,6 +305,7 @@ def test_autoscaling_group_with_elb():
                     "MinSize": 5,
                     "Recurrence": "* * * * *",
                     "StartTime": "2022-07-01T00:00:00Z",
+                    "TimeZone": "Etc/UTC",
                 },
             },
             "my-launch-config": {

tests/test_autoscaling/test_autoscaling_scheduledactions.py

@@ -1,6 +1,9 @@
+from datetime import datetime
 from unittest import TestCase
 
 import boto3
+from botocore.exceptions import ClientError
+from pytest import raises
 
 from moto import mock_aws
 
@@ -72,6 +75,121 @@ class TestAutoScalingScheduledActions(TestCase):
         actions = response["ScheduledUpdateGroupActions"]
         assert len(actions) == 1
 
+    def test_delete_nonexistent_action(self) -> None:
+        with raises(ClientError) as exc_info:
+            self.client.delete_scheduled_action(
+                AutoScalingGroupName=self.asg_name,
+                ScheduledActionName="nonexistent",
+            )
+        assert exc_info.value.response["Error"]["Code"] == "ValidationError"
+        assert (
+            exc_info.value.response["Error"]["Message"]
+            == "Scheduled action name not found"
+        )
+
+    def test_put_actions_content(self) -> None:
+        # actions with different combinations of properties return the correct content
+        self.client.batch_put_scheduled_update_group_action(
+            AutoScalingGroupName=self.asg_name,
+            ScheduledUpdateGroupActions=[
+                {
+                    "ScheduledActionName": "desired-capacity-with-start-time",
+                    "DesiredCapacity": 1,
+                    "StartTime": "2024-01-18T09:00:00Z",
+                },
+                {
+                    "ScheduledActionName": "min-size-with-recurrence",
+                    "MinSize": 3,
+                    "Recurrence": "* * * * *",
+                },
+                {
+                    "ScheduledActionName": "complete",
+                    "MinSize": 1,
+                    "DesiredCapacity": 2,
+                    "MaxSize": 3,
+                    "Recurrence": "* * * * *",
+                    "StartTime": "2024-01-19T09:00:00Z",
+                    "EndTime": "2024-01-20T09:00:00Z",
+                    "TimeZone": "America/New_York",
+                },
+            ],
+        )
+        response = self.client.describe_scheduled_actions(
+            AutoScalingGroupName=self.asg_name
+        )
+        actions = response["ScheduledUpdateGroupActions"]
+        assert len(actions) == 3
+
+        desired_capacity_action = next(
+            filter(
+                lambda action: action["ScheduledActionName"]
+                == "desired-capacity-with-start-time",
+                actions,
+            )
+        )
+        assert desired_capacity_action["DesiredCapacity"] == 1
+        assert desired_capacity_action["StartTime"] == datetime.fromisoformat(
+            "2024-01-18T09:00:00+00:00"
+        )
+        assert "MinSize" not in desired_capacity_action
+        assert "MaxSize" not in desired_capacity_action
+        assert "TimeZone" not in desired_capacity_action
+        assert "Recurrence" not in desired_capacity_action
+        assert "EndTime" not in desired_capacity_action
+
+        min_size_action = next(
+            filter(
+                lambda action: action["ScheduledActionName"]
+                == "min-size-with-recurrence",
+                actions,
+            )
+        )
+        assert min_size_action["MinSize"] == 3
+        assert min_size_action["Recurrence"] == "* * * * *"
+        assert "DesiredCapacity" not in min_size_action
+        assert "MaxSize" not in min_size_action
+        assert "TimeZone" not in min_size_action
+        # StartTime may be present
+        assert "EndTime" not in min_size_action
+
+        complete_action = next(
+            filter(lambda action: action["ScheduledActionName"] == "complete", actions)
+        )
+        assert complete_action["MinSize"] == 1
+        assert complete_action["DesiredCapacity"] == 2
+        assert complete_action["MaxSize"] == 3
+        assert complete_action["Recurrence"] == "* * * * *"
+        assert complete_action["StartTime"] == datetime.fromisoformat(
+            "2024-01-19T09:00:00+00:00"
+        )
+        assert complete_action["EndTime"] == datetime.fromisoformat(
+            "2024-01-20T09:00:00+00:00"
+        )
+        assert complete_action["TimeZone"] == "America/New_York"
+
+    def test_put_replaces_action_with_same_name(self) -> None:
+        self.client.put_scheduled_update_group_action(
+            AutoScalingGroupName=self.asg_name,
+            ScheduledActionName="my-action",
+            Recurrence="* * * * *",
+            MinSize=3,
+        )
+        self.client.put_scheduled_update_group_action(
+            AutoScalingGroupName=self.asg_name,
+            ScheduledActionName="my-action",
+            Recurrence="* * * * *",
+            DesiredCapacity=2,
+        )
+
+        response = self.client.describe_scheduled_actions(
+            AutoScalingGroupName=self.asg_name
+        )
+        actions = response["ScheduledUpdateGroupActions"]
+        assert len(actions) == 1
+
+        assert actions[0]["DesiredCapacity"] == 2
+        assert "MinSize" not in actions[0]
+
     def _create_scheduled_action(self, name, idx, asg_name=None):
         self.client.put_scheduled_update_group_action(
             AutoScalingGroupName=asg_name or self.asg_name,
@@ -82,4 +200,74 @@ class TestAutoScalingScheduledActions(TestCase):
             MinSize=idx + 1,
             MaxSize=idx + 5,
             DesiredCapacity=idx + 3,
+            TimeZone="Etc/UTC",
         )
+
+    def test_batch_put_scheduled_group_action(self) -> None:
+        put_response = self.client.batch_put_scheduled_update_group_action(
+            AutoScalingGroupName=self.asg_name,
+            ScheduledUpdateGroupActions=[
+                {
+                    "ScheduledActionName": "StartAction",
+                    "Recurrence": "0 9 * * 1-5",
+                    "MinSize": 1,
+                    "MaxSize": 1,
+                    "DesiredCapacity": 1,
+                },
+                {
+                    "ScheduledActionName": "StopAction",
+                    "Recurrence": "0 17 * * 1-5",
+                    "MinSize": 0,
+                    "MaxSize": 0,
+                    "DesiredCapacity": 0,
+                },
+            ],
+        )
+        assert len(put_response["FailedScheduledUpdateGroupActions"]) == 0
+
+        response = self.client.describe_scheduled_actions(
+            AutoScalingGroupName=self.asg_name
+        )
+        actions = response["ScheduledUpdateGroupActions"]
+        assert len(actions) == 2
+
+    def test_batch_delete_scheduled_action(self) -> None:
+        action_names = [f"my-action-{i}" for i in range(10)]
+        for j, action_name in enumerate(action_names):
+            self._create_scheduled_action(action_name, j)
+        delete_response = self.client.batch_delete_scheduled_action(
+            AutoScalingGroupName=self.asg_name,
+            ScheduledActionNames=action_names[::2],  # delete even indices
+        )
+        assert len(delete_response["FailedScheduledActions"]) == 0
+
+        describe_response = self.client.describe_scheduled_actions(
+            AutoScalingGroupName=self.asg_name
+        )
+        actions = describe_response["ScheduledUpdateGroupActions"]
+        assert len(actions) == 5
+
+        delete_response = self.client.batch_delete_scheduled_action(
+            AutoScalingGroupName=self.asg_name,
+            ScheduledActionNames=action_names[1::2],  # delete odd indices
+        )
+        assert len(delete_response["FailedScheduledActions"]) == 0
+
+        describe_response = self.client.describe_scheduled_actions(
+            AutoScalingGroupName=self.asg_name
+        )
+        actions = describe_response["ScheduledUpdateGroupActions"]
+        assert len(actions) == 0
+
+    def test_batch_delete_nonexistent_action(self) -> None:
+        action_name = "nonexistent"
+        response = self.client.batch_delete_scheduled_action(
+            AutoScalingGroupName=self.asg_name,
+            ScheduledActionNames=[action_name],
+        )
+        assert len(response["FailedScheduledActions"]) == 1
+        assert response["FailedScheduledActions"][0] == {
+            "ScheduledActionName": action_name,
+            "ErrorCode": "ValidationError",
+            "ErrorMessage": "Scheduled action name not found",
+        }