Repository: getmoto/moto
Pull request #5839 (2023-01-12 14:35:12)

#5838: Support PreferredMaintenanceWindow for RDS DBInstance
Handle `PreferredMaintenanceWindow` parameters for creation and modification of RDS DB instances. Generate the relevant AWS exceptions as part of handling the same.


Patch:
diff --git a/docs/docs/services/sagemaker.rst b/docs/docs/services/sagemaker.rst
index a9e90a411..69b07f598 100644
--- a/docs/docs/services/sagemaker.rst
+++ b/docs/docs/services/sagemaker.rst
@@ -171,8 +171,8 @@ sagemaker
 - [ ] describe_notebook_instance
 - [X] describe_notebook_instance_lifecycle_config
 - [X] describe_pipeline
-- [ ] describe_pipeline_definition_for_execution
-- [ ] describe_pipeline_execution
+- [X] describe_pipeline_definition_for_execution
+- [X] describe_pipeline_execution
 - [X] describe_processing_job
 - [ ] describe_project
 - [ ] describe_space
@@ -247,8 +247,8 @@ sagemaker
 - [ ] list_notebook_instance_lifecycle_configs
 - [ ] list_notebook_instances
 - [ ] list_pipeline_execution_steps
-- [ ] list_pipeline_executions
-- [ ] list_pipeline_parameters_for_execution
+- [X] list_pipeline_executions
+- [X] list_pipeline_parameters_for_execution
 - [X] list_pipelines
 - [X] list_processing_jobs
 - [ ] list_projects
@@ -277,7 +277,7 @@ sagemaker
 - [ ] start_inference_experiment
 - [ ] start_monitoring_schedule
 - [X] start_notebook_instance
-- [ ] start_pipeline_execution
+- [X] start_pipeline_execution
 - [ ] stop_auto_ml_job
 - [ ] stop_compilation_job
 - [ ] stop_edge_deployment_stage
diff --git a/moto/rds/models.py b/moto/rds/models.py
index b55c97025..d394f933e 100644
--- a/moto/rds/models.py
+++ b/moto/rds/models.py
@@ -35,7 +35,13 @@ from .exceptions import (
     SubscriptionNotFoundError,
     SubscriptionAlreadyExistError,
 )
-from .utils import FilterDef, apply_filter, merge_filters, validate_filters
+from .utils import (
+    FilterDef,
+    apply_filter,
+    merge_filters,
+    validate_filters,
+    valid_preferred_maintenance_window,
+)
 
 
 class Cluster:
@@ -446,9 +452,15 @@ class Database(CloudFormationModel):
             self.db_subnet_group = None
         self.security_groups = kwargs.get("security_groups", [])
         self.vpc_security_group_ids = kwargs.get("vpc_security_group_ids", [])
-        self.preferred_maintenance_window = kwargs.get(
-            "preferred_maintenance_window", "wed:06:38-wed:07:08"
+        self.preferred_maintenance_window = kwargs.get("preferred_maintenance_window")
+        self.preferred_backup_window = kwargs.get("preferred_backup_window")
+        msg = valid_preferred_maintenance_window(
+            self.preferred_maintenance_window,
+            self.preferred_backup_window,
         )
+        if msg:
+            raise RDSClientError("InvalidParameterValue", msg)
+
         self.db_parameter_group_name = kwargs.get("db_parameter_group_name")
         if (
             self.db_parameter_group_name
@@ -458,9 +470,6 @@ class Database(CloudFormationModel):
         ):
             raise DBParameterGroupNotFoundError(self.db_parameter_group_name)
 
-        self.preferred_backup_window = kwargs.get(
-            "preferred_backup_window", "13:14-13:44"
-        )
         self.license_model = kwargs.get("license_model", "general-public-license")
         self.option_group_name = kwargs.get("option_group_name", None)
         self.option_group_supplied = self.option_group_name is not None
@@ -554,8 +563,8 @@ class Database(CloudFormationModel):
               <DBInstanceIdentifier>{{ database.db_instance_identifier }}</DBInstanceIdentifier>
               <DbiResourceId>{{ database.dbi_resource_id }}</DbiResourceId>
               <InstanceCreateTime>{{ database.instance_create_time }}</InstanceCreateTime>
-              <PreferredBackupWindow>03:50-04:20</PreferredBackupWindow>
-              <PreferredMaintenanceWindow>wed:06:38-wed:07:08</PreferredMaintenanceWindow>
+              <PreferredBackupWindow>{{ database.preferred_backup_window }}</PreferredBackupWindow>
+              <PreferredMaintenanceWindow>{{ database.preferred_maintenance_window }}</PreferredMaintenanceWindow>
               <ReadReplicaDBInstanceIdentifiers>
                 {% for replica_id in database.replicas %}
                     <ReadReplicaDBInstanceIdentifier>{{ replica_id }}</ReadReplicaDBInstanceIdentifier>
@@ -771,6 +780,12 @@ class Database(CloudFormationModel):
             "db_instance_class": properties.get("DBInstanceClass"),
             "db_instance_identifier": resource_name,
             "db_name": properties.get("DBName"),
+            "preferred_backup_window": properties.get(
+                "PreferredBackupWindow", "13:14-13:44"
+            ),
+            "preferred_maintenance_window": properties.get(
+                "PreferredMaintenanceWindow", "wed:06:38-wed:07:08"
+            ).lower(),
             "db_subnet_group_name": db_subnet_group_name,
             "engine": properties.get("Engine"),
             "engine_version": properties.get("EngineVersion"),
@@ -1448,6 +1463,13 @@ class RDSBackend(BaseBackend):
                 "db_instance_identifier"
             ] = db_kwargs.pop("new_db_instance_identifier")
             self.databases[db_instance_identifier] = database
+        preferred_backup_window = db_kwargs.get("preferred_backup_window")
+        preferred_maintenance_window = db_kwargs.get("preferred_maintenance_window")
+        msg = valid_preferred_maintenance_window(
+            preferred_maintenance_window, preferred_backup_window
+        )
+        if msg:
+            raise RDSClientError("InvalidParameterValue", msg)
         database.update(db_kwargs)
         return database
 
