Repository: getmoto/moto

Issue:
Add list, create, describe and delete operations for activities
The following operations are not implemented for activities:

1.  `create_activity`
2. `delete_activity`
3. `describe_activity`
4. `list_activities`

Hints from issue comments:

Solution patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index 81fdfb9b9..56a6b50cb 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -3185,7 +3185,7 @@
 - [ ] modify_public_ip_dns_name_options
 - [ ] modify_reserved_instances
 - [ ] modify_route_server
-- [ ] modify_security_group_rules
+- [X] modify_security_group_rules
 - [ ] modify_snapshot_attribute
 - [ ] modify_snapshot_tier
 - [X] modify_spot_fleet_request
@@ -9536,14 +9536,14 @@
 <details>
 <summary>54% implemented</summary>
 
-- [ ] create_activity
+- [X] create_activity
 - [X] create_state_machine
 - [ ] create_state_machine_alias
-- [ ] delete_activity
+- [X] delete_activity
 - [X] delete_state_machine
 - [ ] delete_state_machine_alias
 - [ ] delete_state_machine_version
-- [ ] describe_activity
+- [X] describe_activity
 - [X] describe_execution
 - [X] describe_map_run
 - [X] describe_state_machine
@@ -9551,7 +9551,7 @@
 - [X] describe_state_machine_for_execution
 - [ ] get_activity_task
 - [X] get_execution_history
-- [ ] list_activities
+- [X] list_activities
 - [X] list_executions
 - [X] list_map_runs
 - [ ] list_state_machine_aliases
@@ -10365,4 +10365,4 @@
 - workmailmessageflow
 - workspaces-thin-client
 - xray
-</details>
\ No newline at end of file
+</details>
diff --git a/docs/docs/services/ec2.rst b/docs/docs/services/ec2.rst
index 90dfe366d..d1cefa531 100644
--- a/docs/docs/services/ec2.rst
+++ b/docs/docs/services/ec2.rst
@@ -628,7 +628,7 @@ ec2
 - [ ] modify_public_ip_dns_name_options
 - [ ] modify_reserved_instances
 - [ ] modify_route_server
-- [ ] modify_security_group_rules
+- [X] modify_security_group_rules
 - [ ] modify_snapshot_attribute
 - [ ] modify_snapshot_tier
 - [X] modify_spot_fleet_request
diff --git a/moto/batch/models.py b/moto/batch/models.py
index 7eeac5fb6..ffa34d066 100644
--- a/moto/batch/models.py
+++ b/moto/batch/models.py
@@ -1219,6 +1219,7 @@ class BatchBackend(BaseBackend):
                 "type": environment.env_type,
                 "status": "VALID",
                 "statusReason": "Compute environment is available",
+                "tags": self.list_tags_for_resource(arn),
             }
             if environment.env_type == "MANAGED":
                 json_part["computeResources"] = environment.compute_resources
@@ -1234,6 +1235,7 @@ class BatchBackend(BaseBackend):
         state: str,
         compute_resources: Dict[str, Any],
         service_role: str,
+        tags: Optional[Dict[str, str]] = None,
     ) -> ComputeEnvironment:
         # Validate
         if COMPUTE_ENVIRONMENT_NAME_REGEX.match(compute_environment_name) is None:
@@ -1289,6 +1291,9 @@ class BatchBackend(BaseBackend):
         )
         self._compute_environments[new_comp_env.arn] = new_comp_env
 
+        if tags:
+            self.tag_resource(new_comp_env.arn, tags)
+
         # Ok by this point, everything is legit, so if its Managed then start some instances
         if _type == "MANAGED" and "FARGATE" not in compute_resources["type"]:
             cpus = int(
diff --git a/moto/batch/responses.py b/moto/batch/responses.py
index 6d1c1eb93..db59fed0d 100644
--- a/moto/batch/responses.py
+++ b/moto/batch/responses.py
@@ -30,6 +30,7 @@ class BatchResponse(BaseResponse):
         service_role = self._get_param("serviceRole")
         state = self._get_param("state")
         _type = self._get_param("type")
+        tags = self._get_param("tags")
 
         env = self.batch_backend.create_compute_environment(
             compute_environment_name=compute_env_name,
@@ -37,6 +38,7 @@ class BatchResponse(BaseResponse):
             state=state,
             compute_resources=compute_resource,
             service_role=service_role,
+            tags=tags,
         )
 
         result = {"computeEnvironmentArn": env.arn, "computeEnvironmentName": env.name}
diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py
index 4ae7eaaf7..5d3d848be 100644
--- a/moto/ec2/exceptions.py
+++ b/moto/ec2/exceptions.py
@@ -828,6 +828,22 @@ class InvalidParameter(EC2ClientError):
         )
 
 
