SSM: Support tags for Maintenance Windows (#6350)

moto/s3/models.py

@@ -2524,7 +2524,7 @@ class S3BackendDict(BackendDict):
 
         # Maps bucket names to account IDs. This is used to locate the exact S3Backend
         # holding the bucket and to maintain the common bucket namespace.
-        self.bucket_accounts: dict[str, str] = {}
+        self.bucket_accounts: Dict[str, str] = {}
 
 
 s3_backends = S3BackendDict(

moto/ssm/models.py

@@ -1981,6 +1981,9 @@ class SimpleSystemManagerBackend(BaseBackend):
                 raise InvalidResourceId()
             else:
                 return
+        elif resource_type == "MaintenanceWindow":
+            if resource_id not in self.windows:
+                raise InvalidResourceId()
         elif resource_type not in (
             # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.remove_tags_from_resource
             "ManagedInstance",
@@ -2076,6 +2079,7 @@ class SimpleSystemManagerBackend(BaseBackend):
         schedule_offset: int,
         start_date: str,
         end_date: str,
+        tags: Optional[List[Dict[str, str]]],
     ) -> str:
         """
         Creates a maintenance window. No error handling or input validation has been implemented yet.
@@ -2092,6 +2096,11 @@ class SimpleSystemManagerBackend(BaseBackend):
             end_date,
         )
         self.windows[window.id] = window
+
+        if tags:
+            window_tags = {t["Key"]: t["Value"] for t in tags}
+            self.add_tags_to_resource("MaintenanceWindow", window.id, window_tags)
+
         return window.id
 
     def get_maintenance_window(self, window_id: str) -> FakeMaintenanceWindow:

moto/ssm/responses.py

@@ -421,6 +421,7 @@ class SimpleSystemManagerResponse(BaseResponse):
         schedule_offset = self._get_int_param("ScheduleOffset")
         start_date = self._get_param("StartDate")
         end_date = self._get_param("EndDate")
+        tags = self._get_param("Tags")
         window_id = self.ssm_backend.create_maintenance_window(
             name=name,
             description=desc,
@@ -431,6 +432,7 @@ class SimpleSystemManagerResponse(BaseResponse):
             schedule_offset=schedule_offset,
             start_date=start_date,
             end_date=end_date,
+            tags=tags,
         )
         return json.dumps({"WindowId": window_id})
 

tests/test_ssm/test_ssm_maintenance_windows.py

@@ -153,3 +153,62 @@ def test_delete_maintenance_windows():
 
     resp = ssm.describe_maintenance_windows()
     resp.should.have.key("WindowIdentities").equals([])
+
+
+@mock_ssm
+def test_tags():
+    ssm = boto3.client("ssm", region_name="us-east-1")
+
+    # create without & list
+    mw_id = ssm.create_maintenance_window(
+        Name="simple-window",
+        Schedule="cron(15 12 * * ? *)",
+        Duration=2,
+        Cutoff=1,
+        AllowUnassociatedTargets=False,
+    )["WindowId"]
+
+    # add & list
+    ssm.add_tags_to_resource(
+        ResourceType="MaintenanceWindow",
+        ResourceId=mw_id,
+        Tags=[{"Key": "k1", "Value": "v1"}],
+    )
+    tags = ssm.list_tags_for_resource(
+        ResourceType="MaintenanceWindow", ResourceId=mw_id
+    )["TagList"]
+    assert tags == [{"Key": "k1", "Value": "v1"}]
+
+    # create & list
+    mw_id = ssm.create_maintenance_window(
+        Name="simple-window",
+        Schedule="cron(15 12 * * ? *)",
+        Duration=2,
+        Cutoff=1,
+        AllowUnassociatedTargets=False,
+        Tags=[{"Key": "k2", "Value": "v2"}],
+    )["WindowId"]
+    tags = ssm.list_tags_for_resource(
+        ResourceType="MaintenanceWindow", ResourceId=mw_id
+    )["TagList"]
+    assert tags == [{"Key": "k2", "Value": "v2"}]
+
+    # add more & list
+    ssm.add_tags_to_resource(
+        ResourceType="MaintenanceWindow",
+        ResourceId=mw_id,
+        Tags=[{"Key": "k3", "Value": "v3"}],
+    )
+    tags = ssm.list_tags_for_resource(
+        ResourceType="MaintenanceWindow", ResourceId=mw_id
+    )["TagList"]
+    assert tags == [{"Key": "k2", "Value": "v2"}, {"Key": "k3", "Value": "v3"}]
+
+    # remove & list
+    ssm.remove_tags_from_resource(
+        ResourceType="MaintenanceWindow", ResourceId=mw_id, TagKeys=["k3"]
+    )
+    tags = ssm.list_tags_for_resource(
+        ResourceType="MaintenanceWindow", ResourceId=mw_id
+    )["TagList"]
+    assert tags == [{"Key": "k2", "Value": "v2"}]