Repository: getmoto/moto
Pull request #6314 (2023-05-11 09:20:33)

Autoscaling: Add metrics for E2 target tracking scaling
## Purpose

AWS recently added support of metric math for target tracking scaling and changed scheme by adding new field `Metrics` to `CustomizedMetricSpecification`.

AWS blog announcement - https://aws.amazon.com/about-aws/whats-new/2022/12/amazon-ec2-auto-scaling-supports-metric-math-target-tracking-policies/

This PR adds `Metrics` to moto responses for ec2 target tracking scaling.

Ivan-Lukianov: @bblommers  thank you for looking into this!

Extended existing test to check `Metrics`, also during writing a test found a bug with ReturnData, fixed that as well.

Patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index e090a1d48..26f6b9d2a 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -3142,7 +3142,7 @@
 
 ## glue
 <details>
-<summary>30% implemented</summary>
+<summary>33% implemented</summary>
 
 - [X] batch_create_partition
 - [ ] batch_delete_connection
@@ -3181,7 +3181,7 @@
 - [X] create_schema
 - [ ] create_script
 - [ ] create_security_configuration
-- [ ] create_session
+- [X] create_session
 - [X] create_table
 - [X] create_trigger
 - [ ] create_user_defined_function
@@ -3205,7 +3205,7 @@
 - [X] delete_schema
 - [ ] delete_schema_versions
 - [ ] delete_security_configuration
-- [ ] delete_session
+- [X] delete_session
 - [X] delete_table
 - [X] delete_table_version
 - [X] delete_trigger
@@ -3258,7 +3258,7 @@
 - [ ] get_schema_versions_diff
 - [ ] get_security_configuration
 - [ ] get_security_configurations
-- [ ] get_session
+- [X] get_session
 - [ ] get_statement
 - [X] get_table
 - [X] get_table_version
@@ -3291,7 +3291,7 @@
 - [X] list_registries
 - [ ] list_schema_versions
 - [ ] list_schemas
-- [ ] list_sessions
+- [X] list_sessions
 - [ ] list_statements
 - [X] list_triggers
 - [ ] list_workflows
@@ -3320,7 +3320,7 @@
 - [ ] start_workflow_run
 - [X] stop_crawler
 - [ ] stop_crawler_schedule
-- [ ] stop_session
+- [X] stop_session
 - [X] stop_trigger
 - [ ] stop_workflow_run
 - [X] tag_resource
diff --git a/docs/docs/services/glue.rst b/docs/docs/services/glue.rst
index 0e0c7407a..2aeac6467 100644
--- a/docs/docs/services/glue.rst
+++ b/docs/docs/services/glue.rst
@@ -66,7 +66,7 @@ glue
 
 - [ ] create_script
 - [ ] create_security_configuration
-- [ ] create_session
+- [X] create_session
 - [X] create_table
 - [X] create_trigger
 - [ ] create_user_defined_function
@@ -90,7 +90,7 @@ glue
 - [X] delete_schema
 - [ ] delete_schema_versions
 - [ ] delete_security_configuration
-- [ ] delete_session
+- [X] delete_session
 - [X] delete_table
 - [X] delete_table_version
 - [X] delete_trigger
@@ -155,7 +155,7 @@ glue
 - [ ] get_schema_versions_diff
 - [ ] get_security_configuration
 - [ ] get_security_configurations
-- [ ] get_session
+- [X] get_session
 - [ ] get_statement
 - [X] get_table
 - [X] get_table_version
@@ -188,7 +188,7 @@ glue
 - [X] list_registries
 - [ ] list_schema_versions
 - [ ] list_schemas
-- [ ] list_sessions
+- [X] list_sessions
 - [ ] list_statements
 - [X] list_triggers
 - [ ] list_workflows
@@ -217,7 +217,7 @@ glue
 - [ ] start_workflow_run
 - [X] stop_crawler
 - [ ] stop_crawler_schedule
-- [ ] stop_session
+- [X] stop_session
 - [X] stop_trigger
 - [ ] stop_workflow_run
 - [X] tag_resource
diff --git a/moto/autoscaling/responses.py b/moto/autoscaling/responses.py
index e6faab3ba..a209cea55 100644
--- a/moto/autoscaling/responses.py
+++ b/moto/autoscaling/responses.py
@@ -1022,6 +1022,38 @@ DESCRIBE_SCALING_POLICIES_TEMPLATE = """<DescribePoliciesResponse xmlns="http://
               {% if policy.target_tracking_config["CustomizedMetricSpecification"].get("Unit") %}
               <Unit>{{ policy.target_tracking_config["CustomizedMetricSpecification"].get("Unit") }}</Unit>
               {% endif %}
+              {% if policy.target_tracking_config["CustomizedMetricSpecification"].get("Metrics") %}
+              <Metrics>
+                {% for metric in policy.target_tracking_config["CustomizedMetricSpecification"].get("Metrics", []) %}
+                <member>
+                  <Id>{{ metric.get("Id") }}</Id>
+                  {% if metric.get("MetricStat") is none %}
+                  <Expression>{{ metric.get("Expression") }}</Expression>
+                  {% endif %}
+                  {% if metric.get("Expression") is none %}
+                  <MetricStat>
+                    <Metric>
+                      <Namespace>{{ metric.get("MetricStat", {}).get("Metric", {}).get("Namespace") }}</Namespace>
+                      <MetricName>{{ metric.get("MetricStat", {}).get("Metric", {}).get("MetricName") }}</MetricName>
+                      <Dimensions>
+                      {% for dim in metric.get("MetricStat", {}).get("Metric", {}).get("Dimensions", []) %}
+                        <member>
+                          <Name>{{ dim.get("Name") }}</Name>
+                          <Value>{{ dim.get("Value") }}</Value>
+                        </member>
+                      {% endfor %}
+                      </Dimensions>
+                    </Metric>
+                    <Stat>{{ metric.get("MetricStat", {}).get("Stat") }}</Stat>
+                    <Unit>{{ metric.get("MetricStat", {}).get("Unit") }}</Unit>
+                  </MetricStat>
+                  {% endif %}
+                  <Label>{{ metric.get("Label") }}</Label>
+                  <ReturnData>{{ 'true' if metric.get("ReturnData") is none else metric.get("ReturnData") }}</ReturnData>
+                </member>
+                {% endfor %}
+              </Metrics>
+              {% endif %}
             </CustomizedMetricSpecification>
             {% endif %}
             <TargetValue>{{ policy.target_tracking_config.get("TargetValue") }}</TargetValue>
diff --git a/moto/backend_index.py b/moto/backend_index.py
index 8708ee366..7b10be586 100644
--- a/moto/backend_index.py
+++ b/moto/backend_index.py
@@ -158,7 +158,6 @@ backend_url_patterns = [
     ("service-quotas", re.compile("https?://servicequotas\\.(.+)\\.amazonaws\\.com")),
     ("ses", re.compile("https?://email\\.(.+)\\.amazonaws\\.com")),
     ("ses", re.compile("https?://ses\\.(.+)\\.amazonaws\\.com")),
-    ("sesv2", re.compile("https?://ses\\.(.+)\\.amazonaws\\.com")),
     ("sesv2", re.compile("https?://email\\.(.+)\\.amazonaws\\.com")),
     ("signer", re.compile("https?://signer\\.(.+)\\.amazonaws\\.com")),
     ("sns", re.compile("https?://sns\\.(.+)\\.amazonaws\\.com")),
diff --git a/moto/cloudformation/parsing.py b/moto/cloudformation/parsing.py
index 9731008a2..cf4f06113 100644
--- a/moto/cloudformation/parsing.py
+++ b/moto/cloudformation/parsing.py
@@ -793,17 +793,7 @@ class ResourceMap(collections_abc.Mapping):  # type: ignore[type-arg]
         for logical_name in resource_names_by_action["Remove"]:
             resource_json = self._resource_json_map[logical_name]
             resource = self._parsed_resources[logical_name]
-            # ToDo: Standardize this.
-            if hasattr(resource, "physical_resource_id"):
-                resource_name = self._parsed_resources[
-                    logical_name
-                ].physical_resource_id
-            else:
-                resource_name = None
-            parse_and_delete_resource(
-                resource_name, resource_json, self._account_id, self._region_name
-            )
-            self._parsed_resources.pop(logical_name)
+            self._delete_resource(resource, resource_json)
 
         self._template = template
         if parameters:
@@ -859,24 +849,12 @@ class ResourceMap(collections_abc.Mapping):  # type: ignore[type-arg]
                         not isinstance(parsed_resource, str)
                         and parsed_resource is not None
                     ):
-                        try:
-                            parsed_resource.delete(self._account_id, self._region_name)
-                        except (TypeError, AttributeError):
-                            if hasattr(parsed_resource, "physical_resource_id"):
-                                resource_name = parsed_resource.physical_resource_id
-                            else:
-                                resource_name = None
-
-                            resource_json = self._resource_json_map[
-                                parsed_resource.logical_resource_id
-                            ]
-
-                            parse_and_delete_resource(
-                                resource_name,
-                                resource_json,
-                                self._account_id,
-                                self._region_name,
-                            )
+
+                        resource_json = self._resource_json_map[
+                            parsed_resource.logical_resource_id
+                        ]
+
+                        self._delete_resource(parsed_resource, resource_json)
 
                         self._parsed_resources.pop(parsed_resource.logical_resource_id)
                 except Exception as e:
@@ -889,6 +867,24 @@ class ResourceMap(collections_abc.Mapping):  # type: ignore[type-arg]
         if tries == 5:
             raise last_exception
 
+    def _delete_resource(
+        self, parsed_resource: Any, resource_json: Dict[str, Any]
+    ) -> None:
+        try:
+            parsed_resource.delete(self._account_id, self._region_name)
+        except (TypeError, AttributeError):
+            if hasattr(parsed_resource, "physical_resource_id"):
+                resource_name = parsed_resource.physical_resource_id
+            else:
+                resource_name = None
+
+            parse_and_delete_resource(
+                resource_name,
+                resource_json,
+                self._account_id,
+                self._region_name,
+            )
+
 
 class OutputMap(collections_abc.Mapping):  # type: ignore[type-arg]
     def __init__(self, resources: ResourceMap, template: Dict[str, Any], stack_id: str):
diff --git a/moto/core/models.py b/moto/core/models.py
index 58c3b24f3..25db72e5e 100644
--- a/moto/core/models.py
+++ b/moto/core/models.py
@@ -265,6 +265,22 @@ def patch_client(client: botocore.client.BaseClient) -> None:
     :return:
     """
     if isinstance(client, botocore.client.BaseClient):