+class InvalidParameterValue(EC2ClientError):
+    def __init__(self, message: str):
+        super().__init__(
+            "InvalidParameterValue",
+            message,
+        )
+
+
+class MissingParameter(EC2ClientError):
+    def __init__(self, message: str):
+        super().__init__(
+            "MissingParameter",
+            message,
+        )
+
+
 class InvalidSubnetCidrBlockAssociationID(EC2ClientError):
     def __init__(self, association_id: str):
         super().__init__(
diff --git a/moto/ec2/models/security_groups.py b/moto/ec2/models/security_groups.py
index 5bb11d39f..cf3f02d3a 100644
--- a/moto/ec2/models/security_groups.py
+++ b/moto/ec2/models/security_groups.py
@@ -10,11 +10,14 @@ from moto.core.utils import aws_api_matches
 from ..exceptions import (
     InvalidCIDRSubnetError,
     InvalidGroupIdMalformedError,
+    InvalidParameterCombination,
+    InvalidParameterValue,
     InvalidPermissionDuplicateError,
     InvalidPermissionNotFoundError,
     InvalidSecurityGroupDuplicateError,
     InvalidSecurityGroupNotFoundError,
     InvalidSecurityGroupRuleIdNotFoundError,
+    MissingParameter,
     MissingParameterError,
     MotoNotImplementedError,
     RulesPerSecurityGroupLimitExceededError,
@@ -524,6 +527,13 @@ class SecurityGroup(TaggedEC2Resource, CloudFormationModel):
             return self.id
         raise UnformattedGetAttTemplateException()
 
+    def get_rule(self, rule_id: str) -> Optional[SecurityRule]:
+        """Retrieve a security group rule by its ID."""
+        for rule in list(itertools.chain(self.egress_rules, self.ingress_rules)):
+            if rule.id == rule_id:
+                return rule
+        return None
+
     def add_ingress_rule(self, rule: SecurityRule) -> None:
         if rule in self.ingress_rules:
             raise InvalidPermissionDuplicateError()
@@ -785,6 +795,69 @@ class SecurityGroupBackend:
                 tags=tags,
             )
 
+    def modify_security_group_rules(
+        self,
+        group_id: str,
+        rules: dict[str, dict[str, Any]],
+    ) -> None:
+        group = self.get_security_group_by_name_or_id(group_id)
+        if group is None:
+            raise InvalidSecurityGroupNotFoundError(group_id)
+
+        for rule_id, new_rule in rules.items():
+            cidr_ipv4 = new_rule.get("CidrIpv4")
+            cidr_ipv6 = new_rule.get("CidrIpv6")
+            prefix_list_id = new_rule.get("PrefixListId")
+            reference_group_id = new_rule.get("ReferenceGroupId")
+
+            set_param_count = sum(
+                item is not None
+                for item in [cidr_ipv4, cidr_ipv6, prefix_list_id, reference_group_id]
+            )
+            if set_param_count > 1:
+                raise InvalidParameterCombination(
+                    "Only one of cidrIp, cidrIpv6, prefixListId, or referencedGroupId can be specified"
+                )
+            if set_param_count < 1:
+                raise MissingParameter(
+                    "The request must contain exactly one of: cidrIp, cidrIpv6, prefixListId, or referencedGroupId"
+                )
+
+            existing_rule = group.get_rule(rule_id)
+            if existing_rule is None:
+                raise InvalidSecurityGroupRuleIdNotFoundError(rule_id)
+
+            ip_protocol = new_rule.get("IpProtocol")
+            if ip_protocol is None:
+                raise InvalidParameterValue(
+                    "Invalid value 'null' for protocol. VPC security group rules must specify protocols explicitly."
+                )
+
+            from_port = new_rule.get("FromPort")
+            to_port = new_rule.get("ToPort")
+            if from_port is None or to_port is None:
+                raise InvalidParameterValue(
+                    "Invalid value for portRange. Must specify both from and to ports with TCP/UDP."
+                )
+
+            if cidr_ipv4 and "CidrIpv6" in existing_rule.ip_range:
+                raise InvalidParameterValue(
+                    f"Invalid rule type for security group rule '{rule_id}'. You may not specify CidrIpv6 for an existing IPv4 CIDR rule."
+                )
+
+            existing_rule.ip_protocol = ip_protocol
+            existing_rule.from_port = from_port
+            existing_rule.to_port = to_port
+
+            # TODO: Handle cidr_ipv6, prefix_list_id and reference_group_id
+
+            description = new_rule.get("Description")
+            if description is not None:
+                existing_rule.ip_range["Description"] = description
+
+            if cidr_ipv4 is not None:
+                existing_rule.ip_range["CidrIp"] = cidr_ipv4
+
     def authorize_security_group_ingress(
         self,
         group_name_or_id: str,
diff --git a/moto/ec2/responses/security_groups.py b/moto/ec2/responses/security_groups.py
index 1b05adc6a..717603525 100644
--- a/moto/ec2/responses/security_groups.py
+++ b/moto/ec2/responses/security_groups.py
@@ -221,6 +221,19 @@ class SecurityGroups(EC2BaseResponse):
             self.ec2_backend.revoke_security_group_ingress(**args)
         return REVOKE_SECURITY_GROUP_INGRESS_RESPONSE
 
+    def modify_security_group_rules(self) -> str:
+        self.error_on_dryrun()
+
+        rules = {}
+        security_group_rules_param = self._get_params()["SecurityGroupRule"]
+        for idx, sgr in security_group_rules_param.items():
+            rules[sgr["SecurityGroupRuleId"]] = sgr["SecurityGroupRule"]
+
+        group_id = self._get_param("GroupId")
+        self.ec2_backend.modify_security_group_rules(group_id, rules)
+
+        return MODIFY_SECURITY_GROUP_RULES_RESPONSE
+
     def update_security_group_rule_descriptions_ingress(self) -> str:
         for args in self._process_rules_from_querystring():
             # we don't need this parameter to revoke
@@ -577,6 +590,13 @@ REVOKE_SECURITY_GROUP_EGRESS_RESPONSE = """<RevokeSecurityGroupEgressResponse xm
   <return>true</return>
 </RevokeSecurityGroupEgressResponse>"""
 
+MODIFY_SECURITY_GROUP_RULES_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
+<ModifySecurityGroupRulesResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">
+  <requestId>04d8f970-9213-41fa-81d2-50825EXAMPLE</requestId>
+  <return>true</return>
+</ModifySecurityGroupRulesResponse>
+"""
+
 UPDATE_SECURITY_GROUP_RULE_DESCRIPTIONS_INGRESS = """<UpdateSecurityGroupRuleDescriptionsIngressResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
   <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
   <return>true</return>
diff --git a/moto/resourcegroupstaggingapi/models.py b/moto/resourcegroupstaggingapi/models.py
index 3c38c0ec9..136fb0dea 100644
--- a/moto/resourcegroupstaggingapi/models.py
+++ b/moto/resourcegroupstaggingapi/models.py
@@ -759,7 +759,15 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
         # Step Functions
         if not resource_type_filters or "states:stateMachine" in resource_type_filters:
             for state_machine in self.stepfunctions_backend.state_machines:
-                tags = format_tag_keys(state_machine.tags, ["key", "value"])
+                tags = format_tag_keys(
+                    state_machine.backend.get_tags_list_for_state_machine(
+                        state_machine.arn
+                    ),
+                    [
+                        state_machine.backend.tagger.key_name,
+                        state_machine.backend.tagger.value_name,
+                    ],
+                )
                 if not tags or not tag_filter(tags):
                     continue
                 yield {"ResourceARN": state_machine.arn, "Tags": tags}
diff --git a/moto/stepfunctions/exceptions.py b/moto/stepfunctions/exceptions.py
index 8467daf33..d6c4aac7d 100644
--- a/moto/stepfunctions/exceptions.py
+++ b/moto/stepfunctions/exceptions.py
@@ -60,3 +60,18 @@ class NameTooLongException(ValidationException):
             f"1 validation error detected: Value '{name}' at 'name' "
             "failed to satisfy constraint: Member must have length less than or equal to 80"
         )
+
+
+class InvalidEncryptionConfiguration(AWSError):
+    TYPE = "InvalidEncryptionConfiguration"
+    STATUS = 400
+
+
+class ActivityDoesNotExist(AWSError):
+    TYPE = "ActivityDoesNotExist"
+    STATUS = 400
+
+
+class ActivityAlreadyExists(AWSError):
+    TYPE = "ActivityAlreadyExists"
+    STATUS = 400
diff --git a/moto/stepfunctions/models.py b/moto/stepfunctions/models.py
index fd91fcf7c..ef7b5e9e2 100644
--- a/moto/stepfunctions/models.py
+++ b/moto/stepfunctions/models.py
@@ -11,18 +11,23 @@ from moto.core.common_models import CloudFormationModel
 from moto.core.utils import iso_8601_datetime_with_milliseconds
 from moto.moto_api._internal import mock_random
 from moto.utilities.paginator import paginate
+from moto.utilities.tagging_service import TaggingService
 from moto.utilities.utils import ARN_PARTITION_REGEX, get_partition
 
 from .exceptions import (
+    ActivityAlreadyExists,
+    ActivityDoesNotExist,
     ExecutionAlreadyExists,
     ExecutionDoesNotExist,
     InvalidArn,
+    InvalidEncryptionConfiguration,
     InvalidExecutionInput,
     InvalidName,
     NameTooLongException,
     ResourceNotFound,
     StateMachineDoesNotExist,
 )
+from .parser.api import EncryptionType
 from .utils import PAGINATION_MODEL, api_to_cfn_tags, cfn_to_api_tags
 
 
@@ -81,7 +86,7 @@ class StateMachine(StateMachineInstance, CloudFormationModel):
         name: str,
         definition: str,
         roleArn: str,
-        tags: Optional[List[Dict[str, str]]] = None,
+        backend: "StepFunctionBackend",
         encryptionConfiguration: Optional[Dict[str, Any]] = None,
         loggingConfiguration: Optional[Dict[str, Any]] = None,
         tracingConfiguration: Optional[Dict[str, Any]] = None,
@@ -96,13 +101,10 @@ class StateMachine(StateMachineInstance, CloudFormationModel):
             loggingConfiguration=loggingConfiguration,
             tracingConfiguration=tracingConfiguration,
         )
