Repository: getmoto/moto (python)

Issue:
Enhancement: Add Step Functions to support resourcegroupstagging
Per the [AWS docs](https://docs.aws.amazon.com/resourcegroupstagging/latest/APIReference/supported-services.html) Step Function support the Resource Groups Tagging API.

It looks like there is a custom implementation for tags for step functions. @bblommers is there a special case with Step Functions or is this a simple use-case to add this support?


Pull request description:
Step Function - Add support for State Machine to resourcegrouptaggingapi
resolves #8212 

Patch:
diff --git a/moto/resourcegroupstaggingapi/models.py b/moto/resourcegroupstaggingapi/models.py
index 8cafdb18b..b384042f6 100644
--- a/moto/resourcegroupstaggingapi/models.py
+++ b/moto/resourcegroupstaggingapi/models.py
@@ -25,6 +25,7 @@ from moto.sagemaker.models import SageMakerModelBackend, sagemaker_backends
 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 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
     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 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                     "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 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
         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(
diff --git a/moto/workspacesweb/models.py b/moto/workspacesweb/models.py
index bd9e80723..a06c5b0d2 100644
--- a/moto/workspacesweb/models.py
+++ b/moto/workspacesweb/models.py
@@ -8,6 +8,8 @@ from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.utilities.utils import get_partition
 
+from ..utilities.tagging_service import TaggingService
+
 
 class FakeUserSettings(BaseModel):
     def __init__(
@@ -23,7 +25,6 @@ class FakeUserSettings(BaseModel):
         idle_disconnect_timeout_in_minutes: int,
         paste_allowed: bool,
         print_allowed: bool,
-        tags: Dict[str, str],
         upload_allowed: bool,
         region_name: str,
         account_id: str,
@@ -41,7 +42,6 @@ class FakeUserSettings(BaseModel):
         self.idle_disconnect_timeout_in_minutes = idle_disconnect_timeout_in_minutes
         self.paste_allowed = paste_allowed if paste_allowed else "Disabled"
         self.print_allowed = print_allowed if print_allowed else "Disabled"
-        self.tags = tags
         self.upload_allowed = upload_allowed if upload_allowed else "Disabled"
         self.associated_portal_arns: List[str] = []
 
@@ -62,7 +62,6 @@ class FakeUserSettings(BaseModel):
             "idleDisconnectTimeoutInMinutes": self.idle_disconnect_timeout_in_minutes,
             "pasteAllowed": self.paste_allowed,
             "printAllowed": self.print_allowed,
-            "tags": self.tags,
             "uploadAllowed": self.upload_allowed,
             "userSettingsArn": self.arn,
         }
@@ -73,7 +72,6 @@ class FakeUserAccessLoggingSettings(BaseModel):
         self,
         client_token: str,
         kinesis_stream_arn: str,
-        tags: Dict[str, str],
         region_name: str,
         account_id: str,
     ):
@@ -83,7 +81,6 @@ class FakeUserAccessLoggingSettings(BaseModel):
         )
         self.client_token = client_token
         self.kinesis_stream_arn = kinesis_stream_arn
-        self.tags = tags
         self.associated_portal_arns: List[str] = []
 
     def arn_formatter(self, _id: str, account_id: str, region_name: str) -> str:
@@ -93,7 +90,6 @@ class FakeUserAccessLoggingSettings(BaseModel):
         return {
             "associatedPortalArns": self.associated_portal_arns,
             "kinesisStreamArn": self.kinesis_stream_arn,
-            "tags": self.tags,
             "userAccessLoggingSettingsArn": self.arn,
         }
 
@@ -103,7 +99,6 @@ class FakeNetworkSettings(BaseModel):
         self,
         security_group_ids: List[str],
         subnet_ids: List[str],
-        tags: Dict[str, str],
         vpc_id: str,
         region_name: str,
         account_id: str,
@@ -112,7 +107,6 @@ class FakeNetworkSettings(BaseModel):
         self.arn = self.arn_formatter(self.network_settings_id, account_id, region_name)
         self.security_group_ids = security_group_ids
         self.subnet_ids = subnet_ids
-        self.tags = tags
         self.vpc_id = vpc_id
         self.associated_portal_arns: List[str] = []
 
@@ -125,7 +119,6 @@ class FakeNetworkSettings(BaseModel):
             "networkSettingsArn": self.arn,
             "securityGroupIds": self.security_group_ids,
             "subnetIds": self.subnet_ids,