+        # Check if our event handler was already registered
+        try:
+            event_emitter = client._ruleset_resolver._event_emitter._emitter  # type: ignore
+            all_handlers = event_emitter._handlers._root["children"]  # type: ignore
+            handler_trie = list(all_handlers["before-send"].values())[1]  # type: ignore
+            handlers_list = handler_trie.first + handler_trie.middle + handler_trie.last
+            if botocore_stubber in handlers_list:
+                # No need to patch - this client already has the botocore_stubber registered
+                return
+        except:  # noqa: E722 Do not use bare except
+            # Because we're accessing all kinds of private methods, the API may change and newer versions of botocore may throw an exception
+            # One of our tests will fail if this happens (test_patch_can_be_called_on_a_mocked_client)
+            # If this happens for a user, just continue and hope for the best
+            #  - in 99% of the cases there are no duplicate event handlers, so it doesn't matter if the check fails
+            pass
+
         client.meta.events.register("before-send", botocore_stubber)
     else:
         raise Exception(f"Argument {client} should be of type boto3.client")
diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py
index 6a93607d0..4b3373fb7 100644
--- a/moto/ec2/exceptions.py
+++ b/moto/ec2/exceptions.py
@@ -234,6 +234,15 @@ class RouteAlreadyExistsError(EC2ClientError):
         )
 
 
+class RouteNotSupportedError(EC2ClientError):
+    def __init__(self, vpce_id: str):
+        super().__init__(
+            "RouteNotSupported",
+            f"Route table contains unsupported route target: {vpce_id}. "
+            f"VPC Endpoints of this type cannot be used as route targets.",
+        )
+
+
 class InvalidInstanceIdError(EC2ClientError):
     def __init__(self, instance_id: Any):
         if isinstance(instance_id, str):
diff --git a/moto/ec2/models/route_tables.py b/moto/ec2/models/route_tables.py
index 4215a320c..de6217d09 100644
--- a/moto/ec2/models/route_tables.py
+++ b/moto/ec2/models/route_tables.py
@@ -19,6 +19,7 @@ from ..exceptions import (
     InvalidAssociationIdError,
     InvalidDestinationCIDRBlockParameterError,
     RouteAlreadyExistsError,
+    RouteNotSupportedError,
 )
 from ..utils import (
     EC2_RESOURCE_TO_PREFIX,
@@ -141,6 +142,7 @@ class Route(CloudFormationModel):
         interface: Optional[NetworkInterface] = None,
         vpc_pcx: Optional[VPCPeeringConnection] = None,
         carrier_gateway: Optional[CarrierGateway] = None,
+        vpc_endpoint_id: Optional[str] = None,
     ):
         self.id = generate_route_id(
             route_table.id,
@@ -161,6 +163,7 @@ class Route(CloudFormationModel):
         self.interface = interface
         self.vpc_pcx = vpc_pcx
         self.carrier_gateway = carrier_gateway
+        self.vpc_endpoint_id = vpc_endpoint_id
 
     @property
     def physical_resource_id(self) -> str:
@@ -359,6 +362,7 @@ class RouteBackend:
         interface_id: Optional[str] = None,
         vpc_peering_connection_id: Optional[str] = None,
         carrier_gateway_id: Optional[str] = None,
+        vpc_endpoint_id: Optional[str] = None,
     ) -> Route:
         gateway = None
         nat_gateway = None
@@ -368,6 +372,13 @@ class RouteBackend:
         destination_prefix_list = None
         carrier_gateway = None
 
+        if vpc_endpoint_id:
+            vpce = self.describe_vpc_endpoints(vpc_end_point_ids=[vpc_endpoint_id])  # type: ignore[attr-defined]
+            if not vpce[0].endpoint_type == "GatewayLoadBalancer":
+                # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2/client/create_route.html
+                # VpcEndpointId (string) – The ID of a VPC endpoint. Supported for Gateway Load Balancer endpoints only.
+                raise RouteNotSupportedError(vpc_endpoint_id)
+
         route_table = self.get_route_table(route_table_id)
 
         if interface_id:
@@ -414,6 +425,7 @@ class RouteBackend:
             transit_gateway=transit_gateway,
             interface=interface,
             carrier_gateway=carrier_gateway,
+            vpc_endpoint_id=vpc_endpoint_id,
             vpc_pcx=self.get_vpc_peering_connection(vpc_peering_connection_id)  # type: ignore[attr-defined]
             if vpc_peering_connection_id
             else None,
diff --git a/moto/ec2/responses/route_tables.py b/moto/ec2/responses/route_tables.py
index bab437680..be8aeaf04 100644
--- a/moto/ec2/responses/route_tables.py
+++ b/moto/ec2/responses/route_tables.py
@@ -25,6 +25,7 @@ class RouteTables(EC2BaseResponse):
         interface_id = self._get_param("NetworkInterfaceId")
         pcx_id = self._get_param("VpcPeeringConnectionId")
         carrier_gateway_id = self._get_param("CarrierGatewayId")
+        vpc_endpoint_id = self._get_param("VpcEndpointId")
 
         self.ec2_backend.create_route(
             route_table_id,
@@ -39,6 +40,7 @@ class RouteTables(EC2BaseResponse):
             interface_id=interface_id,
             vpc_peering_connection_id=pcx_id,
             carrier_gateway_id=carrier_gateway_id,
+            vpc_endpoint_id=vpc_endpoint_id,
         )
 
         template = self.response_template(CREATE_ROUTE_RESPONSE)