-        self.tags: List[Dict[str, str]] = []
-        if tags:
-            self.add_tags(tags)
-
         self.latest_version_number = 0
         self.versions: Dict[int, StateMachineVersion] = {}
         self.latest_version: Optional[StateMachineVersion] = None
+        self.backend = backend
 
     def publish(self, description: Optional[str]) -> None:
         new_version_number = self.latest_version_number + 1
@@ -165,27 +167,6 @@ class StateMachine(StateMachineInstance, CloudFormationModel):
                 setattr(self, key, value)
         self.update_date = iso_8601_datetime_with_milliseconds()
 
-    def add_tags(self, tags: List[Dict[str, str]]) -> List[Dict[str, str]]:
-        merged_tags = []
-        for tag in self.tags:
-            replacement_index = next(
-                (index for (index, d) in enumerate(tags) if d["key"] == tag["key"]),
-                None,
-            )
-            if replacement_index is not None:
-                replacement = tags.pop(replacement_index)
-                merged_tags.append(replacement)
-            else:
-                merged_tags.append(tag)
-        for tag in tags:
-            merged_tags.append(tag)
-        self.tags = merged_tags
-        return self.tags
-
-    def remove_tags(self, tag_keys: List[str]) -> List[Dict[str, str]]:
-        self.tags = [tag_set for tag_set in self.tags if tag_set["key"] not in tag_keys]
-        return self.tags
-
     @property
     def physical_resource_id(self) -> str:
         return self.arn
@@ -231,7 +212,9 @@ class StateMachine(StateMachineInstance, CloudFormationModel):
         elif attribute_name == "StateMachineName":
             return self.name
         elif attribute_name == "Tags":
-            return api_to_cfn_tags(self.tags)
+            return api_to_cfn_tags(
+                self.backend.get_tags_list_for_state_machine(self.arn)
+            )
 
         raise UnformattedGetAttTemplateException()
 
@@ -304,7 +287,7 @@ class StateMachine(StateMachineInstance, CloudFormationModel):
             state_machine = sf_backend.update_state_machine(
                 original_resource.arn, definition=definition, role_arn=role_arn
             )
-            state_machine.add_tags(tags)
+            sf_backend.tag_resource(state_machine.arn, tags)
             return state_machine
 
 
@@ -448,6 +431,21 @@ class Execution:
         self.stop_date = datetime.now()
 
 
