StepFunctions: add sfn stateMachine to resourcegrouptaggingapi (#8215)

moto/resourcegroupstaggingapi/models.py

@@ -25,6 +25,7 @@
 from moto.sns.models import SNSBackend, sns_backends
 from moto.sqs.models import SQSBackend, sqs_backends
 from moto.ssm.models import SimpleSystemManagerBackend, ssm_backends
+from moto.stepfunctions.models import StepFunctionBackend, stepfunctions_backends
 from moto.utilities.tagging_service import TaggingService
 from moto.utilities.utils import get_partition
 from moto.workspaces.models import WorkSpacesBackend, workspaces_backends
@@ -121,6 +122,10 @@ def ssm_backend(self) -> SimpleSystemManagerBackend:
     def sqs_backend(self) -> SQSBackend:
         return sqs_backends[self.account_id][self.region_name]
 
+    @property
+    def stepfunctions_backend(self) -> StepFunctionBackend:
+        return stepfunctions_backends[self.account_id][self.region_name]
+
     @property
     def backup_backend(self) -> BackupBackend:
         return backup_backends[self.account_id][self.region_name]
@@ -525,6 +530,13 @@ def format_tag_keys(
                     "ResourceARN": f"arn:{get_partition(self.region_name)}:ssm:{self.region_name}:{self.account_id}:document/{doc_name}",
                     "Tags": tags,
                 }
+        # 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"])
+                if not tags or not tag_filter(tags):
+                    continue
+                yield {"ResourceARN": state_machine.arn, "Tags": tags}
 
         # Workspaces
         if self.workspaces_backend and (
@@ -845,10 +857,9 @@ def get_resources(
         new_token = str(mock_random.uuid4())
         self._pages[new_token] = {"gen": generator, "misc": next_item}
 
-        # Token used up, might as well bin now, if you call it again your an idiot
+        # Token used up, might as well bin now, if you call it again you're an idiot
         if pagination_token:
             del self._pages[pagination_token]
-
         return new_token, result
 
     def get_tag_keys(

tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py

@@ -5,6 +5,7 @@
 from botocore.client import ClientError
 
 from moto import mock_aws
+from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
 from tests import EXAMPLE_AMI_ID, EXAMPLE_AMI_ID2
 from tests.test_ds.test_ds_simple_ad_directory import create_test_directory
 
@@ -1175,3 +1176,30 @@ def test_get_resources_efs():
     assert fs_two["FileSystemArn"] in returned_arns
     assert ap_one["AccessPointArn"] not in returned_arns
     assert ap_two["AccessPointArn"] in returned_arns
+
+
+@mock_aws
+def test_get_resources_stepfunction():
+    simple_definition = (
+        '{"Comment": "An example of the Amazon States Language using a choice state.",'
+        '"StartAt": "DefaultState",'
+        '"States": '
+        '{"DefaultState": {"Type": "Fail","Error": "DefaultStateError","Cause": "No Matches!"}}}'
+    )
+    role_arn = "arn:aws:iam::" + ACCOUNT_ID + ":role/unknown_sf_role"
+
+    client = boto3.client("stepfunctions", region_name="us-east-1")
+    client.create_state_machine(
+        name="name1",
+        definition=str(simple_definition),
+        roleArn=role_arn,
+        tags=[{"key": "Name", "value": "Alice"}],
+    )
+
+    rtapi = boto3.client("resourcegroupstaggingapi", region_name="us-east-1")
+    resp = rtapi.get_resources(ResourceTypeFilters=["states:stateMachine"])
+
+    assert len(resp["ResourceTagMappingList"]) == 1
+    assert {"Key": "Name", "Value": "Alice"} in resp["ResourceTagMappingList"][0][
+        "Tags"
+    ]