-            "Tags": self.tags,
             "vpcId": self.vpc_id,
         }
 
@@ -137,7 +130,6 @@ class FakeBrowserSettings(BaseModel):
         browser_policy: str,
         client_token: str,
         customer_managed_key: str,
-        tags: Dict[str, str],
         region_name: str,
         account_id: str,
     ):
@@ -147,7 +139,6 @@ class FakeBrowserSettings(BaseModel):
         self.browser_policy = browser_policy
         self.client_token = client_token
         self.customer_managed_key = customer_managed_key
-        self.tags = tags
         self.associated_portal_arns: List[str] = []
 
     def arn_formatter(self, _id: str, account_id: str, region_name: str) -> str:
@@ -160,7 +151,6 @@ class FakeBrowserSettings(BaseModel):
             "additionalEncryptionContext": self.additional_encryption_context,
             "browserPolicy": self.browser_policy,
             "customerManagedKey": self.customer_managed_key,
-            "tags": self.tags,
         }
 
 
@@ -174,7 +164,6 @@ class FakePortal(BaseModel):
         display_name: str,
         instance_type: str,
         max_concurrent_sessions: str,
-        tags: Dict[str, str],
         region_name: str,
         account_id: str,
     ):
@@ -187,7 +176,6 @@ class FakePortal(BaseModel):
         self.display_name = display_name
         self.instance_type = instance_type
         self.max_concurrent_sessions = max_concurrent_sessions
-        self.tags = tags
         self.portal_endpoint = f"{self.portal_id}.portal.aws"
         self.browser_type = "Chrome"
         self.creation_time = datetime.datetime.now().isoformat()
@@ -225,7 +213,6 @@ class FakePortal(BaseModel):
             "trustStoreArn": self.trust_store_arn,
             "userAccessLoggingSettingsArn": self.user_access_logging_settings_arn,
             "userSettingsArn": self.user_settings_arn,
-            "tags": self.tags,
         }
 
 
@@ -239,6 +226,7 @@ class WorkSpacesWebBackend(BaseBackend):
         self.user_settings: Dict[str, FakeUserSettings] = {}
         self.user_access_logging_settings: Dict[str, FakeUserAccessLoggingSettings] = {}
         self.portals: Dict[str, FakePortal] = {}
+        self.tagger = TaggingService()
 
     def create_network_settings(
         self,
@@ -250,12 +238,13 @@ class WorkSpacesWebBackend(BaseBackend):
         network_settings_object = FakeNetworkSettings(
             security_group_ids,
             subnet_ids,
-            tags,
             vpc_id,
             self.region_name,
             self.account_id,
         )
         self.network_settings[network_settings_object.arn] = network_settings_object
+        if tags:
+            self.tag_resource("TEMP_CLIENT_TOKEN", network_settings_object.arn, tags)
         return network_settings_object.arn
 
     def list_network_settings(self) -> List[Dict[str, str]]:
@@ -276,18 +265,19 @@ class WorkSpacesWebBackend(BaseBackend):
         browser_policy: str,
         client_token: str,
         customer_managed_key: str,
-        tags: Dict[str, str],
+        tags: Optional[List[Dict[str, str]]] = None,
     ) -> str:
         browser_settings_object = FakeBrowserSettings(
             additional_encryption_context,
             browser_policy,
             client_token,
             customer_managed_key,
-            tags,
             self.region_name,
             self.account_id,
         )
         self.browser_settings[browser_settings_object.arn] = browser_settings_object
+        if tags:
+            self.tag_resource(client_token, browser_settings_object.arn, tags)
         return browser_settings_object.arn
 
     def list_browser_settings(self) -> List[Dict[str, str]]:
@@ -311,7 +301,7 @@ class WorkSpacesWebBackend(BaseBackend):
         display_name: str,
         instance_type: str,
         max_concurrent_sessions: str,
-        tags: Dict[str, str],
+        tags: Optional[List[Dict[str, str]]] = None,
     ) -> Tuple[str, str]:
         portal_object = FakePortal(
             additional_encryption_context,
@@ -321,11 +311,12 @@ class WorkSpacesWebBackend(BaseBackend):
             display_name,
             instance_type,
             max_concurrent_sessions,
-            tags,
             self.region_name,
             self.account_id,
         )
         self.portals[portal_object.arn] = portal_object
+        if tags:
+            self.tag_resource(client_token, portal_object.arn, tags)
         return portal_object.arn, portal_object.portal_endpoint
 
     def list_portals(self) -> List[Dict[str, Any]]:
@@ -405,12 +396,13 @@ class WorkSpacesWebBackend(BaseBackend):
             idle_disconnect_timeout_in_minutes,
             paste_allowed,
             print_allowed,
-            tags,
             upload_allowed,
             self.region_name,
             self.account_id,
         )
         self.user_settings[user_settings_object.arn] = user_settings_object
+        if tags:
+            self.tag_resource(client_token, user_settings_object.arn, tags)
         return user_settings_object.arn
 
     def get_user_settings(self, user_settings_arn: str) -> Dict[str, Any]:
@@ -423,11 +415,15 @@ class WorkSpacesWebBackend(BaseBackend):
         self, client_token: Any, kinesis_stream_arn: Any, tags: Any
     ) -> str:
         user_access_logging_settings_object = FakeUserAccessLoggingSettings(
-            client_token, kinesis_stream_arn, tags, self.region_name, self.account_id
+            client_token, kinesis_stream_arn, self.region_name, self.account_id
         )
         self.user_access_logging_settings[user_access_logging_settings_object.arn] = (
             user_access_logging_settings_object
         )
+        if tags:
+            self.tag_resource(
+                client_token, user_access_logging_settings_object.arn, tags
+            )
         return user_access_logging_settings_object.arn
 
     def get_user_access_logging_settings(
@@ -476,5 +472,18 @@ class WorkSpacesWebBackend(BaseBackend):
             for user_access_logging_settings in self.user_access_logging_settings.values()
         ]
 
+    def tag_resource(self, client_token: str, resource_arn: str, tags: Any) -> None:
+        self.tagger.tag_resource(resource_arn, tags)
+
+    def untag_resource(self, resource_arn: str, tag_keys: Any) -> None:
+        self.tagger.untag_resource_using_names(resource_arn, tag_keys)
+
+    def list_tags_for_resource(self, resource_arn: str) -> List[Dict[str, str]]:
+        tags = self.tagger.get_tag_dict_for_resource(resource_arn)
+        Tags = []
+        for key, value in tags.items():
+            Tags.append({"Key": key, "Value": value})
+        return Tags
+
 
 workspacesweb_backends = BackendDict(WorkSpacesWebBackend, "workspaces-web")
diff --git a/moto/workspacesweb/responses.py b/moto/workspacesweb/responses.py
index a9655412e..0878b99ba 100644
--- a/moto/workspacesweb/responses.py
+++ b/moto/workspacesweb/responses.py
@@ -338,3 +338,30 @@ class WorkSpacesWebResponse(BaseResponse):
             self.workspacesweb_backend.list_user_access_logging_settings()
         )
         return json.dumps(dict(userAccessLoggingSettings=user_access_logging_settings))
+
+    def tag_resource(self) -> str:
+        client_token = self._get_param("clientToken")
+        resource_arn = unquote(self._get_param("resourceArn"))
+        tags = self._get_param("tags")
+        self.workspacesweb_backend.tag_resource(
+            client_token=client_token,
+            resource_arn=resource_arn,
+            tags=tags,
+        )
+        return json.dumps(dict())
+
+    def untag_resource(self) -> str:
+        tagKeys = self.__dict__["data"]["tagKeys"]
+        resource_arn = unquote(self._get_param("resourceArn"))
+        self.workspacesweb_backend.untag_resource(
+            resource_arn=resource_arn,
+            tag_keys=tagKeys,
+        )
+        return json.dumps(dict())
+
+    def list_tags_for_resource(self) -> str:
+        resource_arn = unquote(self.parsed_url.path.split("/tags/")[-1])
+        tags = self.workspacesweb_backend.list_tags_for_resource(
+            resource_arn=resource_arn,
+        )
+        return json.dumps(dict(tags=tags))
diff --git a/moto/workspacesweb/urls.py b/moto/workspacesweb/urls.py
index 8c2fb156a..bb3df663f 100644
--- a/moto/workspacesweb/urls.py
+++ b/moto/workspacesweb/urls.py
@@ -21,4 +21,5 @@ url_paths = {
     "{0}/portals/(?P<portalArn>.*)/networkSettings$": WorkSpacesWebResponse.dispatch,
     "{0}/portals/(?P<portalArn>.*)/userSettings$": WorkSpacesWebResponse.dispatch,
     "{0}/portals/(?P<portalArn>.*)/userAccessLoggingSettings$": WorkSpacesWebResponse.dispatch,
+    "{0}/tags/(?P<resourceArn>.+)$": WorkSpacesWebResponse.dispatch,
 }