+class Activity:
+    def __init__(
+        self,
+        arn: str,
+        name: str,
+        encryption_configuration: Optional[Dict[str, Any]] = None,
+    ):
+        self.arn = arn
+        self.name = name
+        self.encryption_configuration = encryption_configuration
+
+        self.creation_date = iso_8601_datetime_with_milliseconds()
+        self.update_date = self.creation_date
+
+
 class StepFunctionBackend(BaseBackend):
     """
     Configure Moto to explicitly parse and execute the StateMachine:
@@ -576,10 +574,19 @@ class StepFunctionBackend(BaseBackend):
         ARN_PARTITION_REGEX
         + r":states:[-0-9a-zA-Z]+:(?P<account_id>[0-9]{12}):execution:.+"
     )
+    accepted_activity_arn_format = re.compile(
+        ARN_PARTITION_REGEX
+        + r":states:[-0-9a-zA-Z]+:(?P<account_id>[0-9]{12}):activity:.+"
+    )
 
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
+        self.tagger = TaggingService(
+            tag_name="tags", key_name="key", value_name="value"
+        )
+
         self.state_machines: List[StateMachine] = []
+        self.activities: Dict[str, Activity] = {}
         self._account_id = None
 
     def create_state_machine(
@@ -605,13 +612,17 @@ class StepFunctionBackend(BaseBackend):
                 name,
                 definition,
                 roleArn,
-                tags,
+                self,
                 encryptionConfiguration,
                 loggingConfiguration,
                 tracingConfiguration,
             )
             if publish:
                 state_machine.publish(description=version_description)
+
+            if tags:
+                self.tag_resource(arn, tags)
+
             self.state_machines.append(state_machine)
             return state_machine
 
@@ -734,26 +745,17 @@ class StepFunctionBackend(BaseBackend):
                     return sm
         raise ResourceNotFound(execution_arn)
 
-    def list_tags_for_resource(self, arn: str) -> List[Dict[str, str]]:
-        try:
-            state_machine = self.describe_state_machine(arn)
-            return state_machine.tags or []
-        except StateMachineDoesNotExist:
-            return []
+    def list_tags_for_resource(self, arn: str) -> Dict[str, List[Dict[str, str]]]:
+        return self.tagger.list_tags_for_resource(arn)
 
     def tag_resource(self, resource_arn: str, tags: List[Dict[str, str]]) -> None:
-        try:
-            state_machine = self.describe_state_machine(resource_arn)
-            state_machine.add_tags(tags)
-        except StateMachineDoesNotExist:
-            raise ResourceNotFound(resource_arn)
+        self.tagger.tag_resource(resource_arn, tags)
 
     def untag_resource(self, resource_arn: str, tag_keys: List[str]) -> None:
-        try:
-            state_machine = self.describe_state_machine(resource_arn)
-            state_machine.remove_tags(tag_keys)
-        except StateMachineDoesNotExist:
-            raise ResourceNotFound(resource_arn)
+        self.tagger.untag_resource_using_names(resource_arn, tag_keys)
+
+    def get_tags_list_for_state_machine(self, arn: str) -> List[Dict[str, str]]:
+        return self.list_tags_for_resource(arn)[self.tagger.tag_name]
 
     def send_task_failure(self, task_token: str, error: Optional[str] = None) -> None:
         pass
@@ -810,11 +812,35 @@ class StepFunctionBackend(BaseBackend):
             invalid_msg="Execution Does Not Exist: '" + execution_arn + "'",
         )
 
+    def _validate_activity_arn(self, activity_arn: str) -> None:
+        self._validate_arn(
+            arn=activity_arn,
+            regex=self.accepted_activity_arn_format,
+            invalid_msg="Invalid Activity Arn: '" + activity_arn + "'",
+        )
+
     def _validate_arn(self, arn: str, regex: Pattern[str], invalid_msg: str) -> None:
         match = regex.match(arn)
         if not arn or not match:
             raise InvalidArn(invalid_msg)
 
+    def _validate_encryption_configuration(
+        self, encryption_configuration: Dict[str, Any]
+    ) -> None:
+        encryption_type = encryption_configuration.get("type")
+        if not encryption_type:
+            raise InvalidEncryptionConfiguration(
+                "Invalid Encryption Configuration: 'type' is required."
+            )
+
+        if (
+            encryption_type == EncryptionType.CUSTOMER_MANAGED_KMS_KEY
+            and not encryption_configuration.get("kmsKeyId")
+        ):
+            raise InvalidEncryptionConfiguration(
+                "Invalid Encryption Configuration: 'kmsKeyId' is required when 'type' is 'CUSTOMER_MANAGED_KMS_KEY'"
+            )
+
     def _get_state_machine_for_execution(self, execution_arn: str) -> StateMachine:
         state_machine_name = execution_arn.split(":")[6]
         state_machine_arn = next(
@@ -828,5 +854,49 @@ class StepFunctionBackend(BaseBackend):
             )
         return self.describe_state_machine(state_machine_arn)
 
+    def create_activity(
+        self,
+        name: str,
+        tags: Optional[List[Dict[str, str]]] = None,
+        encryption_configuration: Optional[Dict[str, Any]] = None,
+    ) -> Activity:
+        self._validate_name(name)
+
+        if encryption_configuration is not None:
+            self._validate_encryption_configuration(encryption_configuration)
+
+        arn = f"arn:{get_partition(self.region_name)}:states:{self.region_name}:{self.account_id}:activity:{name}"
+
+        if arn in self.activities:
+            raise ActivityAlreadyExists("Activity already exists.")
+
+        activity = Activity(
+            arn=arn,
+            name=name,
+            encryption_configuration=encryption_configuration,
+        )
+        self.activities[arn] = activity
+
+        if tags:
+            self.tag_resource(arn, tags)
+
+        return activity
+
+    def describe_activity(self, activity_arn: str) -> Activity:
+        self._validate_activity_arn(activity_arn)
+        if activity_arn in self.activities:
+            return self.activities[activity_arn]
+        else:
+            raise ActivityDoesNotExist(activity_arn)
+
+    def delete_activity(self, activity_arn: str) -> None:
+        self._validate_activity_arn(activity_arn)
+        if activity_arn in self.activities:
+            del self.activities[activity_arn]
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_activities(self) -> List[Activity]:
+        return sorted(self.activities.values(), key=lambda x: x.creation_date)
+
 
 stepfunctions_backends = BackendDict(StepFunctionBackend, "stepfunctions")
diff --git a/moto/stepfunctions/responses.py b/moto/stepfunctions/responses.py
index effaca54e..c07d26e6b 100644
--- a/moto/stepfunctions/responses.py
+++ b/moto/stepfunctions/responses.py
@@ -140,8 +140,7 @@ class StepFunctionResponse(BaseResponse):
 
     def list_tags_for_resource(self) -> TYPE_RESPONSE:
         arn = self._get_param("resourceArn")
-        tags = self.stepfunction_backend.list_tags_for_resource(arn)
-        response = {"tags": tags}
+        response = self.stepfunction_backend.list_tags_for_resource(arn)
         return 200, {}, json.dumps(response)
 
     def tag_resource(self) -> TYPE_RESPONSE:
@@ -297,3 +296,53 @@ class StepFunctionResponse(BaseResponse):
             tolerated_failure_percentage=tolerated_failure_percentage,
         )
         return 200, {}, "{}"
+
+    def create_activity(self) -> TYPE_RESPONSE:
+        name = self._get_param("name")
+        tags = self._get_param("tags")
+        encryption_configuration = self._get_param("encryptionConfiguration")
+        activity = self.stepfunction_backend.create_activity(
+            name=name, tags=tags, encryption_configuration=encryption_configuration
+        )
+        response = {
+            "creationDate": activity.creation_date,
+            "activityArn": activity.arn,
+        }
+        return 200, {}, json.dumps(response)
+
+    def describe_activity(self) -> TYPE_RESPONSE:
+        activity_arn = self._get_param("activityArn")
+        activity = self.stepfunction_backend.describe_activity(activity_arn)
+
+        response = {
+            "activityArn": activity.arn,
+            "name": activity.name,
+            "creationDate": activity.creation_date,
+            "encryptionConfiguration": activity.encryption_configuration,
+        }
+        return 200, {}, json.dumps(response)
+
+    def delete_activity(self) -> TYPE_RESPONSE:
+        activity_arn = self._get_param("activityArn")
+        self.stepfunction_backend.delete_activity(activity_arn)
+        return 200, {}, "{}"
+
+    def list_activities(self) -> TYPE_RESPONSE:
+        max_results = self._get_param("maxResults")
+        next_token = self._get_param("nextToken")
+        results, next_token = self.stepfunction_backend.list_activities(
+            max_results=max_results, next_token=next_token
+        )
+
+        activities = [
+            {
+                "activityArn": activity.arn,
+                "name": activity.name,
+                "creationDate": activity.creation_date,
+            }
+            for activity in results
+        ]
+        response = {"activities": activities}
+        if next_token:
+            response["nextToken"] = next_token
+        return 200, {}, json.dumps(response)
diff --git a/moto/stepfunctions/utils.py b/moto/stepfunctions/utils.py
index bf3e3e89d..522727e8e 100644
--- a/moto/stepfunctions/utils.py
+++ b/moto/stepfunctions/utils.py
@@ -13,6 +13,12 @@ PAGINATION_MODEL = {
         "limit_default": 100,
         "unique_attribute": ["creation_date", "arn"],
     },
+    "list_activities": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "unique_attribute": ["creation_date", "arn"],
+    },
 }
 
 


Test patch:
diff --git a/tests/test_batch/test_batch_compute_envs.py b/tests/test_batch/test_batch_compute_envs.py
index 37b6f43b9..455d9fb3a 100644
--- a/tests/test_batch/test_batch_compute_envs.py
+++ b/tests/test_batch/test_batch_compute_envs.py
@@ -60,6 +60,62 @@ def test_create_managed_compute_environment(use_arn_for_instance_role):
     assert our_env["ecsClusterArn"] in all_clusters
 
 
+@mock_aws
+@pytest.mark.parametrize("use_arn_for_instance_role", [True, False])
+def test_create_managed_compute_environment_with_tags(use_arn_for_instance_role):
+    ec2_client, iam_client, ecs_client, _, batch_client = _get_clients()
+    _, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)
+
+    instance_role = (
+        iam_arn.replace("role", "instance-profile")
+        if use_arn_for_instance_role
+        else iam_arn.split("/")[-1]
+    )
+    compute_name = str(uuid4())
+    resp = batch_client.create_compute_environment(
+        computeEnvironmentName=compute_name,
+        type="MANAGED",
+        state="ENABLED",
+        computeResources={
+            "type": "EC2",
+            "minvCpus": 5,
+            "maxvCpus": 10,
+            "desiredvCpus": 5,
+            "instanceTypes": ["t2.small", "t2.medium"],
+            "imageId": "some_image_id",
+            "subnets": [subnet_id],
+            "securityGroupIds": [sg_id],
+            "ec2KeyPair": "string",
+            "instanceRole": instance_role,
+            "tags": {"string": "string"},
+            "bidPercentage": 123,
+            "spotIamFleetRole": "string",
+        },
+        serviceRole=iam_arn,
+        tags={"name": "compute-environment-1"},
+    )
+    assert "computeEnvironmentArn" in resp
+    assert resp["computeEnvironmentName"] == compute_name
+
+    our_env = batch_client.describe_compute_environments(
+        computeEnvironments=[compute_name]
+    )["computeEnvironments"][0]
+
+    # Given a t2.medium is 2 vcpu and t2.small is 1, therefore 2 mediums and 1 small should be created
+    if not settings.TEST_SERVER_MODE:
+        # Can't verify this in ServerMode, as other tests may have created instances
+        resp = ec2_client.describe_instances()
+        assert "Reservations" in resp
+        assert len(resp["Reservations"]) == 3
+
+    # Should have created 1 ECS cluster
+    all_clusters = ecs_client.list_clusters()["clusterArns"]
+    assert our_env["ecsClusterArn"] in all_clusters
+
+    assert "tags" in our_env
+    assert len(our_env["tags"]) == 1
+
+
 @mock_aws
 def test_create_managed_compute_environment_with_instance_family():
     """
