diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md index a91e89a06..42bb497a3 100644 --- a/IMPLEMENTATION_COVERAGE.md +++ b/IMPLEMENTATION_COVERAGE.md @@ -6744,7 +6744,7 @@ - [ ] delete_resource_policy - [ ] deregister_managed_instance - [ ] deregister_patch_baseline_for_patch_group -- [ ] deregister_target_from_maintenance_window +- [X] deregister_target_from_maintenance_window - [ ] deregister_task_from_maintenance_window - [ ] describe_activations - [ ] describe_association @@ -6767,7 +6767,7 @@ - [ ] describe_maintenance_window_execution_tasks - [ ] describe_maintenance_window_executions - [ ] describe_maintenance_window_schedule -- [ ] describe_maintenance_window_targets +- [X] describe_maintenance_window_targets - [ ] describe_maintenance_window_tasks - [X] describe_maintenance_windows - [ ] describe_maintenance_windows_for_target @@ -6828,7 +6828,7 @@ - [ ] put_resource_policy - [ ] register_default_patch_baseline - [ ] register_patch_baseline_for_patch_group -- [ ] register_target_with_maintenance_window +- [X] register_target_with_maintenance_window - [ ] register_task_with_maintenance_window - [X] remove_tags_from_resource - [ ] reset_service_setting diff --git a/moto/ssm/models.py b/moto/ssm/models.py index 1a31475a6..36865ec1a 100644 --- a/moto/ssm/models.py +++ b/moto/ssm/models.py @@ -941,6 +941,48 @@ def _valid_parameter_data_type(data_type: str) -> bool: return data_type in ("text", "aws:ec2:image") +class FakeMaintenanceWindowTarget: + def __init__( + self, + window_id: str, + resource_type: str, + targets: List[Dict[str, Any]], + owner_information: Optional[str], + name: Optional[str], + description: Optional[str], + ): + self.window_id = window_id + self.window_target_id = self.generate_id() + self.resource_type = resource_type + self.targets = targets + self.name = name + self.description = description + self.owner_information = owner_information + + def to_json(self) -> Dict[str, Any]: + return { + "WindowId": self.window_id, + "WindowTargetId": self.window_target_id, + "ResourceType": self.resource_type, + "Targets": self.targets, + "OwnerInformation": "", + "Name": self.name, + "Description": self.description, + } + + @staticmethod + def generate_id() -> str: + return str(random.uuid4()) + + +def _maintenance_window_target_filter_match( + filters: Optional[List[Dict[str, Any]]], target: FakeMaintenanceWindowTarget +) -> bool: + if not filters and target: + return True + return False + + class FakeMaintenanceWindow: def __init__( self, @@ -964,6 +1006,7 @@ class FakeMaintenanceWindow: self.schedule_offset = schedule_offset self.start_date = start_date self.end_date = end_date + self.targets: Dict[str, FakeMaintenanceWindowTarget] = {} def to_json(self) -> Dict[str, Any]: return { @@ -2254,5 +2297,53 @@ class SimpleSystemManagerBackend(BaseBackend): """ del self.baselines[baseline_id] + def register_target_with_maintenance_window( + self, + window_id: str, + resource_type: str, + targets: List[Dict[str, Any]], + owner_information: Optional[str], + name: Optional[str], + description: Optional[str], + ) -> str: + """ + Registers a target with a maintenance window. No error handling has been implemented yet. + """ + window = self.get_maintenance_window(window_id) + + target = FakeMaintenanceWindowTarget( + window_id, + resource_type, + targets, + owner_information=owner_information, + name=name, + description=description, + ) + window.targets[target.window_target_id] = target + return target.window_target_id + + def deregister_target_from_maintenance_window( + self, window_id: str, window_target_id: str + ) -> None: + """ + Deregisters a target from a maintenance window. No error handling has been implemented yet. + """ + window = self.get_maintenance_window(window_id) + del window.targets[window_target_id] + + def describe_maintenance_window_targets( + self, window_id: str, filters: Optional[List[Dict[str, Any]]] + ) -> List[FakeMaintenanceWindowTarget]: + """ + Describes all targets for a maintenance window. No error handling has been implemented yet. + """ + window = self.get_maintenance_window(window_id) + targets = [ + target + for target in window.targets.values() + if _maintenance_window_target_filter_match(filters, target) + ] + return targets + ssm_backends = BackendDict(SimpleSystemManagerBackend, "ssm") diff --git a/moto/ssm/responses.py b/moto/ssm/responses.py index d781b260d..3f757f03d 100644 --- a/moto/ssm/responses.py +++ b/moto/ssm/responses.py @@ -437,6 +437,36 @@ class SimpleSystemManagerResponse(BaseResponse): window = self.ssm_backend.get_maintenance_window(window_id) return json.dumps(window.to_json()) + def register_target_with_maintenance_window(self) -> str: + window_target_id = self.ssm_backend.register_target_with_maintenance_window( + window_id=self._get_param("WindowId"), + resource_type=self._get_param("ResourceType"), + targets=self._get_param("Targets"), + owner_information=self._get_param("OwnerInformation"), + name=self._get_param("Name"), + description=self._get_param("Description"), + ) + return json.dumps({"WindowTargetId": window_target_id}) + + def describe_maintenance_window_targets(self) -> str: + window_id = self._get_param("WindowId") + filters = self._get_param("Filters", []) + targets = [ + target.to_json() + for target in self.ssm_backend.describe_maintenance_window_targets( + window_id, filters + ) + ] + return json.dumps({"Targets": targets}) + + def deregister_target_from_maintenance_window(self) -> str: + window_id = self._get_param("WindowId") + window_target_id = self._get_param("WindowTargetId") + self.ssm_backend.deregister_target_from_maintenance_window( + window_id, window_target_id + ) + return "{}" + def describe_maintenance_windows(self) -> str: filters = self._get_param("Filters", None) windows = [ diff --git a/tests/test_ssm/test_ssm_maintenance_windows.py b/tests/test_ssm/test_ssm_maintenance_windows.py index 08b3c1ddc..f484a4252 100644 --- a/tests/test_ssm/test_ssm_maintenance_windows.py +++ b/tests/test_ssm/test_ssm_maintenance_windows.py @@ -212,3 +212,65 @@ def test_tags(): ResourceType="MaintenanceWindow", ResourceId=mw_id )["TagList"] assert tags == [{"Key": "k2", "Value": "v2"}] + + +@mock_ssm +def test_register_maintenance_window_target(): + ssm = boto3.client("ssm", region_name="us-east-1") + + resp = ssm.create_maintenance_window( + Name="simple-window", + Schedule="cron(15 12 * * ? *)", + Duration=2, + Cutoff=1, + AllowUnassociatedTargets=False, + ) + window_id = resp["WindowId"] + + resp = ssm.register_target_with_maintenance_window( + WindowId=window_id, + ResourceType="INSTANCE", + Targets=[{"Key": "tag:Name", "Values": ["my-instance"]}], + ) + resp.should.have.key("WindowTargetId") + _id = resp["WindowTargetId"] + + resp = ssm.describe_maintenance_window_targets( + WindowId=window_id, + ) + resp.should.have.key("Targets").should.have.length_of(1) + resp["Targets"][0].should.have.key("ResourceType").equal("INSTANCE") + resp["Targets"][0].should.have.key("WindowTargetId").equal(_id) + resp["Targets"][0]["Targets"][0].should.have.key("Key").equal("tag:Name") + resp["Targets"][0]["Targets"][0].should.have.key("Values").equal(["my-instance"]) + + +@mock_ssm +def test_deregister_target_from_maintenance_window(): + ssm = boto3.client("ssm", region_name="us-east-1") + + resp = ssm.create_maintenance_window( + Name="simple-window", + Schedule="cron(15 12 * * ? *)", + Duration=2, + Cutoff=1, + AllowUnassociatedTargets=False, + ) + window_id = resp["WindowId"] + + resp = ssm.register_target_with_maintenance_window( + WindowId=window_id, + ResourceType="INSTANCE", + Targets=[{"Key": "tag:Name", "Values": ["my-instance"]}], + ) + _id = resp["WindowTargetId"] + + ssm.deregister_target_from_maintenance_window( + WindowId=window_id, + WindowTargetId=_id, + ) + + resp = ssm.describe_maintenance_window_targets( + WindowId=window_id, + ) + resp.should.have.key("Targets").should.have.length_of(0)