@@ -212,6 +214,11 @@ DESCRIBE_ROUTE_TABLES_RESPONSE = """
                   <origin>CreateRoute</origin>
                   <state>active</state>
                 {% endif %}
+                {% if route.vpc_endpoint_id %}
+                  <gatewayId>{{ route.vpc_endpoint_id }}</gatewayId>
+                  <origin>CreateRoute</origin>
+                  <state>active</state>
+                {% endif %}
                 {% if route.instance %}
                   <instanceId>{{ route.instance.id }}</instanceId>
                   <origin>CreateRoute</origin>
diff --git a/moto/ecs/models.py b/moto/ecs/models.py
index b89bd8a7c..6e230c9b9 100644
--- a/moto/ecs/models.py
+++ b/moto/ecs/models.py
@@ -2,6 +2,7 @@ import re
 from copy import copy
 from datetime import datetime, timezone
 from typing import Any, Dict, Iterator, List, Optional, Tuple
+from os import getenv
 
 from moto import settings
 from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel
@@ -488,6 +489,14 @@ class CapacityProviderFailure(BaseObject):
 
 
 class Service(BaseObject, CloudFormationModel):
+    """Set the environment variable MOTO_ECS_SERVICE_RUNNING to a number of running tasks you want
+    the service to transition to, ie if set to 2:
+
+    MOTO_ECS_SERVICE_RUNNING=2
+
+    then describe_services call to return runningCount of the service AND deployment to 2
+    """
+
     def __init__(
         self,
         cluster: Cluster,
@@ -507,7 +516,6 @@ class Service(BaseObject, CloudFormationModel):
         self.cluster_arn = cluster.arn
         self.name = service_name
         self.status = "ACTIVE"
-        self.running_count = 0
         self.task_definition = task_definition.arn if task_definition else None
         self.desired_count = desired_count
         self.task_sets: List[TaskSet] = []
@@ -515,6 +523,25 @@ class Service(BaseObject, CloudFormationModel):
         self.events: List[Dict[str, Any]] = []
         self.launch_type = launch_type
         self.service_registries = service_registries or []
+        self.load_balancers = load_balancers if load_balancers is not None else []
+        self.scheduling_strategy = (
+            scheduling_strategy if scheduling_strategy is not None else "REPLICA"
+        )
+        self.platform_version = platform_version
+        self.tags = tags if tags is not None else []
+        self.region_name = cluster.region_name
+        self._account_id = backend.account_id
+        self._backend = backend
+
+        try:
+            # negative running count not allowed, set to 0 if so
+            ecs_running_count = max(int(getenv("MOTO_ECS_SERVICE_RUNNING", 0)), 0)
+        except ValueError:
+            # Unable to parse value of MOTO_ECS_SERVICE_RUNNING as an integer, set to default 0
+            ecs_running_count = 0
+
+        self.running_count = ecs_running_count
+        self.pending_count = desired_count - ecs_running_count
         if self.deployment_controller["type"] == "ECS":
             self.deployments = [
                 {
@@ -522,8 +549,8 @@ class Service(BaseObject, CloudFormationModel):
                     "desiredCount": self.desired_count,
                     "id": f"ecs-svc/{mock_random.randint(0, 32**12)}",
                     "launchType": self.launch_type,
-                    "pendingCount": self.desired_count,
-                    "runningCount": 0,
+                    "pendingCount": self.pending_count,
+                    "runningCount": ecs_running_count,
                     "status": "PRIMARY",
                     "taskDefinition": self.task_definition,
                     "updatedAt": datetime.now(timezone.utc),
@@ -531,16 +558,6 @@ class Service(BaseObject, CloudFormationModel):
             ]
         else:
             self.deployments = []
-        self.load_balancers = load_balancers if load_balancers is not None else []
-        self.scheduling_strategy = (
-            scheduling_strategy if scheduling_strategy is not None else "REPLICA"
-        )
-        self.platform_version = platform_version
-        self.tags = tags if tags is not None else []
-        self.pending_count = 0
-        self.region_name = cluster.region_name
-        self._account_id = backend.account_id
-        self._backend = backend
 
     @property
     def arn(self) -> str:
@@ -1641,6 +1658,7 @@ class EC2ContainerServiceBackend(BaseBackend):
             # It is only deleted later on
             # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs.html#ECS.Client.delete_service
             service.status = "INACTIVE"
+            service.pending_count = 0
             return service
 
     def register_container_instance(
diff --git a/moto/glue/exceptions.py b/moto/glue/exceptions.py
index d31932042..dfa71bcfe 100644
--- a/moto/glue/exceptions.py
+++ b/moto/glue/exceptions.py
@@ -31,6 +31,11 @@ class CrawlerAlreadyExistsException(AlreadyExistsException):
         super().__init__("Crawler")
 
 
+class SessionAlreadyExistsException(AlreadyExistsException):
+    def __init__(self) -> None:
+        super().__init__("Session")
+
+
 class EntityNotFoundException(GlueClientError):
     def __init__(self, msg: str):
         super().__init__("EntityNotFoundException", msg)
@@ -107,6 +112,16 @@ class SchemaVersionNotFoundFromSchemaVersionIdException(EntityNotFoundException)
         )
 
 
+class SessionNotFoundException(EntityNotFoundException):
+    def __init__(self, session: str):
+        super().__init__(f"Session {session} not found.")
+
+
+class IllegalSessionStateException(GlueClientError):
+    def __init__(self, msg: str):
+        super().__init__("IllegalSessionStateException", msg)
+
+
 class RegistryNotFoundException(EntityNotFoundException):
     def __init__(self, resource: str, param_name: str, param_value: Optional[str]):
         super().__init__(
diff --git a/moto/glue/models.py b/moto/glue/models.py
index 6b288d650..a15c89152 100644
--- a/moto/glue/models.py
+++ b/moto/glue/models.py
@@ -30,6 +30,9 @@ from .exceptions import (
     SchemaVersionNotFoundFromSchemaIdException,
     SchemaNotFoundException,
     SchemaVersionMetadataAlreadyExistsException,
+    SessionAlreadyExistsException,
+    SessionNotFoundException,
+    IllegalSessionStateException,
     TriggerNotFoundException,
 )
 from .utils import PartitionFilter
@@ -82,6 +85,12 @@ class GlueBackend(BaseBackend):
             "limit_default": 100,
             "unique_attribute": "name",
         },
+        "list_sessions": {
+            "input_token": "next_token",
+            "limit_key": "max_results",
+            "limit_default": 100,
+            "unique_attribute": "session_id",
+        },
         "list_triggers": {
             "input_token": "next_token",
             "limit_key": "max_results",
@@ -96,6 +105,7 @@ class GlueBackend(BaseBackend):
         self.crawlers: Dict[str, FakeCrawler] = OrderedDict()
         self.jobs: Dict[str, FakeJob] = OrderedDict()
         self.job_runs: Dict[str, FakeJobRun] = OrderedDict()
+        self.sessions: Dict[str, FakeSession] = OrderedDict()
         self.tagger = TaggingService()
         self.triggers: Dict[str, FakeTrigger] = OrderedDict()
         self.registries: Dict[str, FakeRegistry] = OrderedDict()
@@ -771,6 +781,67 @@ class GlueBackend(BaseBackend):
 
         return schema.as_dict()
 
+    def create_session(
+        self,
+        session_id: str,
+        description: str,
+        role: str,
+        command: Dict[str, str],
+        timeout: int,
+        idle_timeout: int,
+        default_arguments: Dict[str, str],
+        connections: Dict[str, List[str]],
+        max_capacity: float,
+        number_of_workers: int,
+        worker_type: str,
+        security_configuration: str,
+        glue_version: str,
+        tags: Dict[str, str],
+        request_origin: str,
+    ) -> None:
+        if session_id in self.sessions:
+            raise SessionAlreadyExistsException()
+
+        session = FakeSession(
+            session_id,
+            description,
+            role,
+            command,
+            timeout,
+            idle_timeout,
+            default_arguments,
+            connections,
+            max_capacity,
+            number_of_workers,
+            worker_type,
+            security_configuration,
+            glue_version,
+            tags,
+            request_origin,
+            backend=self,
+        )
+        self.sessions[session_id] = session
+
+    def get_session(self, session_id: str) -> "FakeSession":
+        try:
+            return self.sessions[session_id]
+        except KeyError:
+            raise SessionNotFoundException(session_id)
+
+    @paginate(pagination_model=PAGINATION_MODEL)  # type: ignore[misc]
+    def list_sessions(self) -> List["FakeSession"]:  # type: ignore[misc]
+        return [session for _, session in self.sessions.items()]
+
+    def stop_session(self, session_id: str) -> None:
+        session = self.get_session(session_id)
+        session.stop_session()
+
+    def delete_session(self, session_id: str) -> None:
+        try:
+            del self.sessions[session_id]
+        except KeyError:
+            raise SessionNotFoundException(session_id)
+
     def create_trigger(
         self,
         name: str,
@@ -1517,6 +1588,74 @@ class FakeSchemaVersion(BaseModel):
         }
 
 
+class FakeSession(BaseModel):
+    def __init__(
+        self,
+        session_id: str,
+        description: str,
+        role: str,
+        command: Dict[str, str],
+        timeout: int,
+        idle_timeout: int,
+        default_arguments: Dict[str, str],
+        connections: Dict[str, List[str]],
+        max_capacity: float,
+        number_of_workers: int,
+        worker_type: str,
+        security_configuration: str,
+        glue_version: str,
+        tags: Dict[str, str],
+        request_origin: str,
+        backend: GlueBackend,
+    ):
+        self.session_id = session_id
+        self.description = description
+        self.role = role
+        self.command = command
+        self.timeout = timeout
+        self.idle_timeout = idle_timeout
+        self.default_arguments = default_arguments
+        self.connections = connections
+        self.max_capacity = max_capacity
+        self.number_of_workers = number_of_workers
+        self.worker_type = worker_type
+        self.security_configuration = security_configuration
+        self.glue_version = glue_version
+        self.tags = tags
+        self.request_origin = request_origin
+        self.creation_time = datetime.utcnow()
+        self.last_updated = self.creation_time
+        self.arn = f"arn:aws:glue:{backend.region_name}:{backend.account_id}:session/{self.session_id}"
+        self.backend = backend
+        self.backend.tag_resource(self.arn, tags)
+        self.state = "READY"
+
+    def get_id(self) -> str:
+        return self.session_id
+
+    def as_dict(self) -> Dict[str, Any]:
+        return {
+            "Id": self.session_id,
+            "CreatedOn": self.creation_time.isoformat(),
+            "Status": self.state,
+            "ErrorMessage": "string",
+            "Description": self.description,
+            "Role": self.role,
+            "Command": self.command,
+            "DefaultArguments": self.default_arguments,
+            "Connections": self.connections,
+            "Progress": 12.3,
+            "MaxCapacity": 123.0,
+            "SecurityConfiguration": self.security_configuration,
+            "GlueVersion": self.glue_version,
+        }
+
+    def stop_session(self) -> None:
+        if self.state != "READY":
+            raise IllegalSessionStateException(f"Session is in {self.state} status")
+        self.state = "STOPPING"
+
+
 class FakeTrigger(BaseModel):
     def __init__(
         self,
diff --git a/moto/glue/responses.py b/moto/glue/responses.py
index b35214b2f..591aa6def 100644
--- a/moto/glue/responses.py
+++ b/moto/glue/responses.py
@@ -3,7 +3,7 @@ from typing import Any, Dict, List
 
 from moto.core.common_types import TYPE_RESPONSE
 from moto.core.responses import BaseResponse
-from .models import glue_backends, GlueBackend, FakeJob, FakeCrawler
+from .models import glue_backends, GlueBackend, FakeJob, FakeCrawler, FakeSession
 
 
 class GlueResponse(BaseResponse):
@@ -527,6 +527,72 @@ class GlueResponse(BaseResponse):
         schema = self.glue_backend.update_schema(schema_id, compatibility, description)
         return json.dumps(schema)
 
+    def create_session(self) -> str:
+        self.glue_backend.create_session(
+            session_id=self.parameters.get("Id"),  # type: ignore[arg-type]
+            description=self.parameters.get("Description"),  # type: ignore[arg-type]
+            role=self.parameters.get("Role"),  # type: ignore[arg-type]
+            command=self.parameters.get("Command"),  # type: ignore[arg-type]
+            timeout=self.parameters.get("Timeout"),  # type: ignore[arg-type]
+            idle_timeout=self.parameters.get("IdleTimeout"),  # type: ignore[arg-type]
+            default_arguments=self.parameters.get("DefaultArguments"),  # type: ignore[arg-type]
+            connections=self.parameters.get("Connections"),  # type: ignore[arg-type]
+            max_capacity=self.parameters.get("MaxCapacity"),  # type: ignore[arg-type]
+            number_of_workers=self.parameters.get("NumberOfWorkers"),  # type: ignore[arg-type]
+            worker_type=self.parameters.get("WorkerType"),  # type: ignore[arg-type]
+            security_configuration=self.parameters.get("SecurityConfiguration"),  # type: ignore[arg-type]
+            glue_version=self.parameters.get("GlueVersion"),  # type: ignore[arg-type]
+            tags=self.parameters.get("Tags"),  # type: ignore[arg-type]
+            request_origin=self.parameters.get("RequestOrigin"),  # type: ignore[arg-type]
+        )
+        return ""
+
+    def get_session(self) -> str:
+        session_id = self.parameters.get("Id")
+        session = self.glue_backend.get_session(session_id)  # type: ignore[arg-type]
+        return json.dumps({"Session": session.as_dict()})
+
+    def list_sessions(self) -> str:
+        next_token = self._get_param("NextToken")
+        max_results = self._get_int_param("MaxResults")
+        tags = self._get_param("Tags")
+        sessions, next_token = self.glue_backend.list_sessions(
+            next_token=next_token, max_results=max_results
+        )
+        filtered_session_ids = self._filter_sessions_by_tags(sessions, tags)
+
+        return json.dumps(
+            dict(
+                Ids=[session_id for session_id in filtered_session_ids],
+                Sessions=[
+                    self.glue_backend.get_session(session_id).as_dict()
+                    for session_id in filtered_session_ids
+                ],
+                NextToken=next_token,
+            )
+        )
+
+    def _filter_sessions_by_tags(
+        self, sessions: List[FakeSession], tags: Dict[str, str]
+    ) -> List[str]:
+        if not tags:
+            return [session.get_id() for session in sessions]
+        return [
+            session.get_id()
+            for session in sessions
+            if self.is_tags_match(session.arn, tags)
+        ]
+
+    def stop_session(self) -> str:
+        session_id = self.parameters.get("Id")
+        self.glue_backend.stop_session(session_id)  # type: ignore[arg-type]
+        return json.dumps({"Id": session_id})
+
+    def delete_session(self) -> str:
+        session_id = self.parameters.get("Id")
+        self.glue_backend.delete_session(session_id)  # type: ignore[arg-type]
+        return json.dumps({"Id": session_id})
+
     def batch_get_crawlers(self) -> str:
         crawler_names = self._get_param("CrawlerNames")
         crawlers = self.glue_backend.batch_get_crawlers(crawler_names)
diff --git a/moto/s3/models.py b/moto/s3/models.py
index 75ad72fac..c7f4e9527 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -2418,6 +2418,10 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
             # Object copied from Glacier object should not have expiry
             new_key.set_expiry(None)
 
+        if src_key.checksum_value:
+            new_key.checksum_value = src_key.checksum_value
+            new_key.checksum_algorithm = src_key.checksum_algorithm
+
         # Send notifications that an object was copied
         notifications.send_event(
             self.account_id, notifications.S3_OBJECT_CREATE_COPY, bucket, new_key
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index 7bf70dfbf..c0f24fe85 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -2102,10 +2102,10 @@ class S3Response(BaseResponse):
         ps = minidom.parseString(body).getElementsByTagName("Part")
         prev = 0
         for p in ps:
-            pn = int(p.getElementsByTagName("PartNumber")[0].firstChild.wholeText)
+            pn = int(p.getElementsByTagName("PartNumber")[0].firstChild.wholeText)  # type: ignore[union-attr]
             if pn <= prev:
                 raise InvalidPartOrder()
-            yield (pn, p.getElementsByTagName("ETag")[0].firstChild.wholeText)
+            yield (pn, p.getElementsByTagName("ETag")[0].firstChild.wholeText)  # type: ignore[union-attr]
 
     def _key_response_post(
         self,
diff --git a/moto/s3/utils.py b/moto/s3/utils.py
index 855ce4a4f..fafabdcd1 100644
--- a/moto/s3/utils.py
+++ b/moto/s3/utils.py
@@ -200,7 +200,7 @@ def compute_checksum(body: bytes, algorithm: str) -> bytes:
     if algorithm == "SHA1":
         hashed_body = _hash(hashlib.sha1, (body,))
     elif algorithm == "CRC32" or algorithm == "CRC32C":
-        hashed_body = f"{binascii.crc32(body)}".encode("utf-8")
+        hashed_body = binascii.crc32(body).to_bytes(4, "big")
     else:
         hashed_body = _hash(hashlib.sha256, (body,))
     return base64.b64encode(hashed_body)
@@ -208,10 +208,10 @@ def compute_checksum(body: bytes, algorithm: str) -> bytes:
 
 def _hash(fn: Any, args: Any) -> bytes:
     try:
-        return fn(*args, usedforsecurity=False).hexdigest().encode("utf-8")
+        return fn(*args, usedforsecurity=False).digest()
     except TypeError:
         # The usedforsecurity-parameter is only available as of Python 3.9
-        return fn(*args).hexdigest().encode("utf-8")
+        return fn(*args).digest()
 
 
 def cors_matches_origin(origin_header: str, allowed_origins: List[str]) -> bool:
diff --git a/moto/secretsmanager/models.py b/moto/secretsmanager/models.py
index bb492c4ab..53d049ef0 100644
--- a/moto/secretsmanager/models.py
+++ b/moto/secretsmanager/models.py
@@ -88,6 +88,8 @@ class FakeSecret:
         self.auto_rotate_after_days = 0
         self.deleted_date: Optional[float] = None
         self.policy: Optional[str] = None
+        self.next_rotation_date: Optional[int] = None
+        self.last_rotation_date: Optional[int] = None
 
     def update(
         self,
@@ -167,9 +169,10 @@ class FakeSecret:
             "RotationEnabled": self.rotation_enabled,
             "RotationLambdaARN": self.rotation_lambda_arn,
             "RotationRules": {"AutomaticallyAfterDays": self.auto_rotate_after_days},
-            "LastRotatedDate": None,
+            "LastRotatedDate": self.last_rotation_date,
             "LastChangedDate": self.last_changed_date,
             "LastAccessedDate": None,
+            "NextRotationDate": self.next_rotation_date,
             "DeletedDate": self.deleted_date,
             "Tags": self.tags,
             "VersionIdsToStages": version_id_to_stages,
@@ -535,6 +538,10 @@ class SecretsManagerBackend(BaseBackend):
                     )
                     raise InvalidParameterException(msg)
 
+                self.secrets[secret_id].next_rotation_date = int(time.time()) + (
+                    int(rotation_period) * 86400
+                )
+
         secret = self.secrets[secret_id]
 
         # The rotation function must end with the versions of the secret in
@@ -625,7 +632,7 @@ class SecretsManagerBackend(BaseBackend):
                 secret.versions[new_version_id], new_version_id
             )
             secret.versions[new_version_id]["version_stages"] = ["AWSCURRENT"]
-
+        self.secrets[secret_id].last_rotation_date = int(time.time())
         return secret.to_short_dict()
 
     def get_random_password(
diff --git a/moto/sns/models.py b/moto/sns/models.py
index a3cf2afd0..fd43eb181 100644
--- a/moto/sns/models.py
+++ b/moto/sns/models.py
@@ -139,36 +139,20 @@ class Topic(CloudFormationModel):
     @classmethod
     def update_from_cloudformation_json(  # type: ignore[misc]
         cls,
-        original_resource: Any,
+        original_resource: "Topic",
         new_resource_name: str,
         cloudformation_json: Any,
         account_id: str,
         region_name: str,
     ) -> "Topic":
-        cls.delete_from_cloudformation_json(
-            original_resource.name, cloudformation_json, account_id, region_name
-        )
+        original_resource.delete(account_id, region_name)
         return cls.create_from_cloudformation_json(
             new_resource_name, cloudformation_json, account_id, region_name
         )
 
-    @classmethod
-    def delete_from_cloudformation_json(  # type: ignore[misc]
-        cls,
-        resource_name: str,
-        cloudformation_json: Any,
-        account_id: str,
-        region_name: str,
-    ) -> None:
-        sns_backend = sns_backends[account_id][region_name]
-        properties = cloudformation_json["Properties"]
-
-        topic_name = properties.get(cls.cloudformation_name_type()) or resource_name
-        topic_arn = make_arn_for_topic(account_id, topic_name, sns_backend.region_name)
-        subscriptions, _ = sns_backend.list_subscriptions(topic_arn)
-        for subscription in subscriptions:
-            sns_backend.unsubscribe(subscription.arn)
-        sns_backend.delete_topic(topic_arn)
+    def delete(self, account_id: str, region_name: str) -> None:
+        sns_backend: SNSBackend = sns_backends[account_id][region_name]
+        sns_backend.delete_topic(self.arn)
 
     def _create_default_topic_policy(
         self, region_name: str, account_id: str, name: str


Test patch:
diff --git a/tests/test_autoscaling/test_autoscaling.py b/tests/test_autoscaling/test_autoscaling.py
index 293c66159..8e4fade10 100644
--- a/tests/test_autoscaling/test_autoscaling.py
+++ b/tests/test_autoscaling/test_autoscaling.py
@@ -771,6 +771,30 @@ def test_create_autoscaling_policy_with_policytype__targettrackingscaling():
                 "PredefinedMetricType": "ASGAverageNetworkIn",
             },
             "TargetValue": 1000000.0,
+            "CustomizedMetricSpecification": {
+                "Metrics": [
+                    {
+                        "Label": "Get ASGAverageCPUUtilization",
+                        "Id": "cpu",
+                        "MetricStat": {
+                            "Metric": {
+                                "MetricName": "CPUUtilization",
+                                "Namespace": "AWS/EC2",
+                                "Dimensions": [
+                                    {"Name": "AutoScalingGroupName", "Value": asg_name}
+                                ],
+                            },
+                            "Stat": "Average",
+                        },
+                        "ReturnData": False,
+                    },
+                    {
+                        "Label": "Calculate square cpu",
+                        "Id": "load",
+                        "Expression": "cpu^2",
+                    },
+                ],
+            },
         },
     )
 
@@ -786,6 +810,36 @@ def test_create_autoscaling_policy_with_policytype__targettrackingscaling():
             "PredefinedMetricSpecification": {
                 "PredefinedMetricType": "ASGAverageNetworkIn",
             },
+            "CustomizedMetricSpecification": {
+                "MetricName": "None",
+                "Namespace": "None",
+                "Dimensions": [],
+                "Statistic": "None",
+                "Metrics": [
+                    {
+                        "Label": "Get ASGAverageCPUUtilization",
+                        "Id": "cpu",
+                        "MetricStat": {
+                            "Metric": {
+                                "MetricName": "CPUUtilization",
+                                "Namespace": "AWS/EC2",
+                                "Dimensions": [
+                                    {"Name": "AutoScalingGroupName", "Value": asg_name}
+                                ],
+                            },
+                            "Stat": "Average",
+                            "Unit": "None",
+                        },
+                        "ReturnData": False,
+                    },
+                    {
+                        "Label": "Calculate square cpu",
+                        "Id": "load",
+                        "Expression": "cpu^2",
+                        "ReturnData": True,
+                    },
+                ],
+            },
             "TargetValue": 1000000.0,
         }
     )
diff --git a/tests/test_core/test_importorder.py b/tests/test_core/test_importorder.py
index 5f864f802..13a2e6e2a 100644
--- a/tests/test_core/test_importorder.py
+++ b/tests/test_core/test_importorder.py
@@ -59,7 +59,7 @@ def test_mock_works_with_resource_created_outside(
     m = mock_s3()
     m.start()
 
-    # So remind us to mock this client
+    # So remind us to mock this resource
     from moto.core import patch_resource
 
     patch_resource(outside_resource)
@@ -69,6 +69,28 @@ def test_mock_works_with_resource_created_outside(
     m.stop()
 
 
+def test_patch_can_be_called_on_a_mocked_client():
+    # start S3 after the mock, ensuring that the client contains our event-handler
+    m = mock_s3()
+    m.start()
+    client = boto3.client("s3", region_name="us-east-1")
+
+    from moto.core import patch_client
+
+    patch_client(client)
+    patch_client(client)
+
+    # At this point, our event-handler should only be registered once, by `mock_s3`
+    # `patch_client` should not re-register the event-handler
+    test_object = b"test_object_text"
+    client.create_bucket(Bucket="buck")
+    client.put_object(Bucket="buck", Key="to", Body=test_object)
+    got_object = client.get_object(Bucket="buck", Key="to")["Body"].read()
+    assert got_object == test_object
+
+    m.stop()
+
+
 def test_patch_client_does_not_work_for_random_parameters():
     from moto.core import patch_client
 
diff --git a/tests/test_ec2/test_route_tables.py b/tests/test_ec2/test_route_tables.py
index 51c37a646..70b70bd67 100644
--- a/tests/test_ec2/test_route_tables.py
+++ b/tests/test_ec2/test_route_tables.py
@@ -1041,6 +1041,62 @@ def test_create_route_with_unknown_egress_only_igw():
     err["Message"].should.equal("The eigw ID 'eoigw' does not exist")
 
 
+@mock_ec2
+def test_create_route_with_vpc_endpoint():
+    # Setup
+    _, ec2_client, route_table, vpc = setup_vpc()
+    dest_cidr = "0.0.0.0/0"
+    vpc_end_point = ec2_client.create_vpc_endpoint(
+        VpcId=vpc.id,
+        ServiceName="com.amazonaws.vpce.eu-central-1.vpce-svc-084fa044c50cb1290",
+        RouteTableIds=[route_table.id],
+        VpcEndpointType="GatewayLoadBalancer",
+    )
+    vpce_id = vpc_end_point["VpcEndpoint"]["VpcEndpointId"]
+
+    # Execute
+    ec2_client.create_route(
+        DestinationCidrBlock=dest_cidr,
+        VpcEndpointId=vpce_id,
+        RouteTableId=route_table.id,
+    )
+    rt = ec2_client.describe_route_tables()
+    new_route = rt["RouteTables"][-1]["Routes"][1]
+
+    # Verify
+    assert new_route["DestinationCidrBlock"] == dest_cidr
+    assert new_route["GatewayId"] == vpce_id
+
+
+@mock_ec2
+def test_create_route_with_invalid_vpc_endpoint():
+    # Setup
+    _, ec2_client, route_table, vpc = setup_vpc()
+    dest_cidr = "0.0.0.0/0"
+    vpc_end_point = ec2_client.create_vpc_endpoint(
+        VpcId=vpc.id,
+        ServiceName="com.amazonaws.us-east-1.s3",
+        RouteTableIds=[route_table.id],
+        VpcEndpointType="Gateway",
+    )
+    vpce_id = vpc_end_point["VpcEndpoint"]["VpcEndpointId"]
+
+    # Execute
+    with pytest.raises(ClientError) as ex:
+        ec2_client.create_route(
+            DestinationCidrBlock=dest_cidr,
+            VpcEndpointId=vpce_id,
+            RouteTableId=route_table.id,
+        )
+    # Verify
+    err = ex.value.response["Error"]
+    assert err["Code"] == "RouteNotSupported"
+    assert (
+        err["Message"] == f"Route table contains unsupported route target: {vpce_id}. "
+        "VPC Endpoints of this type cannot be used as route targets."
+    )
+
+
 @mock_ec2
 def test_associate_route_table_by_gateway():
     ec2 = boto3.client("ec2", region_name="us-west-1")
@@ -1087,3 +1143,17 @@ def test_associate_route_table_by_subnet():
     verify[0]["Associations"][0]["SubnetId"].should.equals(subnet_id)
     verify[0]["Associations"][0]["RouteTableAssociationId"].should.equal(assoc_id)
     verify[0]["Associations"][0].doesnt.have.key("GatewayId")
+
+
+def setup_vpc():
+    ec2_resource = boto3.resource("ec2", region_name="eu-central-1")
+    ec2_client = boto3.client("ec2", region_name="eu-central-1")
+
+    vpc = ec2_resource.create_vpc(CidrBlock="10.0.0.0/16")
+    ec2_resource.create_subnet(
+        VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-2a"
+    )
+
+    route_table = ec2_resource.create_route_table(VpcId=vpc.id)
+
+    return ec2_resource, ec2_client, route_table, vpc
diff --git a/tests/test_ecs/test_ecs_boto3.py b/tests/test_ecs/test_ecs_boto3.py
index d834162f0..5ef1374f6 100644
--- a/tests/test_ecs/test_ecs_boto3.py
+++ b/tests/test_ecs/test_ecs_boto3.py
@@ -589,7 +589,7 @@ def test_create_service():
     response["service"]["desiredCount"].should.equal(2)
     len(response["service"]["events"]).should.equal(0)
     len(response["service"]["loadBalancers"]).should.equal(0)
-    response["service"]["pendingCount"].should.equal(0)
+    response["service"]["pendingCount"].should.equal(2)
     response["service"]["runningCount"].should.equal(0)
     response["service"]["serviceArn"].should.equal(
         f"arn:aws:ecs:us-east-1:{ACCOUNT_ID}:service/test_ecs_cluster/test_ecs_service"
@@ -604,6 +604,77 @@ def test_create_service():
     response["service"]["platformVersion"].should.equal("2")
 
 
+@mock_ecs
+@mock_ec2
+def test_create_running_service():
+    if settings.TEST_SERVER_MODE:
+        raise SkipTest(
+            "Can't set environment variables in server mode for a single test"
+        )
+    running_service_count = 3
+    with mock.patch.dict(
+        os.environ, {"MOTO_ECS_SERVICE_RUNNING": str(running_service_count)}
+    ):
+        client = boto3.client("ecs", region_name="us-east-1")
+        ec2 = boto3.resource("ec2", region_name="us-east-1")
+        setup_ecs(client, ec2)
+
+        response = client.create_service(
+            cluster="test_ecs_cluster",
+            serviceName="test_ecs_service",
+            taskDefinition="test_ecs_task",
+            desiredCount=4,
+            platformVersion="2",
+        )
+
+        assert response["service"]["runningCount"] == running_service_count
+        assert response["service"]["pendingCount"] == 1
+
+
+@mock_ecs
+@mock_ec2
+def test_create_running_service_bad_env_var():
+    running_service_count = "ALSDHLHA;''"
+    with mock.patch.dict(
+        os.environ, {"MOTO_ECS_SERVICE_RUNNING": str(running_service_count)}
+    ):
+        client = boto3.client("ecs", region_name="us-east-1")
+        ec2 = boto3.resource("ec2", region_name="us-east-1")
+        setup_ecs(client, ec2)
+
+        response = client.create_service(
+            cluster="test_ecs_cluster",
+            serviceName="test_ecs_service",
+            taskDefinition="test_ecs_task",
+            desiredCount=2,
+            platformVersion="2",
+        )
+
+        assert response["service"]["runningCount"] == 0
+
+
+@mock_ecs
+@mock_ec2
+def test_create_running_service_negative_env_var():
+    running_service_count = "-20"
+    with mock.patch.dict(
+        os.environ, {"MOTO_ECS_SERVICE_RUNNING": str(running_service_count)}
+    ):
+        client = boto3.client("ecs", region_name="us-east-1")
+        ec2 = boto3.resource("ec2", region_name="us-east-1")
+        setup_ecs(client, ec2)
+
+        response = client.create_service(
+            cluster="test_ecs_cluster",
+            serviceName="test_ecs_service",
+            taskDefinition="test_ecs_task",
+            desiredCount=2,
+            platformVersion="2",
+        )
+
+        assert response["service"]["runningCount"] == 0
+
+
 @mock_ecs
 def test_create_service_errors():
     # given
@@ -680,7 +751,7 @@ def test_create_service_scheduling_strategy():
     response["service"]["desiredCount"].should.equal(2)
     len(response["service"]["events"]).should.equal(0)
     len(response["service"]["loadBalancers"]).should.equal(0)
-    response["service"]["pendingCount"].should.equal(0)
+    response["service"]["pendingCount"].should.equal(2)
     response["service"]["runningCount"].should.equal(0)
     response["service"]["serviceArn"].should.equal(
         f"arn:aws:ecs:us-east-1:{ACCOUNT_ID}:service/test_ecs_cluster/test_ecs_service"
@@ -870,7 +941,9 @@ def test_describe_services():
 @mock.patch.dict(os.environ, {"MOTO_ECS_NEW_ARN": "TrUe"})
 def test_describe_services_new_arn():
     if settings.TEST_SERVER_MODE:
-        raise SkipTest("Cant set environment variables in server mode")
+        raise SkipTest(
+            "Can't set environment variables in server mode for a single test"
+        )
     client = boto3.client("ecs", region_name="us-east-1")
     _ = client.create_cluster(clusterName="test_ecs_cluster")
     _ = client.register_task_definition(
@@ -1338,7 +1411,9 @@ def test_register_container_instance():
 @mock.patch.dict(os.environ, {"MOTO_ECS_NEW_ARN": "TrUe"})
 def test_register_container_instance_new_arn_format():
     if settings.TEST_SERVER_MODE:
-        raise SkipTest("Cant set environment variables in server mode")
+        raise SkipTest(
+            "Can't set environment variables in server mode for a single test"
+        )
     ecs_client = boto3.client("ecs", region_name="us-east-1")
     ec2 = boto3.resource("ec2", region_name="us-east-1")
 
@@ -1876,7 +1951,9 @@ def test_run_task_default_cluster():
 @mock.patch.dict(os.environ, {"MOTO_ECS_NEW_ARN": "TrUe"})
 def test_run_task_default_cluster_new_arn_format():
     if settings.TEST_SERVER_MODE:
-        raise SkipTest("Cant set environment variables in server mode")
+        raise SkipTest(
+            "Can't set environment variables in server mode for a single test"
+        )
     client = boto3.client("ecs", region_name="us-east-1")
     ec2 = boto3.resource("ec2", region_name="us-east-1")
 
@@ -2953,7 +3030,7 @@ def test_create_service_load_balancing():
         "test_container_name"
     )
     response["service"]["loadBalancers"][0]["containerPort"].should.equal(123)
-    response["service"]["pendingCount"].should.equal(0)
+    response["service"]["pendingCount"].should.equal(2)
     response["service"]["runningCount"].should.equal(0)
     response["service"]["serviceArn"].should.equal(
         f"arn:aws:ecs:us-east-1:{ACCOUNT_ID}:service/test_ecs_cluster/test_ecs_service"
diff --git a/tests/test_glue/test_glue.py b/tests/test_glue/test_glue.py
index b86e5ccce..a0014cdc1 100644
--- a/tests/test_glue/test_glue.py
+++ b/tests/test_glue/test_glue.py
@@ -789,3 +789,83 @@ def test_delete_trigger():
         client.get_trigger(Name=trigger_name)
 
     assert exc.value.response["Error"]["Code"] == "EntityNotFoundException"
+
+
+def create_test_session(client):
+    session_id = str(uuid4())
+    client.create_session(
+        Id=session_id,
+        Description="string",
+        Role="arn_of_a_testing_role",
+        Command={"Name": "string", "PythonVersion": "string"},
+        Timeout=123,
+        IdleTimeout=123,
+        DefaultArguments={"string": "string"},
+        Connections={
+            "Connections": [
+                "string",
+            ]
+        },
+        MaxCapacity=123.0,
+        NumberOfWorkers=123,
+        WorkerType="Standard",
+        SecurityConfiguration="string",
+        GlueVersion="string",
+        Tags={"string": "string"},
+        RequestOrigin="string",
+    )
+    return session_id
+
+
+@mock_glue
+def test_create_session():
+    client = create_glue_client()
+    session_id = create_test_session(client)
+
+    resp = client.get_session(Id=session_id)
+    assert resp["Session"]["Id"] == session_id
+
+
+@mock_glue
+def test_get_session():
+    client = create_glue_client()
+    session_id = create_test_session(client)
+
+    resp = client.get_session(Id=session_id)
+    assert resp["Session"]["Id"] == session_id
+
+
+@mock_glue
+def test_list_sessions():
+    client = create_glue_client()
+    session_id = create_test_session(client)
+
+    resp = client.list_sessions()
+    assert session_id in resp["Ids"]
+
+
+@mock_glue
+def test_delete_session():
+    client = create_glue_client()
+    session_id = create_test_session(client)
+
+    resp = client.delete_session(Id=session_id)
+    assert resp["Id"] == session_id
+
+    resp = client.list_sessions()
+    assert session_id not in resp["Ids"]
+
+
+@mock_glue
+def test_stop_session():
+    client = create_glue_client()
+    session_id = create_test_session(client)
+
+    resp = client.get_session(Id=session_id)
+    assert resp["Session"]["Status"] == "READY"
+
+    resp = client.stop_session(Id=session_id)
+    assert resp["Id"] == session_id
+
+    resp = client.get_session(Id=session_id)
+    assert resp["Session"]["Status"] == "STOPPING"
diff --git a/tests/test_s3/test_s3_copyobject.py b/tests/test_s3/test_s3_copyobject.py
index 5a54a0537..701e8df58 100644
--- a/tests/test_s3/test_s3_copyobject.py
+++ b/tests/test_s3/test_s3_copyobject.py
@@ -46,7 +46,7 @@ def test_copy_key_boto3_with_sha256_checksum():
     key_name = "key"
     new_key = "new_key"
     bucket = "foobar"
-    expected_hash = "YWIzZDA3ZjMxNjljY2JkMGVkNmM0YjQ1ZGUyMTUxOWY5ZjkzOGM3MmQyNDEyNDk5OGFhYjk0OWNlODNiYjUxYg=="
+    expected_hash = "qz0H8xacy9DtbEtF3iFRn5+TjHLSQSSZiquUnOg7tRs="
 
     s3.create_bucket(Bucket=bucket)
     key = s3.Object("foobar", key_name)
@@ -730,3 +730,112 @@ def test_copy_object_in_place_with_bucket_encryption():
         Key=key,
     )
     assert response["ServerSideEncryption"] == "AES256"
+
+
+@mock_s3
+@pytest.mark.parametrize(
+    "algorithm",
+    ["CRC32", "SHA1", "SHA256"],
+)
+def test_copy_key_boto3_with_both_sha256_checksum(algorithm):
+    # This test will validate that moto S3 checksum calculations are correct
+    # We first create an object with a Checksum calculated by boto, by specifying ChecksumAlgorithm="SHA256"
+    # we then retrieve the right checksum from this request
+    # we copy the object while requesting moto to recalculate the checksum for that key
+    # we verify that both checksums are equal
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    source_key = "source-key"
+    dest_key = "dest-key"
+    bucket = "foobar"
+    body = b"checksum-test"
+    client.create_bucket(Bucket=bucket)
+
+    checksum_key = f"Checksum{algorithm}"
+
+    resp = client.put_object(
+        Bucket=bucket,
+        Key=source_key,
+        Body=body,
+        ChecksumAlgorithm=algorithm,
+    )
+    assert checksum_key in resp
+    checksum_by_boto = resp[checksum_key]
+
+    resp = client.copy_object(
+        Bucket=bucket,
+        CopySource=f"{bucket}/{source_key}",
+        Key=dest_key,
+        ChecksumAlgorithm=algorithm,
+    )
+
+    assert checksum_key in resp["CopyObjectResult"]
+    assert resp["CopyObjectResult"][checksum_key] == checksum_by_boto
+
+
+@mock_s3
+@pytest.mark.parametrize(
+    "algorithm, checksum",
+    [
+        ("CRC32", "lVk/nw=="),
+        ("SHA1", "jbXkHAsXUrubtL3dqDQ4w+7WXc0="),
+        ("SHA256", "1YQo81vx2VFUl0q5ccWISq8AkSBQQ0WO80S82TmfdIQ="),
+    ],
+)
+def test_copy_object_calculates_checksum(algorithm, checksum):
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    source_key = "source-key"
+    dest_key = "dest-key"
+    bucket = "foobar"
+    body = b"test-checksum"
+    client.create_bucket(Bucket=bucket)
+
+    checksum_key = f"Checksum{algorithm}"
+
+    resp = client.put_object(
+        Bucket=bucket,
+        Key=source_key,
+        Body=body,
+    )
+    assert checksum_key not in resp
+
+    resp = client.copy_object(
+        Bucket=bucket,
+        CopySource=f"{bucket}/{source_key}",
+        Key=dest_key,
+        ChecksumAlgorithm=algorithm,
+    )
+
+    assert checksum_key in resp["CopyObjectResult"]
+    assert resp["CopyObjectResult"][checksum_key] == checksum
+
+
+@mock_s3
+def test_copy_object_keeps_checksum():
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    source_key = "source-key"
+    dest_key = "dest-key"
+    bucket = "foobar"
+    body = b"test-checksum"
+    expected_checksum = "1YQo81vx2VFUl0q5ccWISq8AkSBQQ0WO80S82TmfdIQ="
+    client.create_bucket(Bucket=bucket)
+
+    # put an object with a checksum
+    resp = client.put_object(
+        Bucket=bucket,
+        Key=source_key,
+        Body=body,
+        ChecksumAlgorithm="SHA256",
+    )
+    assert "ChecksumSHA256" in resp
+    assert resp["ChecksumSHA256"] == expected_checksum
+
+    # do not specify the checksum
+    resp = client.copy_object(
+        Bucket=bucket,
+        CopySource=f"{bucket}/{source_key}",
+        Key=dest_key,
+    )
+
+    # assert that it kept the checksum from the source key
+    assert "ChecksumSHA256" in resp["CopyObjectResult"]
+    assert resp["CopyObjectResult"]["ChecksumSHA256"] == expected_checksum
diff --git a/tests/test_s3/test_s3_utils.py b/tests/test_s3/test_s3_utils.py
index a5bdd7344..5edd5a24b 100644
--- a/tests/test_s3/test_s3_utils.py
+++ b/tests/test_s3/test_s3_utils.py
@@ -124,24 +124,22 @@ def test_undo_clean_key_name(key, expected):
 
 
 def test_checksum_sha256():
-    checksum = b"ODdkMTQ5Y2I0MjRjMDM4NzY1NmYyMTFkMjU4OWZiNWIxZTE2MjI5OTIxMzA5ZTk4NTg4NDE5Y2NjYThhNzM2Mg=="
+    checksum = b"h9FJy0JMA4dlbyEdJYn7Wx4WIpkhMJ6YWIQZzMqKc2I="
     compute_checksum(b"somedata", "SHA256").should.equal(checksum)
     # Unknown algorithms fallback to SHA256 for now
     compute_checksum(b"somedata", algorithm="unknown").should.equal(checksum)
 
 
 def test_checksum_sha1():
-    compute_checksum(b"somedata", "SHA1").should.equal(
-        b"ZWZhYTMxMWFlNDQ4YTczNzRjMTIyMDYxYmZlZDk1MmQ5NDBlOWUzNw=="
-    )
+    compute_checksum(b"somedata", "SHA1").should.equal(b"76oxGuRIpzdMEiBhv+2VLZQOnjc=")
 
 
 def test_checksum_crc32():
-    compute_checksum(b"somedata", "CRC32").should.equal(b"MTM5MzM0Mzk1Mg==")
+    compute_checksum(b"somedata", "CRC32").should.equal(b"Uwy90A==")
 
 
 def test_checksum_crc32c():
-    compute_checksum(b"somedata", "CRC32C").should.equal(b"MTM5MzM0Mzk1Mg==")
+    compute_checksum(b"somedata", "CRC32C").should.equal(b"Uwy90A==")
 
 
 def test_cors_utils():
diff --git a/tests/test_secretsmanager/test_secretsmanager.py b/tests/test_secretsmanager/test_secretsmanager.py
index 6eee8b762..f2f78ec22 100644
--- a/tests/test_secretsmanager/test_secretsmanager.py
+++ b/tests/test_secretsmanager/test_secretsmanager.py
@@ -711,21 +711,41 @@ def test_cancel_rotate_secret():
 
 @mock_secretsmanager
 def test_rotate_secret():
-    conn = boto3.client("secretsmanager", region_name="us-west-2")
-    conn.create_secret(
-        Name=DEFAULT_SECRET_NAME, SecretString="foosecret", Description="foodescription"
-    )
-
-    rotated_secret = conn.rotate_secret(SecretId=DEFAULT_SECRET_NAME)
-
-    assert rotated_secret
-    assert rotated_secret["ARN"] != ""  # Test arn not empty
-    assert rotated_secret["Name"] == DEFAULT_SECRET_NAME
-    assert rotated_secret["VersionId"] != ""
-
-    describe_secret = conn.describe_secret(SecretId=DEFAULT_SECRET_NAME)
+    # Setup
+    frozen_time = datetime(2023, 5, 20, 10, 20, 30, tzinfo=tzlocal())
+    rotate_after_days = 10
+    with freeze_time(frozen_time):
+        conn = boto3.client("secretsmanager", region_name="us-west-2")
+        conn.create_secret(
+            Name=DEFAULT_SECRET_NAME,
+            SecretString="foosecret",
+            Description="foodescription",
+        )
 
-    assert describe_secret["Description"] == "foodescription"
+        # Execute
+        rotated_secret = conn.rotate_secret(
+            SecretId=DEFAULT_SECRET_NAME,
+            RotationRules={"AutomaticallyAfterDays": rotate_after_days},
+        )
+        describe_secret = conn.describe_secret(SecretId=DEFAULT_SECRET_NAME)
+
+        # Verify
+        assert rotated_secret
+        assert rotated_secret["ARN"] != ""  # Test arn not empty
+        assert rotated_secret["Name"] == DEFAULT_SECRET_NAME
+        assert rotated_secret["VersionId"] != ""
+        assert describe_secret["Description"] == "foodescription"
+        assert "NextRotationDate" in describe_secret
+        assert "LastRotatedDate" in describe_secret
+
+        # can't do freeze time tests in servermode tests
+        if settings.TEST_SERVER_MODE:
+            return
+
+        assert describe_secret["LastChangedDate"] == frozen_time
+        assert describe_secret["NextRotationDate"] == frozen_time + timedelta(
+            days=rotate_after_days
+        )
 
 
 @mock_secretsmanager
diff --git a/tests/test_sns/test_sns_cloudformation.py b/tests/test_sns/test_sns_cloudformation.py
index 17cd25e05..df8e1eb03 100644
--- a/tests/test_sns/test_sns_cloudformation.py
+++ b/tests/test_sns/test_sns_cloudformation.py
@@ -1,6 +1,7 @@
 import boto3
 import json
 import sure  # noqa # pylint: disable=unused-import
+import pytest
 
 from moto import mock_cloudformation, mock_sns
 
@@ -8,24 +9,7 @@ from moto import mock_cloudformation, mock_sns
 @mock_cloudformation
 @mock_sns
 def test_sns_topic():
-    dummy_template = {
-        "AWSTemplateFormatVersion": "2010-09-09",
-        "Resources": {
-            "MySNSTopic": {
-                "Type": "AWS::SNS::Topic",
-                "Properties": {
-                    "Subscription": [
-                        {"Endpoint": "https://example.com", "Protocol": "https"}
-                    ],
-                    "TopicName": "my_topics",
-                },
-            }
-        },
-        "Outputs": {
-            "topic_name": {"Value": {"Fn::GetAtt": ["MySNSTopic", "TopicName"]}},
-            "topic_arn": {"Value": {"Ref": "MySNSTopic"}},
-        },
-    }
+    dummy_template = get_template(with_properties=True)
     template_json = json.dumps(dummy_template)
     cf = boto3.client("cloudformation", region_name="us-west-1")
     cf.create_stack(StackName="test_stack", TemplateBody=template_json)
@@ -56,24 +40,7 @@ def test_sns_topic():
 @mock_cloudformation
 @mock_sns
 def test_sns_update_topic():
-    dummy_template = {
-        "AWSTemplateFormatVersion": "2010-09-09",
-        "Resources": {
-            "MySNSTopic": {
-                "Type": "AWS::SNS::Topic",
-                "Properties": {
-                    "Subscription": [
-                        {"Endpoint": "https://example.com", "Protocol": "https"}
-                    ],
-                    "TopicName": "my_topics",
-                },
-            }
-        },
-        "Outputs": {
-            "topic_name": {"Value": {"Fn::GetAtt": ["MySNSTopic", "TopicName"]}},
-            "topic_arn": {"Value": {"Ref": "MySNSTopic"}},
-        },
-    }
+    dummy_template = get_template(with_properties=True)
     sns_template_json = json.dumps(dummy_template)
     cf = boto3.client("cloudformation", region_name="us-west-1")
     cf.create_stack(StackName="test_stack", TemplateBody=sns_template_json)
@@ -104,25 +71,9 @@ def test_sns_update_topic():
 
 @mock_cloudformation
 @mock_sns
-def test_sns_update_remove_topic():
-    dummy_template = {
-        "AWSTemplateFormatVersion": "2010-09-09",
-        "Resources": {
-            "MySNSTopic": {
-                "Type": "AWS::SNS::Topic",
-                "Properties": {
-                    "Subscription": [
-                        {"Endpoint": "https://example.com", "Protocol": "https"}
-                    ],
-                    "TopicName": "my_topics",
-                },
-            }
-        },
-        "Outputs": {
-            "topic_name": {"Value": {"Fn::GetAtt": ["MySNSTopic", "TopicName"]}},
-            "topic_arn": {"Value": {"Ref": "MySNSTopic"}},
-        },
-    }
+@pytest.mark.parametrize("with_properties", [True, False])
+def test_sns_update_remove_topic(with_properties):
+    dummy_template = get_template(with_properties)
     sns_template_json = json.dumps(dummy_template)
     cf = boto3.client("cloudformation", region_name="us-west-1")
     cf.create_stack(StackName="test_stack", TemplateBody=sns_template_json)
@@ -142,26 +93,9 @@ def test_sns_update_remove_topic():
 
 @mock_cloudformation
 @mock_sns
-def test_sns_delete_topic():
-    dummy_template = {
-        "AWSTemplateFormatVersion": "2010-09-09",
-        "Resources": {
-            "MySNSTopic": {
-                "Type": "AWS::SNS::Topic",
-                "Properties": {
-                    "Subscription": [
-                        {"Endpoint": "https://example.com", "Protocol": "https"}
-                    ],
-                    "TopicName": "my_topics",
-                },
-            }
-        },
-        "Outputs": {
-            "topic_name": {"Value": {"Fn::GetAtt": ["MySNSTopic", "TopicName"]}},
-            "topic_arn": {"Value": {"Ref": "MySNSTopic"}},
-        },
-    }
-    sns_template_json = json.dumps(dummy_template)
+@pytest.mark.parametrize("with_properties", [True, False])
+def test_sns_delete_topic(with_properties):
+    sns_template_json = json.dumps(get_template(with_properties))
     cf = boto3.client("cloudformation", region_name="us-west-1")
     cf.create_stack(StackName="test_stack", TemplateBody=sns_template_json)
 
@@ -173,3 +107,20 @@ def test_sns_delete_topic():
 
     topics = sns.list_topics()["Topics"]
     topics.should.have.length_of(0)
+
+
+def get_template(with_properties):
+    dummy_template = {
+        "AWSTemplateFormatVersion": "2010-09-09",
+        "Resources": {"MySNSTopic": {"Type": "AWS::SNS::Topic"}},
+        "Outputs": {
+            "topic_name": {"Value": {"Fn::GetAtt": ["MySNSTopic", "TopicName"]}},
+            "topic_arn": {"Value": {"Ref": "MySNSTopic"}},
+        },
+    }
+    if with_properties:
+        dummy_template["Resources"]["MySNSTopic"]["Properties"] = {
+            "Subscription": [{"Endpoint": "https://example.com", "Protocol": "https"}],
+            "TopicName": "my_topics",
+        }
+    return dummy_template