diff --git a/tests/test_ec2/test_security_groups.py b/tests/test_ec2/test_security_groups.py
index 63bc9b74d..9a979c062 100644
--- a/tests/test_ec2/test_security_groups.py
+++ b/tests/test_ec2/test_security_groups.py
@@ -768,6 +768,256 @@ def test_create_and_describe_security_grp_rule():
     _verify_egress_rule(rules[0])
 
 
+@ec2_aws_verified(create_vpc=True, create_sg=True)
+@pytest.mark.aws_verified
+def test_modify_security_group_rules(ec2_client=None, vpc_id=None, sg_id=None):
+    client = ec2_client
+    group_id = sg_id
+
+    ip_permissions = [
+        {
+            "IpProtocol": "tcp",
+            "FromPort": 30900,
+            "ToPort": 31000,
+            "IpRanges": [
+                {"CidrIp": "1.2.3.4/32", "Description": "original rule"},
+            ],
+        }
+    ]
+    rule_id = client.authorize_security_group_ingress(
+        GroupId=group_id, IpPermissions=ip_permissions
+    )["SecurityGroupRules"][0]["SecurityGroupRuleId"]
+
+    # Ensure bad group ID raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId="sg-00000000",
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": rule_id,
+                    "SecurityGroupRule": {
+                        "IpProtocol": "udp",
+                        "FromPort": 27010,
+                        "ToPort": 27011,
+                        "CidrIpv4": "2.3.4.5/32",
+                        "Description": "modification that should fail",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "InvalidGroup.NotFound"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "The security group 'sg-00000000' does not exist"
+    )
+
+    # Ensure bad rule ID raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId=group_id,
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": "sgr-00000000",
+                    "SecurityGroupRule": {
+                        "IpProtocol": "udp",
+                        "FromPort": 27010,
+                        "ToPort": 27011,
+                        "CidrIpv4": "2.3.4.5/32",
+                        "Description": "modification that should fail",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "InvalidSecurityGroupRuleId.NotFound"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "The security group rule ID 'sgr-00000000' does not exist"
+    )
+
+    # Ensure missing protocol raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId=group_id,
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": rule_id,
+                    "SecurityGroupRule": {
+                        "FromPort": 27010,
+                        "ToPort": 27011,
+                        "CidrIpv4": "2.3.4.5/32",
+                        "Description": "modification that should fail",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "InvalidParameterValue"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "Invalid value 'null' for protocol. VPC security group rules must specify protocols explicitly."
+    )
+
+    # Ensure missing port range raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId=group_id,
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": rule_id,
+                    "SecurityGroupRule": {
+                        "IpProtocol": "udp",
+                        "ToPort": 27011,
+                        "CidrIpv4": "2.3.4.5/32",
+                        "Description": "modification that should fail",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "InvalidParameterValue"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "Invalid value for portRange. Must specify both from and to ports with TCP/UDP."
+    )
+
+    # Ensure missing cidr raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId=group_id,
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": rule_id,
+                    "SecurityGroupRule": {
+                        "IpProtocol": "udp",
+                        "FromPort": 27010,
+                        "ToPort": 27011,
+                        "Description": "modification that should fail",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "MissingParameter"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "The request must contain exactly one of: cidrIp, cidrIpv6, prefixListId, or referencedGroupId"
+    )
+
+    # Ensure multiple cidr raises
+    with pytest.raises(ClientError) as ex:
+        client.modify_security_group_rules(
+            GroupId=group_id,
+            SecurityGroupRules=[
+                {
+                    "SecurityGroupRuleId": rule_id,
+                    "SecurityGroupRule": {
+                        "IpProtocol": "udp",
+                        "FromPort": 27010,
+                        "ToPort": 27011,
+                        "CidrIpv4": "2.3.4.5/32",
+                        "CidrIpv6": "2001:db8::/32",
+                        "Description": "modification that should fail ",
+                    },
+                }
+            ],
+        )
+    assert ex.value.response["Error"]["Code"] == "InvalidParameterCombination"
+    assert (
+        ex.value.response["Error"]["Message"]
+        == "Only one of cidrIp, cidrIpv6, prefixListId, or referencedGroupId can be specified"
+    )
+
+    # Happy path: Ensure all attributes can be modified
+    response = client.modify_security_group_rules(
+        GroupId=group_id,
+        SecurityGroupRules=[
+            {
+                "SecurityGroupRuleId": rule_id,
+                "SecurityGroupRule": {
+                    "IpProtocol": "udp",
+                    "FromPort": 26000,
+                    "ToPort": 26500,
+                    "CidrIpv4": "2.3.4.5/32",
+                    "Description": "first modification",
+                },
+            }
+        ],
+    )
+    assert response["Return"] is True
+
+    response = client.describe_security_group_rules(
+        Filters=[
+            {"Name": "group-id", "Values": [group_id]},
+            {"Name": "security-group-rule-id", "Values": [rule_id]},
+        ]
+    )
+    assert len(response["SecurityGroupRules"]) == 1
+    assert response["SecurityGroupRules"][0]["IpProtocol"] == "udp"
+    assert response["SecurityGroupRules"][0]["FromPort"] == 26000
+    assert response["SecurityGroupRules"][0]["ToPort"] == 26500
+    assert response["SecurityGroupRules"][0]["CidrIpv4"] == "2.3.4.5/32"
+    assert response["SecurityGroupRules"][0]["Description"] == "first modification"
+    assert "CidrIpv6" not in response["SecurityGroupRules"][0]
+
+    # Add another rule
+    rule_id_2 = client.authorize_security_group_egress(
+        GroupId=group_id, IpPermissions=ip_permissions
+    )["SecurityGroupRules"][0]["SecurityGroupRuleId"]
+
+    # Ensure multiple sg rules can be modified together
+    response = client.modify_security_group_rules(
+        GroupId=group_id,
+        SecurityGroupRules=[
+            {
+                "SecurityGroupRuleId": rule_id,
+                "SecurityGroupRule": {
+                    "IpProtocol": "tcp",
+                    "FromPort": 10,
+                    "ToPort": 11,
+                    "CidrIpv4": "1.1.1.1/32",
+                    "Description": "second modification",
+                },
+            },
+            {
+                "SecurityGroupRuleId": rule_id_2,
+                "SecurityGroupRule": {
+                    "IpProtocol": "udp",
+                    "FromPort": 20,
+                    "ToPort": 21,
+                    "CidrIpv4": "2.2.2.2/32",
+                    "Description": "first modification",
+                },
+            },
+        ],
+    )
+    assert response["Return"] is True
+
+    # Ensure first rule
+    response = client.describe_security_group_rules(
+        Filters=[
+            {"Name": "group-id", "Values": [group_id]},
+            {"Name": "security-group-rule-id", "Values": [rule_id]},
+        ]
+    )
+    assert len(response["SecurityGroupRules"]) == 1
+    assert response["SecurityGroupRules"][0]["IpProtocol"] == "tcp"
+    assert response["SecurityGroupRules"][0]["FromPort"] == 10
+    assert response["SecurityGroupRules"][0]["ToPort"] == 11
+    assert response["SecurityGroupRules"][0]["CidrIpv4"] == "1.1.1.1/32"
+    assert response["SecurityGroupRules"][0]["Description"] == "second modification"
+
+    # Ensure second rule
+    response = client.describe_security_group_rules(
+        Filters=[
+            {"Name": "group-id", "Values": [group_id]},
+            {"Name": "security-group-rule-id", "Values": [rule_id_2]},
+        ]
+    )
+    assert len(response["SecurityGroupRules"]) == 1
+    assert response["SecurityGroupRules"][0]["IpProtocol"] == "udp"
+    assert response["SecurityGroupRules"][0]["FromPort"] == 20
+    assert response["SecurityGroupRules"][0]["ToPort"] == 21
+    assert response["SecurityGroupRules"][0]["CidrIpv4"] == "2.2.2.2/32"
+    assert response["SecurityGroupRules"][0]["Description"] == "first modification"
+
+
 @mock_aws
 @pytest.mark.parametrize("use_vpc", [True, False], ids=["Use VPC", "Without VPC"])
 def test_sec_group_rule_limit(use_vpc):