diff --git a/moto/rds/responses.py b/moto/rds/responses.py
index 814e9d828..61cf35441 100644
--- a/moto/rds/responses.py
+++ b/moto/rds/responses.py
@@ -44,8 +44,12 @@ class RDSResponse(BaseResponse):
             "multi_az": self._get_bool_param("MultiAZ"),
             "option_group_name": self._get_param("OptionGroupName"),
             "port": self._get_param("Port"),
-            # PreferredBackupWindow
-            # PreferredMaintenanceWindow
+            "preferred_backup_window": self._get_param(
+                "PreferredBackupWindow", "13:14-13:44"
+            ),
+            "preferred_maintenance_window": self._get_param(
+                "PreferredMaintenanceWindow", "wed:06:38-wed:07:08"
+            ).lower(),
             "publicly_accessible": self._get_param("PubliclyAccessible"),
             "account_id": self.current_account,
             "region": self.region,
diff --git a/moto/rds/utils.py b/moto/rds/utils.py
index d5c3dca72..667aad638 100644
--- a/moto/rds/utils.py
+++ b/moto/rds/utils.py
@@ -3,7 +3,10 @@ from collections import namedtuple
 from botocore.utils import merge_dicts
 
 from collections import OrderedDict
+import datetime
+import re
 
+SECONDS_IN_ONE_DAY = 24 * 60 * 60
 FilterDef = namedtuple(
     "FilterDef",
     [
@@ -133,3 +136,155 @@ def apply_filter(resources, filters, filter_defs):
         if matches_filter:
             resources_filtered[identifier] = obj
     return resources_filtered
+
+
+def get_start_date_end_date(base_date, window):
+    """Gets the start date and end date given DDD:HH24:MM-DDD:HH24:MM.
+
+    :param base_date:
+        type datetime
+    :param window:
+        DDD:HH24:MM-DDD:HH24:MM
+    :returns:
+        Start and End Date in datetime format
+    :rtype:
+        tuple
+    """
+    days = {"mon": 1, "tue": 2, "wed": 3, "thu": 4, "fri": 5, "sat": 6, "sun": 7}
+    start = datetime.datetime.strptime(
+        base_date + " " + window[4:9], "%d/%m/%y %H:%M"
+    ) + datetime.timedelta(days=days[window[0:3]])
+    end = datetime.datetime.strptime(
+        base_date + " " + window[14::], "%d/%m/%y %H:%M"
+    ) + datetime.timedelta(days=days[window[10:13]])
+    return start, end
+
+
+def get_start_date_end_date_from_time(base_date, window):
+    """Gets the start date and end date given HH24:MM-HH24:MM.
+
+    :param base_date:
+        type datetime
+    :param window:
+        HH24:MM-HH24:MM
+    :returns:
+        Start and End Date in datetime format
+        along with flag for spills over a day
+        This is useful when determine time overlaps
+    :rtype:
+        tuple
+    """
+    times = window.split("-")
+    spillover = False
+    start = datetime.datetime.strptime(base_date + " " + times[0], "%d/%m/%y %H:%M")
+    end = datetime.datetime.strptime(base_date + " " + times[1], "%d/%m/%y %H:%M")
+    if end < start:
+        end += datetime.timedelta(days=1)
+        spillover = True
+    return start, end, spillover
+
+
+def get_overlap_between_two_date_ranges(
+    start_time_1, end_time_1, start_time_2, end_time_2
+):
+    """Determines overlap between 2 date ranges.
+
+    :param start_time_1:
+        type datetime
+    :param start_time_2:
+        type datetime
+    :param end_time_1:
+        type datetime
+    :param end_time_2:
+        type datetime
+    :returns:
+        overlap in seconds
+    :rtype:
+        int
+    """
+    latest_start = max(start_time_1, start_time_2)
+    earliest_end = min(end_time_1, end_time_2)
+    delta = earliest_end - latest_start
+    overlap = (delta.days * SECONDS_IN_ONE_DAY) + delta.seconds
+    return overlap
+
+
+def valid_preferred_maintenance_window(maintenance_window, backup_window):
+    """Determines validity of preferred_maintenance_window
+
+    :param maintenance_windown:
+        type DDD:HH24:MM-DDD:HH24:MM
+    :param backup_window:
+        type HH24:MM-HH24:MM
+    :returns:
+        message
+    :rtype:
+        str
+    """
+    MINUTES_30 = 1800
+    HOURS_24 = 86400
+    base_date = datetime.datetime.now().strftime("%d/%m/%y")
+    try:
+        p = re.compile(
+            "([a-z]{3}):([0-9]{2}):([0-9]{2})-([a-z]{3}):([0-9]{2}):([0-9]{2})"
+        )
+        if len(maintenance_window) != 19 or re.search(p, maintenance_window) is None:
+            return f"Invalid maintenance window format: {maintenance_window}. Should be specified as a range ddd:hh24:mi-ddd:hh24:mi (24H Clock UTC). Example: Sun:23:45-Mon:00:15"
+        if backup_window:
+            (
+                backup_window_start,
+                backup_window_end,
+                backup_spill,
+            ) = get_start_date_end_date_from_time(base_date, backup_window)
+            (
+                maintenance_window_start,
+                maintenance_window_end,
+                maintenance_spill,
+            ) = get_start_date_end_date_from_time(
+                base_date, maintenance_window[4:10] + maintenance_window[14::]
+            )
+            if (
+                get_overlap_between_two_date_ranges(
+                    backup_window_start,
+                    backup_window_end,
+                    maintenance_window_start,
+                    maintenance_window_end,
+                )
+                >= 0
+            ):
+                return "The backup window and maintenance window must not overlap."
+
+            # Due to spill overs, adjust the windows
+            elif maintenance_spill:
+                backup_window_start += datetime.timedelta(days=1)
+                backup_window_end += datetime.timedelta(days=1)
+            elif backup_spill:
+                maintenance_window_start += datetime.timedelta(days=1)
+                maintenance_window_end += datetime.timedelta(days=1)
+
+            # If spills, rerun overlap test with adjusted windows
+            if maintenance_spill or backup_spill:
+                if (
+                    get_overlap_between_two_date_ranges(
+                        backup_window_start,
+                        backup_window_end,
+                        maintenance_window_start,
+                        maintenance_window_end,
+                    )
+                    >= 0
+                ):
+                    return "The backup window and maintenance window must not overlap."
+
+        maintenance_window_start, maintenance_window_end = get_start_date_end_date(
+            base_date, maintenance_window
+        )
+        delta = maintenance_window_end - maintenance_window_start
+        delta_seconds = delta.seconds + (delta.days * SECONDS_IN_ONE_DAY)
+        if delta_seconds >= MINUTES_30 and delta_seconds <= HOURS_24:
+            return
+        elif delta_seconds >= 0 and delta_seconds <= MINUTES_30:
+            return "The maintenance window must be at least 30 minutes."
+        else:
+            return "Maintenance window must be less than 24 hours."
+    except Exception:
+        return f"Invalid day:hour:minute value: {maintenance_window}"
diff --git a/moto/sagemaker/models.py b/moto/sagemaker/models.py
index 515091d46..2d5a2103b 100644
--- a/moto/sagemaker/models.py
+++ b/moto/sagemaker/models.py
@@ -1,5 +1,7 @@
 import json
 import os
+import random
+import string
 from datetime import datetime
 
 from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel
@@ -11,6 +13,11 @@ from .exceptions import (
     AWSValidationException,
     ResourceNotFound,
 )
+from .utils import (
+    get_pipeline_from_name,
+    get_pipeline_execution_from_arn,
+    get_pipeline_name_from_execution_arn,
+)
 from .utils import load_pipeline_definition_from_s3, arn_formatter
 
 
@@ -72,6 +79,50 @@ class BaseObject(BaseModel):
         return self.gen_response_object()
 
 
+class FakePipelineExecution(BaseObject):
+    def __init__(
+        self,
+        pipeline_execution_arn,
+        pipeline_execution_display_name,
+        pipeline_parameters,
+        pipeline_execution_description,
+        parallelism_configuration,
+        pipeline_definition,
+    ):
+        self.pipeline_execution_arn = pipeline_execution_arn
+        self.pipeline_execution_display_name = pipeline_execution_display_name
+        self.pipeline_parameters = pipeline_parameters
+        self.pipeline_execution_description = pipeline_execution_description
+        self.pipeline_execution_status = "Succeeded"
+        self.pipeline_execution_failure_reason = None
+        self.parallelism_configuration = parallelism_configuration
+        self.pipeline_definition_for_execution = pipeline_definition
+
+        now_string = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        self.creation_time = now_string
+        self.last_modified_time = now_string
+        self.start_time = now_string
+
+        fake_user_profile_name = "fake-user-profile-name"
+        fake_domain_id = "fake-domain-id"
+        fake_user_profile_arn = arn_formatter(
+            "user-profile",
+            f"{fake_domain_id}/{fake_user_profile_name}",
+            pipeline_execution_arn.split(":")[4],
+            pipeline_execution_arn.split(":")[3],
+        )
+        self.created_by = {
+            "UserProfileArn": fake_user_profile_arn,
+            "UserProfileName": fake_user_profile_name,
+            "DomainId": fake_domain_id,
+        }
+        self.last_modified_by = {
+            "UserProfileArn": fake_user_profile_arn,
+            "UserProfileName": fake_user_profile_name,
+            "DomainId": fake_domain_id,
+        }
+
+
 class FakePipeline(BaseObject):
     def __init__(
         self,
@@ -92,6 +143,7 @@ class FakePipeline(BaseObject):
         self.pipeline_display_name = pipeline_display_name or pipeline_name
         self.pipeline_definition = pipeline_definition
         self.pipeline_description = pipeline_description
+        self.pipeline_executions = dict()
         self.role_arn = role_arn
         self.tags = tags or []
         self.parallelism_configuration = parallelism_configuration
@@ -1088,6 +1140,7 @@ class SageMakerModelBackend(BaseBackend):
         self.endpoints = {}
         self.experiments = {}
         self.pipelines = {}
+        self.pipeline_executions = {}
         self.processing_jobs = {}
         self.trials = {}
         self.trial_components = {}
@@ -1815,27 +1868,16 @@ class SageMakerModelBackend(BaseBackend):
         self,
         pipeline_name,
     ):
-        try:
-            pipeline_arn = self.pipelines[pipeline_name].pipeline_arn
-        except KeyError:
-            raise ValidationError(
-                message=f"Could not find pipeline with name {pipeline_name}."
-            )
-        del self.pipelines[pipeline_name]
-        return pipeline_arn
+        pipeline = get_pipeline_from_name(self.pipelines, pipeline_name)
+        del self.pipelines[pipeline.pipeline_name]
+        return pipeline.pipeline_arn
 
     def update_pipeline(
         self,
         pipeline_name,
         **kwargs,
     ):
-        try:
-            pipeline_arn = self.pipelines[pipeline_name].pipeline_arn
-        except KeyError:
-            raise ValidationError(
-                message=f"Could not find pipeline with name {pipeline_name}."
-            )
-
+        pipeline = get_pipeline_from_name(self.pipelines, pipeline_name)
         if all(
             [
                 kwargs.get("pipeline_definition"),
@@ -1858,19 +1900,127 @@ class SageMakerModelBackend(BaseBackend):
                     continue
                 setattr(self.pipelines[pipeline_name], attr_key, attr_value)
 
-        return pipeline_arn
+        return pipeline.pipeline_arn
 
-    def describe_pipeline(
+    def start_pipeline_execution(
         self,
         pipeline_name,
+        pipeline_execution_display_name,
+        pipeline_parameters,
+        pipeline_execution_description,
+        parallelism_configuration,
     ):
-        try:
-            pipeline = self.pipelines[pipeline_name]
-        except KeyError:
-            raise ValidationError(
-                message=f"Could not find pipeline with name {pipeline_name}."
-            )
+        pipeline = get_pipeline_from_name(self.pipelines, pipeline_name)
+        execution_id = "".join(
+            random.choices(string.ascii_lowercase + string.digits, k=12)
+        )
+        pipeline_execution_arn = arn_formatter(
+            _type="pipeline",
+            _id=f"{pipeline.pipeline_name}/execution/{execution_id}",
+            account_id=self.account_id,
+            region_name=self.region_name,
+        )
+
+        fake_pipeline_execution = FakePipelineExecution(
+            pipeline_execution_arn=pipeline_execution_arn,
+            pipeline_execution_display_name=pipeline_execution_display_name,
+            pipeline_parameters=pipeline_parameters,
+            pipeline_execution_description=pipeline_execution_description,
+            pipeline_definition=pipeline.pipeline_definition,
+            parallelism_configuration=parallelism_configuration
+            or pipeline.parallelism_configuration,
+        )
+
+        self.pipelines[pipeline_name].pipeline_executions[
+            pipeline_execution_arn
+        ] = fake_pipeline_execution
+        self.pipelines[
+            pipeline_name
+        ].last_execution_time = fake_pipeline_execution.start_time
+
+        response = {"PipelineExecutionArn": pipeline_execution_arn}
+        return response
+
+    def list_pipeline_executions(
+        self,
+        pipeline_name,
+    ):
+        pipeline = get_pipeline_from_name(self.pipelines, pipeline_name)
+        response = {
+            "PipelineExecutionSummaries": [
+                {
+                    "PipelineExecutionArn": pipeline_execution_arn,
+                    "StartTime": pipeline_execution.start_time,
+                    "PipelineExecutionStatus": pipeline_execution.pipeline_execution_status,
+                    "PipelineExecutionDescription": pipeline_execution.pipeline_execution_description,
+                    "PipelineExecutionDisplayName": pipeline_execution.pipeline_execution_display_name,
+                    "PipelineExecutionFailureReason": str(
+                        pipeline_execution.pipeline_execution_failure_reason
+                    ),
+                }
+                for pipeline_execution_arn, pipeline_execution in pipeline.pipeline_executions.items()
+            ]
+        }
+        return response
+
+    def describe_pipeline_definition_for_execution(
+        self,
+        pipeline_execution_arn,
+    ):
+        pipeline_execution = get_pipeline_execution_from_arn(
+            self.pipelines, pipeline_execution_arn
+        )
+        response = {
+            "PipelineDefinition": str(
+                pipeline_execution.pipeline_definition_for_execution
+            ),
+            "CreationTime": pipeline_execution.creation_time,
+        }
+        return response
+
+    def list_pipeline_parameters_for_execution(
+        self,
+        pipeline_execution_arn,
+    ):
+        pipeline_execution = get_pipeline_execution_from_arn(
+            self.pipelines, pipeline_execution_arn
+        )
+        response = {
+            "PipelineParameters": pipeline_execution.pipeline_parameters,
+        }
+        return response
 
+    def describe_pipeline_execution(
+        self,
+        pipeline_execution_arn,
+    ):
+        pipeline_execution = get_pipeline_execution_from_arn(
+            self.pipelines, pipeline_execution_arn
+        )
+        pipeline_name = get_pipeline_name_from_execution_arn(pipeline_execution_arn)
+        pipeline = get_pipeline_from_name(self.pipelines, pipeline_name)
+
+        pipeline_execution_summaries = {
+            "PipelineArn": pipeline.pipeline_arn,
+            "PipelineExecutionArn": pipeline_execution.pipeline_execution_arn,
+            "PipelineExecutionDisplayName": pipeline_execution.pipeline_execution_display_name,
+            "PipelineExecutionStatus": pipeline_execution.pipeline_execution_status,
+            "PipelineExecutionDescription": pipeline_execution.pipeline_execution_description,
+            "PipelineExperimentConfig": {},
+            "FailureReason": "",
+            "CreationTime": pipeline_execution.creation_time,
+            "LastModifiedTime": pipeline_execution.last_modified_time,
+            "CreatedBy": pipeline_execution.created_by,
+            "LastModifiedBy": pipeline_execution.last_modified_by,
+            "ParallelismConfiguration": pipeline_execution.parallelism_configuration,
+        }
+        return pipeline_execution_summaries
+
+    def describe_pipeline(
+        self,
+        pipeline_name,
+    ):
+        pipeline = get_pipeline_from_name(self.pipelines, pipeline_name)
         response = {
             "PipelineArn": pipeline.pipeline_arn,
             "PipelineName": pipeline.pipeline_name,
diff --git a/moto/sagemaker/responses.py b/moto/sagemaker/responses.py
index 3c33ce99a..2a6cc4bfc 100644
--- a/moto/sagemaker/responses.py
+++ b/moto/sagemaker/responses.py
@@ -472,6 +472,45 @@ class SageMakerResponse(BaseResponse):
         )
         return 200, {}, json.dumps(response)
 
+    @amzn_request_id
+    def start_pipeline_execution(self):
+        response = self.sagemaker_backend.start_pipeline_execution(
+            self._get_param("PipelineName"),
+            self._get_param("PipelineExecutionDisplayName"),
+            self._get_param("PipelineParameters"),
+            self._get_param("PipelineExecutionDescription"),
+            self._get_param("ParallelismConfiguration"),
+        )
+        return 200, {}, json.dumps(response)
+
+    @amzn_request_id
+    def describe_pipeline_execution(self):
+        response = self.sagemaker_backend.describe_pipeline_execution(
+            self._get_param("PipelineExecutionArn")
+        )
+        return 200, {}, json.dumps(response)
+
+    @amzn_request_id
+    def describe_pipeline_definition_for_execution(self):
+        response = self.sagemaker_backend.describe_pipeline_definition_for_execution(
+            self._get_param("PipelineExecutionArn")
+        )
+        return 200, {}, json.dumps(response)
+
+    @amzn_request_id
+    def list_pipeline_parameters_for_execution(self):
+        response = self.sagemaker_backend.list_pipeline_parameters_for_execution(
+            self._get_param("PipelineExecutionArn")
+        )
+        return 200, {}, json.dumps(response)
+
+    @amzn_request_id
+    def list_pipeline_executions(self):
+        response = self.sagemaker_backend.list_pipeline_executions(
+            self._get_param("PipelineName")
+        )
+        return 200, {}, json.dumps(response)
+
     @amzn_request_id
     def create_pipeline(self):
         pipeline = self.sagemaker_backend.create_pipeline(
diff --git a/moto/sagemaker/utils.py b/moto/sagemaker/utils.py
index 1a83177ff..130ce41d2 100644
--- a/moto/sagemaker/utils.py
+++ b/moto/sagemaker/utils.py
@@ -1,5 +1,32 @@
 from moto.s3.models import s3_backends
 import json
+from .exceptions import ValidationError
+
+
+def get_pipeline_from_name(pipelines, pipeline_name):
+    try:
+        pipeline = pipelines[pipeline_name]
+        return pipeline
+    except KeyError:
+        raise ValidationError(
+            message=f"Could not find pipeline with PipelineName {pipeline_name}."
+        )
+
+
+def get_pipeline_name_from_execution_arn(pipeline_execution_arn):
+    return pipeline_execution_arn.split("/")[1].split(":")[-1]
+
+
+def get_pipeline_execution_from_arn(pipelines, pipeline_execution_arn):
+    try:
+        pipeline_name = get_pipeline_name_from_execution_arn(pipeline_execution_arn)
+        pipeline = get_pipeline_from_name(pipelines, pipeline_name)
+        pipeline_execution = pipeline.pipeline_executions[pipeline_execution_arn]
+        return pipeline_execution
+    except KeyError:
+        raise ValidationError(
+            message=f"Could not find pipeline execution with PipelineExecutionArn {pipeline_execution_arn}."
+        )
 
 
 def load_pipeline_definition_from_s3(pipeline_definition_s3_location, account_id):


Test patch:
diff --git a/tests/test_rds/test_rds.py b/tests/test_rds/test_rds.py
index 006903788..cb452c06b 100644
--- a/tests/test_rds/test_rds.py
+++ b/tests/test_rds/test_rds.py
@@ -75,6 +75,179 @@ def test_create_database_no_allocated_storage():
     db_instance["Engine"].should.equal("postgres")
     db_instance["StorageType"].should.equal("gp2")
     db_instance["AllocatedStorage"].should.equal(20)
+    db_instance["PreferredMaintenanceWindow"].should.equal("wed:06:38-wed:07:08")
+
+
+@mock_rds
+def test_create_database_invalid_preferred_maintenance_window_more_24_hours():
+    conn = boto3.client("rds", region_name="us-west-2")
+    with pytest.raises(ClientError) as ex:
+        conn.create_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            Engine="postgres",
+            DBName="staging-postgres",
+            DBInstanceClass="db.m1.small",
+            PreferredMaintenanceWindow="mon:16:00-tue:17:00",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.equal("Maintenance window must be less than 24 hours.")
+
+
+@mock_rds
+def test_create_database_invalid_preferred_maintenance_window_less_30_mins():
+    conn = boto3.client("rds", region_name="us-west-2")
+    with pytest.raises(ClientError) as ex:
+        conn.create_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            Engine="postgres",
+            DBName="staging-postgres",
+            DBInstanceClass="db.m1.small",
+            PreferredMaintenanceWindow="mon:16:00-mon:16:05",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.equal("The maintenance window must be at least 30 minutes.")
+
+
+@mock_rds
+def test_create_database_invalid_preferred_maintenance_window_value():
+    conn = boto3.client("rds", region_name="us-west-2")
+    with pytest.raises(ClientError) as ex:
+        conn.create_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            Engine="postgres",
+            DBName="staging-postgres",
+            DBInstanceClass="db.m1.small",
+            PreferredMaintenanceWindow="sim:16:00-mon:16:30",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.contain("Invalid day:hour:minute")
+
+
+@mock_rds
+def test_create_database_invalid_preferred_maintenance_window_format():
+    conn = boto3.client("rds", region_name="us-west-2")
+    with pytest.raises(ClientError) as ex:
+        conn.create_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            Engine="postgres",
+            DBName="staging-postgres",
+            DBInstanceClass="db.m1.small",
+            PreferredMaintenanceWindow="mon:16tue:17:00",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.contain(
+        "Should be specified as a range ddd:hh24:mi-ddd:hh24:mi (24H Clock UTC). Example: Sun:23:45-Mon:00:15"
+    )
+
+
+@mock_rds
+def test_create_database_preferred_backup_window_overlap_no_spill():
+    conn = boto3.client("rds", region_name="us-west-2")
+    with pytest.raises(ClientError) as ex:
+        conn.create_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            Engine="postgres",
+            DBName="staging-postgres",
+            DBInstanceClass="db.m1.small",
+            PreferredMaintenanceWindow="wed:18:00-wed:22:00",
+            PreferredBackupWindow="20:00-20:30",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.contain(
+        "The backup window and maintenance window must not overlap."
+    )
+
+
+@mock_rds
+def test_create_database_preferred_backup_window_overlap_maintenance_window_spill():
+    conn = boto3.client("rds", region_name="us-west-2")
+    with pytest.raises(ClientError) as ex:
+        conn.create_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            Engine="postgres",
+            DBName="staging-postgres",
+            DBInstanceClass="db.m1.small",
+            PreferredMaintenanceWindow="wed:18:00-thu:01:00",
+            PreferredBackupWindow="00:00-00:30",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.contain(
+        "The backup window and maintenance window must not overlap."
+    )
+
+
+@mock_rds
+def test_create_database_preferred_backup_window_overlap_backup_window_spill():
+    conn = boto3.client("rds", region_name="us-west-2")
+    with pytest.raises(ClientError) as ex:
+        conn.create_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            Engine="postgres",
+            DBName="staging-postgres",
+            DBInstanceClass="db.m1.small",
+            PreferredMaintenanceWindow="thu:00:00-thu:14:00",
+            PreferredBackupWindow="23:50-00:20",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.contain(
+        "The backup window and maintenance window must not overlap."
+    )
+
+
+@mock_rds
+def test_create_database_preferred_backup_window_overlap_both_spill():
+    conn = boto3.client("rds", region_name="us-west-2")
+    with pytest.raises(ClientError) as ex:
+        conn.create_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            Engine="postgres",
+            DBName="staging-postgres",
+            DBInstanceClass="db.m1.small",
+            PreferredMaintenanceWindow="wed:18:00-thu:01:00",
+            PreferredBackupWindow="23:50-00:20",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.contain(
+        "The backup window and maintenance window must not overlap."
+    )
+
+
+@mock_rds
+def test_create_database_valid_preferred_maintenance_window_format():
+    conn = boto3.client("rds", region_name="us-west-2")
+    database = conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        Engine="postgres",
+        DBName="staging-postgres",
+        DBInstanceClass="db.m1.small",
+        PreferredMaintenanceWindow="sun:16:00-sun:16:30",
+    )
+    db_instance = database["DBInstance"]
+    db_instance["DBInstanceClass"].should.equal("db.m1.small")
+    db_instance["PreferredMaintenanceWindow"].should.equal("sun:16:00-sun:16:30")
+
+
+@mock_rds
+def test_create_database_valid_preferred_maintenance_window_uppercase_format():
+    conn = boto3.client("rds", region_name="us-west-2")
+    database = conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        Engine="postgres",
+        DBName="staging-postgres",
+        DBInstanceClass="db.m1.small",
+        PreferredMaintenanceWindow="MON:16:00-TUE:01:30",
+    )
+    db_instance = database["DBInstance"]
+    db_instance["DBInstanceClass"].should.equal("db.m1.small")
+    db_instance["PreferredMaintenanceWindow"].should.equal("mon:16:00-tue:01:30")
 
 
 @mock_rds
@@ -397,6 +570,9 @@ def test_modify_db_instance():
     )
     instances = conn.describe_db_instances(DBInstanceIdentifier="db-master-1")
     instances["DBInstances"][0]["AllocatedStorage"].should.equal(20)
+    instances["DBInstances"][0]["PreferredMaintenanceWindow"].should.equal(
+        "wed:06:38-wed:07:08"
+    )
     instances["DBInstances"][0]["VpcSecurityGroups"][0][
         "VpcSecurityGroupId"
     ].should.equal("sg-123456")
@@ -423,6 +599,252 @@ def test_modify_db_instance_not_existent_db_parameter_group_name():
     ).should.throw(ClientError)
 
 
+@mock_rds
+def test_modify_db_instance_valid_preferred_maintenance_window():
+    conn = boto3.client("rds", region_name="us-west-2")
+    conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        AllocatedStorage=10,
+        DBInstanceClass="postgres",
+        Engine="db.m1.small",
+        MasterUsername="root",
+        MasterUserPassword="hunter2",
+        Port=1234,
+        DBSecurityGroups=["my_sg"],
+    )
+    instances = conn.describe_db_instances(DBInstanceIdentifier="db-master-1")
+    conn.modify_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        PreferredMaintenanceWindow="sun:16:00-sun:16:30",
+    )
+    instances = conn.describe_db_instances(DBInstanceIdentifier="db-master-1")
+    instances["DBInstances"][0]["PreferredMaintenanceWindow"].should.equal(
+        "sun:16:00-sun:16:30"
+    )
+
+
+@mock_rds
+def test_modify_db_instance_valid_preferred_maintenance_window_uppercase():
+    conn = boto3.client("rds", region_name="us-west-2")
+    conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        AllocatedStorage=10,
+        DBInstanceClass="postgres",
+        Engine="db.m1.small",
+        MasterUsername="root",
+        MasterUserPassword="hunter2",
+        Port=1234,
+        DBSecurityGroups=["my_sg"],
+    )
+    instances = conn.describe_db_instances(DBInstanceIdentifier="db-master-1")
+    conn.modify_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        PreferredMaintenanceWindow="SUN:16:00-SUN:16:30",
+    )
+    instances = conn.describe_db_instances(DBInstanceIdentifier="db-master-1")
+    instances["DBInstances"][0]["PreferredMaintenanceWindow"].should.equal(
+        "sun:16:00-sun:16:30"
+    )
+
+
+@mock_rds
+def test_modify_db_instance_invalid_preferred_maintenance_window_more_than_24_hours():
+    conn = boto3.client("rds", region_name="us-west-2")
+    conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        AllocatedStorage=10,
+        DBInstanceClass="postgres",
+        Engine="db.m1.small",
+        MasterUsername="root",
+        MasterUserPassword="hunter2",
+        Port=1234,
+        DBSecurityGroups=["my_sg"],
+    )
+    with pytest.raises(ClientError) as ex:
+        conn.modify_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            PreferredMaintenanceWindow="sun:16:00-sat:16:30",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.equal("Maintenance window must be less than 24 hours.")
+
+
+@mock_rds
+def test_modify_db_instance_invalid_preferred_maintenance_window_less_than_30_mins():
+    conn = boto3.client("rds", region_name="us-west-2")
+    conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        AllocatedStorage=10,
+        DBInstanceClass="postgres",
+        Engine="db.m1.small",
+        MasterUsername="root",
+        MasterUserPassword="hunter2",
+        Port=1234,
+        DBSecurityGroups=["my_sg"],
+    )
+    with pytest.raises(ClientError) as ex:
+        conn.modify_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            PreferredMaintenanceWindow="sun:16:00-sun:16:10",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.equal("The maintenance window must be at least 30 minutes.")
+
+
+@mock_rds
+def test_modify_db_instance_invalid_preferred_maintenance_window_value():
+    conn = boto3.client("rds", region_name="us-west-2")
+    conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        AllocatedStorage=10,
+        DBInstanceClass="postgres",
+        Engine="db.m1.small",
+        MasterUsername="root",
+        MasterUserPassword="hunter2",
+        Port=1234,
+        DBSecurityGroups=["my_sg"],
+    )
+    with pytest.raises(ClientError) as ex:
+        conn.modify_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            PreferredMaintenanceWindow="sin:16:00-sun:16:30",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.contain("Invalid day:hour:minute value")
+
+
+@mock_rds
+def test_modify_db_instance_invalid_preferred_maintenance_window_format():
+    conn = boto3.client("rds", region_name="us-west-2")
+    conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        AllocatedStorage=10,
+        DBInstanceClass="postgres",
+        Engine="db.m1.small",
+        MasterUsername="root",
+        MasterUserPassword="hunter2",
+        Port=1234,
+        DBSecurityGroups=["my_sg"],
+    )
+    with pytest.raises(ClientError) as ex:
+        conn.modify_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            PreferredMaintenanceWindow="sun:16:00sun:16:30",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.contain(
+        "Should be specified as a range ddd:hh24:mi-ddd:hh24:mi (24H Clock UTC). Example: Sun:23:45-Mon:00:15"
+    )
+
+
+@mock_rds
+def test_modify_db_instance_maintenance_backup_window_no_spill():
+    conn = boto3.client("rds", region_name="us-west-2")
+    conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        AllocatedStorage=10,
+        DBInstanceClass="postgres",
+        Engine="db.m1.small",
+        MasterUsername="root",
+        MasterUserPassword="hunter2",
+        Port=1234,
+        DBSecurityGroups=["my_sg"],
+    )
+    with pytest.raises(ClientError) as ex:
+        conn.modify_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            PreferredMaintenanceWindow="sun:16:00-sun:16:30",
+            PreferredBackupWindow="15:50-16:20",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.equal(
+        "The backup window and maintenance window must not overlap."
+    )
+
+
+@mock_rds
+def test_modify_db_instance_maintenance_backup_window_maintenance_spill():
+    conn = boto3.client("rds", region_name="us-west-2")
+    conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        AllocatedStorage=10,
+        DBInstanceClass="postgres",
+        Engine="db.m1.small",
+        MasterUsername="root",
+        MasterUserPassword="hunter2",
+        Port=1234,
+        DBSecurityGroups=["my_sg"],
+    )
+    with pytest.raises(ClientError) as ex:
+        conn.modify_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            PreferredMaintenanceWindow="sun:16:00-mon:15:00",
+            PreferredBackupWindow="00:00-00:30",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.equal(
+        "The backup window and maintenance window must not overlap."
+    )
+
+
+@mock_rds
+def test_modify_db_instance_maintenance_backup_window_backup_spill():
+    conn = boto3.client("rds", region_name="us-west-2")
+    conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        AllocatedStorage=10,
+        DBInstanceClass="postgres",
+        Engine="db.m1.small",
+        MasterUsername="root",
+        MasterUserPassword="hunter2",
+        Port=1234,
+        DBSecurityGroups=["my_sg"],
+    )
+    with pytest.raises(ClientError) as ex:
+        conn.modify_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            PreferredMaintenanceWindow="mon:00:00-mon:15:00",
+            PreferredBackupWindow="23:50-00:20",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.equal(
+        "The backup window and maintenance window must not overlap."
+    )
+
+
+@mock_rds
+def test_modify_db_instance_maintenance_backup_window_both_spill():
+    conn = boto3.client("rds", region_name="us-west-2")
+    conn.create_db_instance(
+        DBInstanceIdentifier="db-master-1",
+        AllocatedStorage=10,
+        DBInstanceClass="postgres",
+        Engine="db.m1.small",
+        MasterUsername="root",
+        MasterUserPassword="hunter2",
+        Port=1234,
+        DBSecurityGroups=["my_sg"],
+    )
+    with pytest.raises(ClientError) as ex:
+        conn.modify_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            PreferredMaintenanceWindow="sun:16:00-mon:15:00",
+            PreferredBackupWindow="23:20-00:20",
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.equal(
+        "The backup window and maintenance window must not overlap."
+    )
+
+
 @mock_rds
 def test_rename_db_instance():
     conn = boto3.client("rds", region_name="us-west-2")