Test patch:
diff --git a/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py b/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py
index 05e656675..fc73f210d 100644
--- a/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py
+++ b/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py
@@ -5,6 +5,7 @@ import pytest
 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"
+    ]
diff --git a/tests/test_workspacesweb/test_workspacesweb.py b/tests/test_workspacesweb/test_workspacesweb.py
index 47a46c26f..ef5f570de 100644
--- a/tests/test_workspacesweb/test_workspacesweb.py
+++ b/tests/test_workspacesweb/test_workspacesweb.py
@@ -13,6 +13,7 @@ FAKE_SECURITY_GROUP_IDS = ["sg-0123456789abcdef0"]
 FAKE_SUBNET_IDS = ["subnet-0123456789abcdef0", "subnet-abcdef0123456789"]
 FAKE_TAGS = [
     {"Key": "TestKey", "Value": "TestValue"},
+    {"Key": "TestKey2", "Value": "TestValue2"},
 ]
 FAKE_VPC_ID = "vpc-0123456789abcdef0"
 FAKE_KMS_KEY_ID = "abcd1234-5678-90ab-cdef-FAKEKEY"
@@ -436,3 +437,80 @@ def test_associate_user_access_logging_settings():
         userAccessLoggingSettingsArn=user_access_logging_settings_arn
     )["userAccessLoggingSettings"]
     assert resp["associatedPortalArns"] == [portal_arn]
+
+
+@mock_aws
+def test_list_tags_for_resource():
+    client = boto3.client("workspaces-web", region_name="ap-southeast-1")
+
+    OTHER_FAKE_TAGS = [
+        {"Key": "FAKEKEY1", "Value": "FAKEVALUE1"},
+        {"Key": "FAKEKEY2", "Value": "FAKEVALUE2"},
+    ]
+
+    arns = []
+
+    arns.append(
+        client.create_portal(
+            additionalEncryptionContext={"Key1": "Encryption", "Key2": "Context"},
+            authenticationType="Standard",
+            clientToken="TestClient",
+            customerManagedKey=FAKE_KMS_KEY_ID,
+            displayName="TestDisplayName",
+            instanceType="TestInstanceType",
+            maxConcurrentSessions=5,
+            tags=FAKE_TAGS,
+        )["portalArn"]
+    )
+
+    arns.append(
+        client.create_network_settings(
+            securityGroupIds=FAKE_SECURITY_GROUP_IDS,
+            subnetIds=FAKE_SUBNET_IDS,
+            tags=FAKE_TAGS,
+            vpcId=FAKE_VPC_ID,
+        )["networkSettingsArn"]
+    )
+
+    arns.append(
+        client.create_user_settings(
+            copyAllowed="Disabled",
+            pasteAllowed="Disabled",
+            printAllowed="Disabled",
+            uploadAllowed="Disabled",
+            downloadAllowed="Disabled",
+            tags=FAKE_TAGS,
+        )["userSettingsArn"]
+    )
+
+    arns.append(
+        client.create_user_access_logging_settings(
+            kinesisStreamArn="arn:aws:kinesis:ap-southeast-1:123456789012:stream/TestStream",
+            tags=FAKE_TAGS,
+        )["userAccessLoggingSettingsArn"]
+    )
+
+    arns.append(
+        client.create_browser_settings(
+            additionalEncryptionContext={"Key1": "Value1", "Key2": "Value2"},
+            browserPolicy="TestBrowserPolicy",
+            clientToken="TestClient",
+            customerManagedKey=FAKE_KMS_KEY_ID,
+            tags=FAKE_TAGS,
+        )["browserSettingsArn"]
+    )
+
+    for arn in arns:
+        resp = client.list_tags_for_resource(resourceArn=arn)
+        assert resp["tags"] == FAKE_TAGS
+
+        client.tag_resource(resourceArn=arn, tags=OTHER_FAKE_TAGS)
+        resp = client.list_tags_for_resource(resourceArn=arn)
+        assert resp["tags"] == FAKE_TAGS + OTHER_FAKE_TAGS
+
+        client.untag_resource(resourceArn=arn, tagKeys=["FAKEKEY1", "TestKey"])
+        resp = client.list_tags_for_resource(resourceArn=arn)
+        assert resp["tags"] == [
+            {"Key": "TestKey2", "Value": "TestValue2"},
+            {"Key": "FAKEKEY2", "Value": "FAKEVALUE2"},
+        ]