diff --git a/tests/test_stepfunctions/test_stepfunctions.py b/tests/test_stepfunctions/test_stepfunctions.py
index dccfa013e..84eec095d 100644
--- a/tests/test_stepfunctions/test_stepfunctions.py
+++ b/tests/test_stepfunctions/test_stepfunctions.py
@@ -7,7 +7,7 @@ from uuid import uuid4
 
 import boto3
 import pytest
-from botocore.exceptions import ClientError
+from botocore.exceptions import ClientError, ParamValidationError
 from dateutil.tz import tzutc
 
 from moto import mock_aws
@@ -390,26 +390,6 @@ def test_state_machine_can_deleted_nonexisting_machine():
     assert len(sm_list["stateMachines"]) == 0
 
 
-@mock_aws
-def test_state_machine_tagging_non_existent_resource_fails():
-    client = boto3.client("stepfunctions", region_name=region)
-    non_existent_arn = f"arn:aws:states:{region}:{ACCOUNT_ID}:stateMachine:non-existent"
-    with pytest.raises(ClientError) as ex:
-        client.tag_resource(resourceArn=non_existent_arn, tags=[])
-    assert ex.value.response["Error"]["Code"] == "ResourceNotFound"
-    assert non_existent_arn in ex.value.response["Error"]["Message"]
-
-
-@mock_aws
-def test_state_machine_untagging_non_existent_resource_fails():
-    client = boto3.client("stepfunctions", region_name=region)
-    non_existent_arn = f"arn:aws:states:{region}:{ACCOUNT_ID}:stateMachine:non-existent"
-    with pytest.raises(ClientError) as ex:
-        client.untag_resource(resourceArn=non_existent_arn, tagKeys=[])
-    assert ex.value.response["Error"]["Code"] == "ResourceNotFound"
-    assert non_existent_arn in ex.value.response["Error"]["Message"]
-
-
 @mock_aws
 def test_state_machine_tagging():
     client = boto3.client("stepfunctions", region_name=region)
@@ -1087,3 +1067,388 @@ def test_version_is_only_available_when_published():
 
 def _get_default_role():
     return "arn:aws:iam::" + ACCOUNT_ID + ":role/unknown_sf_role"