diff --git a/tests/test_sagemaker/test_sagemaker_pipeline.py b/tests/test_sagemaker/test_sagemaker_pipeline.py
index db76a42e5..610c85b8c 100644
--- a/tests/test_sagemaker/test_sagemaker_pipeline.py
+++ b/tests/test_sagemaker/test_sagemaker_pipeline.py
@@ -10,8 +10,16 @@ from unittest import SkipTest
 
 from moto.s3 import mock_s3
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
+from moto.sagemaker.exceptions import ValidationError
+from moto.sagemaker.utils import (
+    get_pipeline_from_name,
+    get_pipeline_execution_from_arn,
+    get_pipeline_name_from_execution_arn,
+)
+from moto.sagemaker.models import FakePipeline
 from moto.sagemaker.utils import arn_formatter, load_pipeline_definition_from_s3
 
+
 FAKE_ROLE_ARN = f"arn:aws:iam::{ACCOUNT_ID}:role/FakeRole"
 TEST_REGION_NAME = "us-west-1"
 
@@ -48,6 +56,190 @@ def create_sagemaker_pipelines(sagemaker_client, pipelines, wait_seconds=0.0):
     return responses
 
 
+def test_utils_get_pipeline_from_name_exists():
+    fake_pipeline_names = ["APipelineName", "BPipelineName"]
+    pipelines = {
+        fake_pipeline_name: FakePipeline(
+            pipeline_name="BFakePipeline",
+            pipeline_display_name="BFakePipeline",
+            pipeline_description=" ",
+            tags=[],
+            parallelism_configuration={},
+            pipeline_definition=" ",
+            role_arn=FAKE_ROLE_ARN,
+            account_id=ACCOUNT_ID,
+            region_name=TEST_REGION_NAME,
+        )
+        for fake_pipeline_name in fake_pipeline_names
+    }
+    retrieved_pipeline = get_pipeline_from_name(
+        pipelines=pipelines, pipeline_name=fake_pipeline_names[0]
+    )
+    assert retrieved_pipeline == pipelines[fake_pipeline_names[0]]
+
+
+def test_utils_get_pipeline_from_name_not_exists():
+    with pytest.raises(ValidationError):
+        _ = get_pipeline_from_name(pipelines={}, pipeline_name="foo")
+
+
+def test_utils_get_pipeline_name_from_execution_arn():
+    expected_pipeline_name = "some-pipeline-name"
+    pipeline_execution_arn = f"arn:aws:sagemaker:{TEST_REGION_NAME}:{ACCOUNT_ID}:pipeline/{expected_pipeline_name}/execution/abc123def456"
+    observed_pipeline_name = get_pipeline_name_from_execution_arn(
+        pipeline_execution_arn=pipeline_execution_arn
+    )
+    assert expected_pipeline_name == observed_pipeline_name
+
+
+def test_utils_get_pipeline_execution_from_arn_not_exists():
+    with pytest.raises(ValidationError):
+        _ = get_pipeline_execution_from_arn(
+            pipelines={},
+            pipeline_execution_arn="some/random/non/existent/arn",
+        )
+
+
+def test_utils_arn_formatter():
+    expected_arn = (
+        f"arn:aws:sagemaker:{TEST_REGION_NAME}:{ACCOUNT_ID}:pipeline/some-pipeline-name"
+    )
+    observed_arn = arn_formatter(
+        _type="pipeline",
+        _id="some-pipeline-name",
+        region_name=TEST_REGION_NAME,
+        account_id=ACCOUNT_ID,
+    )
+    assert expected_arn == observed_arn
+
+
+def test_list_pipeline_executions(sagemaker_client):
+    fake_pipeline_names = ["APipelineName"]
+    pipelines = [
+        {
+            "PipelineName": fake_pipeline_names[0],
+            "RoleArn": FAKE_ROLE_ARN,
+            "PipelineDefinition": " ",
+        },
+    ]
+    _ = create_sagemaker_pipelines(sagemaker_client, pipelines)
+    _ = sagemaker_client.start_pipeline_execution(PipelineName=fake_pipeline_names[0])
+    _ = sagemaker_client.start_pipeline_execution(PipelineName=fake_pipeline_names[0])
+    response = sagemaker_client.list_pipeline_executions(
+        PipelineName=fake_pipeline_names[0]
+    )
+    assert len(response["PipelineExecutionSummaries"]) == 2
+    assert (
+        fake_pipeline_names[0]
+        in response["PipelineExecutionSummaries"][0]["PipelineExecutionArn"]
+    )
+
+
+def test_describe_pipeline_definition_for_execution(sagemaker_client):
+    fake_pipeline_names = ["APipelineName"]
+    pipeline_definition = "some-pipeline-definition"
+    pipelines = [
+        {
+            "PipelineName": fake_pipeline_names[0],
+            "RoleArn": FAKE_ROLE_ARN,
+            "PipelineDefinition": pipeline_definition,
+        },
+    ]
+    _ = create_sagemaker_pipelines(sagemaker_client, pipelines)
+    response = sagemaker_client.start_pipeline_execution(
+        PipelineName=fake_pipeline_names[0]
+    )
+    pipeline_execution_arn = response["PipelineExecutionArn"]
+    response = sagemaker_client.describe_pipeline_definition_for_execution(
+        PipelineExecutionArn=pipeline_execution_arn
+    )
+    assert set(response.keys()) == {
+        "PipelineDefinition",
+        "CreationTime",
+        "ResponseMetadata",
+    }
+    assert response["PipelineDefinition"] == pipeline_definition
+
+
+def test_list_pipeline_parameters_for_execution(sagemaker_client):
+    fake_pipeline_names = ["APipelineName"]
+    pipelines = [
+        {
+            "PipelineName": fake_pipeline_names[0],
+            "RoleArn": FAKE_ROLE_ARN,
+            "PipelineDefinition": " ",
+        },
+    ]
+    _ = create_sagemaker_pipelines(sagemaker_client, pipelines)
+    pipeline_execution_arn = sagemaker_client.start_pipeline_execution(
+        PipelineName=fake_pipeline_names[0],
+        PipelineParameters=[
+            {"Name": "foo", "Value": "bar"},
+        ],
+    )["PipelineExecutionArn"]
+
+    response = sagemaker_client.list_pipeline_parameters_for_execution(
+        PipelineExecutionArn=pipeline_execution_arn
+    )
+    assert isinstance(response["PipelineParameters"], list)
+    assert len(response["PipelineParameters"]) == 1
+    assert response["PipelineParameters"][0]["Name"] == "foo"
+    assert response["PipelineParameters"][0]["Value"] == "bar"
+
+
+def test_start_pipeline_execution(sagemaker_client):
+    fake_pipeline_names = ["APipelineName"]
+    pipelines = [
+        {
+            "PipelineName": fake_pipeline_names[0],
+            "RoleArn": FAKE_ROLE_ARN,
+            "PipelineDefinition": " ",
+        },
+    ]
+    _ = create_sagemaker_pipelines(sagemaker_client, pipelines)
+    pipeline_execution_arn = sagemaker_client.start_pipeline_execution(
+        PipelineName=fake_pipeline_names[0]
+    )
+    assert fake_pipeline_names[0] in pipeline_execution_arn["PipelineExecutionArn"]
+
+
+def test_describe_pipeline_execution_not_exists(sagemaker_client):
+    pipeline_execution_arn = arn_formatter(
+        # random ID (execution ID)
+        "pipeline-execution",
+        "some-pipeline-name",
+        ACCOUNT_ID,
+        TEST_REGION_NAME,
+    )
+    with pytest.raises(botocore.exceptions.ClientError):
+        _ = sagemaker_client.describe_pipeline_execution(
+            PipelineExecutionArn=pipeline_execution_arn
+        )
+
+
+def test_describe_pipeline_execution(sagemaker_client):
+    fake_pipeline_names = ["APipelineName", "BPipelineName"]
+    pipelines = [
+        {
+            "PipelineName": fake_pipeline_name,
+            "RoleArn": FAKE_ROLE_ARN,
+            "PipelineDefinition": " ",
+        }
+        for fake_pipeline_name in fake_pipeline_names
+    ]
+    _ = create_sagemaker_pipelines(sagemaker_client, pipelines)
+    response = sagemaker_client.start_pipeline_execution(
+        PipelineName=fake_pipeline_names[0]
+    )
+    _ = sagemaker_client.start_pipeline_execution(PipelineName=fake_pipeline_names[1])
+    expected_pipeline_execution_arn = response["PipelineExecutionArn"]
+    pipeline_execution_summary = sagemaker_client.describe_pipeline_execution(
+        PipelineExecutionArn=response["PipelineExecutionArn"]
+    )
+    observed_pipeline_execution_arn = pipeline_execution_summary["PipelineExecutionArn"]
+    observed_pipeline_execution_arn.should.be.equal(expected_pipeline_execution_arn)
+
+
 def test_load_pipeline_definition_from_s3():
     if settings.TEST_SERVER_MODE:
         raise SkipTest(