+
+
+@mock_aws
+def test_create_activity():
+    client = boto3.client("stepfunctions", region_name=region)
+    response = client.create_activity(
+        name="test-activity",
+        tags=[{"key": "activity-name", "value": "test-activity"}],
+        encryptionConfiguration={
+            "kmsKeyId": "test-id",
+            "kmsDataKeyReusePeriodSeconds": 123,
+            "type": "CUSTOMER_MANAGED_KMS_KEY",
+        },
+    )
+
+    assert "creationDate" in response
+    assert "activityArn" in response
+
+
+@mock_aws
+def test_create_activity_with_invalid_name():
+    client = boto3.client("stepfunctions", region_name=region)
+
+    invalid_names = [
+        "with space",
+        "with<bracket",
+        "with>bracket",
+        "with{bracket",
+        "with}bracket",
+        "with[bracket",
+        "with]bracket",
+        "with?wildcard",
+        "with*wildcard",
+        'special"char',
+        "special#char",
+        "special%char",
+        "special\\char",
+        "special^char",
+        "special|char",
+        "special~char",
+        "special`char",
+        "special$char",
+        "special&char",
+        "special,char",
+        "special;char",
+        "special:char",
+        "special/char",
+        "uni\u0000code",
+        "uni\u0001code",
+        "uni\u0002code",
+        "uni\u0003code",
+        "uni\u0004code",
+        "uni\u0005code",
+        "uni\u0006code",
+        "uni\u0007code",
+        "uni\u0008code",
+        "uni\u0009code",
+        "uni\u000acode",
+        "uni\u000bcode",
+        "uni\u000ccode",
+        "uni\u000dcode",
+        "uni\u000ecode",
+        "uni\u000fcode",
+        "uni\u0010code",
+        "uni\u0011code",
+        "uni\u0012code",
+        "uni\u0013code",
+        "uni\u0014code",
+        "uni\u0015code",
+        "uni\u0016code",
+        "uni\u0017code",
+        "uni\u0018code",
+        "uni\u0019code",
+        "uni\u001acode",
+        "uni\u001bcode",
+        "uni\u001ccode",
+        "uni\u001dcode",
+        "uni\u001ecode",
+        "uni\u001fcode",
+        "uni\u007fcode",
+        "uni\u0080code",
+        "uni\u0081code",
+        "uni\u0082code",
+        "uni\u0083code",
+        "uni\u0084code",
+        "uni\u0085code",
+        "uni\u0086code",
+        "uni\u0087code",
+        "uni\u0088code",
+        "uni\u0089code",
+        "uni\u008acode",
+        "uni\u008bcode",
+        "uni\u008ccode",
+        "uni\u008dcode",
+        "uni\u008ecode",
+        "uni\u008fcode",
+        "uni\u0090code",
+        "uni\u0091code",
+        "uni\u0092code",
+        "uni\u0093code",
+        "uni\u0094code",
+        "uni\u0095code",
+        "uni\u0096code",
+        "uni\u0097code",
+        "uni\u0098code",
+        "uni\u0099code",
+        "uni\u009acode",
+        "uni\u009bcode",
+        "uni\u009ccode",
+        "uni\u009dcode",
+        "uni\u009ecode",
+        "uni\u009fcode",
+    ]
+
+    for invalid_name in invalid_names:
+        with pytest.raises(ClientError) as exc:
+            client.create_activity(
+                name=invalid_name,
+                tags=[{"key": "activity-name", "value": "test-activity"}],
+            )
+
+        assert exc.value.response["Error"]["Code"] == "InvalidName"
+
+    # Validate name too long error.
+    with pytest.raises(ClientError) as exc:
+        client.create_activity(
+            name="test" * 25,  # 100 characters long.
+            tags=[{"key": "activity-name", "value": "test-activity"}],
+        )
+
+    assert exc.value.response["Error"]["Code"] == "ValidationException"
+    assert (
+        "Member must have length less than or equal to 80"
+        in exc.value.response["Error"]["Message"]
+    )
+
+
+@mock_aws
+def test_create_activity_with_invalid_encryption_configuration():
+    client = boto3.client("stepfunctions", region_name=region)
+
+    # Missing `type` in encryptionConfiguration
+    with pytest.raises(ParamValidationError) as exc:
+        client.create_activity(
+            name="test-activity",
+            tags=[{"key": "activity-name", "value": "test-activity"}],
+            encryptionConfiguration={
+                "kmsKeyId": "test-id",
+                "kmsDataKeyReusePeriodSeconds": 123,
+            },
+        )
+
+    # `kmsKeyId` missing when `type` is `CUSTOMER_MANAGED_KMS_KEY`
+    with pytest.raises(ClientError) as exc:
+        client.create_activity(
+            name="test-activity",
+            tags=[{"key": "activity-name", "value": "test-activity"}],
+            encryptionConfiguration={
+                "kmsDataKeyReusePeriodSeconds": 123,
+                "type": "CUSTOMER_MANAGED_KMS_KEY",
+            },
+        )
+    assert exc.value.response["Error"]["Code"] == "InvalidEncryptionConfiguration"
+
+
+@mock_aws
+def test_create_activity_with_duplicate_name():
+    client = boto3.client("stepfunctions", region_name=region)
+    client.create_activity(
+        name="test-activity",
+        tags=[{"key": "activity-name", "value": "test-activity"}],
+        encryptionConfiguration={
+            "kmsKeyId": "test-id",
+            "kmsDataKeyReusePeriodSeconds": 123,
+            "type": "CUSTOMER_MANAGED_KMS_KEY",
+        },
+    )
+
+    # Validate error if user tries to create activity with the existing name
+    with pytest.raises(ClientError) as exc:
+        client.create_activity(
+            name="test-activity",
+            tags=[{"key": "activity-name", "value": "test-activity"}],
+            encryptionConfiguration={
+                "kmsKeyId": "test-id",
+                "kmsDataKeyReusePeriodSeconds": 123,
+                "type": "CUSTOMER_MANAGED_KMS_KEY",
+            },
+        )
+
+    assert exc.value.response["Error"]["Code"] == "ActivityAlreadyExists"
+
+
+@mock_aws
+def test_describe_activity():
+    client = boto3.client("stepfunctions", region_name=region)
+    activity = client.create_activity(
+        name="test-activity",
+        tags=[{"key": "activity-name", "value": "test-activity"}],
+        encryptionConfiguration={
+            "kmsKeyId": "test-id",
+            "kmsDataKeyReusePeriodSeconds": 123,
+            "type": "CUSTOMER_MANAGED_KMS_KEY",
+        },
+    )
+
+    response = client.describe_activity(activityArn=activity["activityArn"])
+    assert response["name"] == "test-activity"
+    assert response["activityArn"] == activity["activityArn"]
+    assert response["creationDate"] == activity["creationDate"]
+    assert response["encryptionConfiguration"]["kmsKeyId"] == "test-id"
+    assert response["encryptionConfiguration"]["kmsDataKeyReusePeriodSeconds"] == 123
+    assert response["encryptionConfiguration"]["type"] == "CUSTOMER_MANAGED_KMS_KEY"
+
+
+@mock_aws
+def test_delete_activity():
+    client = boto3.client("stepfunctions", region_name=region)
+    activity = client.create_activity(
+        name="test-activity",
+        tags=[{"key": "activity-name", "value": "test-activity"}],
+        encryptionConfiguration={
+            "kmsKeyId": "test-id",
+            "kmsDataKeyReusePeriodSeconds": 123,
+            "type": "CUSTOMER_MANAGED_KMS_KEY",
+        },
+    )
+
+    response = client.describe_activity(activityArn=activity["activityArn"])
+    assert response["name"] == "test-activity"
+
+    client.delete_activity(activityArn=activity["activityArn"])
+
+    # Make sure activity is deleted.
+    with pytest.raises(ClientError) as exc:
+        client.describe_activity(activityArn=activity["activityArn"])
+
+    assert exc.value.response["Error"]["Code"] == "ActivityDoesNotExist"
+
+
+@mock_aws
+def test_list_activities_returns_empty_list_by_default():
+    client = boto3.client("stepfunctions", region_name=region)
+    #
+    activities = client.list_activities()
+    assert activities["activities"] == []
+
+
+@mock_aws
+def test_list_activities_returns_created_activities():
+    client = boto3.client("stepfunctions", region_name=region)
+    activity1 = client.create_activity(
+        name="test-activity-1",
+        tags=[{"key": "activity-name", "value": "test-activity"}],
+    )
+    activity2 = client.create_activity(
+        name="test-activity-2",
+        tags=[{"key": "activity-name", "value": "test-activity"}],
+    )
+    activities = client.list_activities()
+
+    assert activities["ResponseMetadata"]["HTTPStatusCode"] == 200
+    assert len(activities["activities"]) == 2
+    assert isinstance(activities["activities"][0]["creationDate"], datetime)
+    assert activities["activities"][0]["creationDate"] == activity1["creationDate"]
+    assert activities["activities"][0]["name"] == "test-activity-1"
+    assert activities["activities"][0]["activityArn"] == activity1["activityArn"]
+    assert isinstance(activities["activities"][1]["creationDate"], datetime)
+    assert activities["activities"][1]["creationDate"] == activity2["creationDate"]
+    assert activities["activities"][1]["name"] == "test-activity-2"
+    assert activities["activities"][1]["activityArn"] == activity2["activityArn"]
+
+
+@mock_aws
+def test_list_activities_pagination():
+    client = boto3.client("stepfunctions", region_name=region)
+    for i in range(25):
+        activity_name = f"test-activity-{i}"
+        client.create_activity(
+            name=activity_name,
+            tags=[{"key": "activity-name", "value": f"test-activity-{i}"}],
+        )
+
+    resp = client.list_activities()
+    assert "nextToken" not in resp
+    assert len(resp["activities"]) == 25
+
+    paginator = client.get_paginator("list_activities")
+    page_iterator = paginator.paginate(maxResults=5)
+    page_list = list(page_iterator)
+    for page in page_list:
+        assert len(page["activities"]) == 5
+    assert "24" in page_list[-1]["activities"][-1]["name"]
+
+
+@mock_aws
+def test_activity_tagging():
+    client = boto3.client("stepfunctions", region_name=region)
+    # Test tags are added on resource creation
+    tags = [
+        {"key": "tag_key1", "value": "tag_value1"},
+        {"key": "tag_key2", "value": "tag_value2"},
+    ]
+    activity = client.create_activity(
+        name="test-with-tags",
+        tags=tags,
+    )
+    resp = client.list_tags_for_resource(resourceArn=activity["activityArn"])
+    assert resp["tags"] == tags
+
+    # Test tags are added after creation with tag_resource
+    activity = client.create_activity(
+        name="test-activity",
+    )
+    client.tag_resource(resourceArn=activity["activityArn"], tags=tags)
+    resp = client.list_tags_for_resource(resourceArn=activity["activityArn"])
+    assert resp["tags"] == tags
+
+    tags_update = [
+        {"key": "tag_key1", "value": "tag_value1_new"},
+        {"key": "tag_key3", "value": "tag_value3"},
+    ]
+    client.tag_resource(resourceArn=activity["activityArn"], tags=tags_update)
+    resp = client.list_tags_for_resource(resourceArn=activity["activityArn"])
+    tags_expected = [
+        tags_update[0],
+        tags[1],
+        tags_update[1],
+    ]
+    assert resp["tags"] == tags_expected
+
+
+@mock_aws
+def test_activity_untagging():
+    client = boto3.client("stepfunctions", region_name=region)
+    tags = [
+        {"key": "tag_key1", "value": "tag_value1"},
+        {"key": "tag_key2", "value": "tag_value2"},
+        {"key": "tag_key3", "value": "tag_value3"},
+    ]
+    activity = client.create_activity(
+        name="test-activity",
+        tags=tags,
+    )
+    resp = client.list_tags_for_resource(resourceArn=activity["activityArn"])
+    assert resp["tags"] == tags
+    tags_to_delete = ["tag_key1", "tag_key2"]
+    client.untag_resource(resourceArn=activity["activityArn"], tagKeys=tags_to_delete)
+    resp = client.list_tags_for_resource(resourceArn=activity["activityArn"])
+    expected_tags = [tag for tag in tags if tag["key"] not in tags_to_delete]
+    assert resp["tags"] == expected_tags
+
+
+@mock_aws
+def test_activity_list_tags_for_created_activity():
+    client = boto3.client("stepfunctions", region_name=region)
+    #
+    activity = client.create_activity(
+        name="test-activity",
+        tags=[{"key": "tag_key", "value": "tag_value"}],
+    )
+    response = client.list_tags_for_resource(resourceArn=activity["activityArn"])
+    tags = response["tags"]
+    assert len(tags) == 1
+    assert tags[0] == {"key": "tag_key", "value": "tag_value"}
+
+
+@mock_aws
+def test_activity_list_tags_for_activity_without_tags():
+    client = boto3.client("stepfunctions", region_name=region)
+    #
+    activity = client.create_activity(name="test-activity")
+    response = client.list_tags_for_resource(resourceArn=activity["activityArn"])
+    tags = response["tags"]
+    assert len(tags) == 0
+
+
+@mock_aws
+def test_activity_list_tags_for_nonexisting_activity():
+    client = boto3.client("stepfunctions", region_name=region)
+    #
+    non_existing_activity = f"arn:aws:states:{region}:{ACCOUNT_ID}:activity:unknown"
+    response = client.list_tags_for_resource(resourceArn=non_existing_activity)
+    tags = response["tags"]
+    assert len(tags) == 0