Repository: getmoto/moto

Issue:
Added Tagging support for DirectConnect resources
`create_connection` does not handle tags correctly, and the following three functions to manage tags have not been implemented.

1. `describe_tags`
2. `tag_resource`
3. `untag_resource`

Discussion:

Solution patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index 56a6b50cb..3d35c5c87 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -2294,7 +2294,7 @@
 - [ ] describe_loa
 - [ ] describe_locations
 - [ ] describe_router_configuration
-- [ ] describe_tags
+- [X] describe_tags
 - [ ] describe_virtual_gateways
 - [ ] describe_virtual_interfaces
 - [ ] disassociate_connection_from_lag
@@ -2302,8 +2302,8 @@
 - [ ] list_virtual_interface_test_history
 - [ ] start_bgp_failover_test
 - [ ] stop_bgp_failover_test
-- [ ] tag_resource
-- [ ] untag_resource
+- [X] tag_resource
+- [X] untag_resource
 - [X] update_connection
 - [ ] update_direct_connect_gateway
 - [ ] update_direct_connect_gateway_association
diff --git a/moto/directconnect/models.py b/moto/directconnect/models.py
index f402294a8..b0bdcc72c 100644
--- a/moto/directconnect/models.py
+++ b/moto/directconnect/models.py
@@ -6,6 +6,7 @@ from typing import Any, Dict, List, Optional, Tuple
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
+from moto.utilities.tagging_service import TaggingService
 
 from .enums import (
     ConnectionStateType,
@@ -60,9 +61,9 @@ class Connection(BaseModel):
     port_encryption_status: PortEncryptionStatusType
     provider_name: Optional[str]
     region: str
-    tags: Optional[List[Dict[str, str]]]
     vlan: int
     connection_id: str = field(default="", init=False)
+    backend: "DirectConnectBackend"
 
     def __post_init__(self) -> None:
         if self.connection_id == "":
@@ -91,7 +92,7 @@ class Connection(BaseModel):
             "portEncryptionStatus": self.port_encryption_status,
             "providerName": self.provider_name,
             "region": self.region,
-            "tags": self.tags,
+            "tags": self.backend.list_tags_for_resource(self.connection_id),
             "vlan": self.vlan,
         }
 
@@ -158,6 +159,7 @@ class DirectConnectBackend(BaseBackend):
         super().__init__(region_name, account_id)
         self.connections: Dict[str, Connection] = {}
         self.lags: Dict[str, LAG] = {}
+        self.tagger = TaggingService(key_name="key", value_name="value")
 
     def describe_connections(self, connection_id: Optional[str]) -> List[Connection]:
         if connection_id and connection_id not in self.connections:
@@ -209,12 +211,41 @@ class DirectConnectBackend(BaseBackend):
             port_encryption_status=PortEncryptionStatusType.DOWN,
             provider_name=provider_name,
             region=self.region_name,
-            tags=tags,
             vlan=0,
+            backend=self,
         )
+        if tags:
+            self.tag_resource(connection.connection_id, tags)
         self.connections[connection.connection_id] = connection
         return connection
 
+    def tag_resource(self, resource_arn: str, tags: List[Dict[str, str]]) -> None:
+        self.tagger.tag_resource(
+            resource_arn,
+            tags=tags if tags else [],
+        )
+
+    def list_tags_for_resource(self, resource_arn: str) -> List[Dict[str, str]]:
+        tags = self.tagger.get_tag_dict_for_resource(resource_arn)
+        if not tags:
+            return []
+        return [{"key": k, "value": v} for (k, v) in tags.items()]
+
+    def list_tags_for_resources(self, resource_arns: List[str]) -> Dict[str, List[Any]]:
+        response: Dict[str, List[Any]] = {"resourceTags": []}
+        for resource_arn in resource_arns:
+            response["resourceTags"].append(
+                {
+                    "resourceArn": resource_arn,
+                    "tags": self.list_tags_for_resource(resource_arn),
+                }
+            )
+
+        return response
+
+    def untag_resource(self, resource_arn: str, tag_keys: List[str]) -> None:
+        self.tagger.untag_resource_using_names(resource_arn, tag_keys)
+
     def delete_connection(self, connection_id: str) -> Connection:
         if not connection_id:
             raise ConnectionIdMissing()
diff --git a/moto/directconnect/responses.py b/moto/directconnect/responses.py
index 7e2960d6d..88a8d7bea 100644
--- a/moto/directconnect/responses.py
+++ b/moto/directconnect/responses.py
@@ -121,3 +121,27 @@ class DirectConnectResponse(BaseResponse):
                 macSecKeys=[mac_sec_key.to_dict()],
             )
         )
+
+    def tag_resource(self) -> str:
+        params = json.loads(self.body)
+        resource_arn = params.get("resourceArn")
+        tags = params.get("tags")
+        self.directconnect_backend.tag_resource(resource_arn=resource_arn, tags=tags)
+        return json.dumps(dict())
+
+    def untag_resource(self) -> str:
+        params = json.loads(self.body)
+        resource_arn = params.get("resourceArn")
+        tag_keys = params.get("tagKeys", [])
+        self.directconnect_backend.untag_resource(
+            resource_arn=resource_arn, tag_keys=tag_keys
+        )
+        return json.dumps(dict())
+
+    def describe_tags(self) -> str:
+        params = json.loads(self.body)
+        resource_arns = params.get("resourceArns")
+        tags = self.directconnect_backend.list_tags_for_resources(
+            resource_arns=resource_arns
+        )
+        return json.dumps(tags)


Test patch:
diff --git a/tests/test_directconnect/test_directconnect.py b/tests/test_directconnect/test_directconnect.py
index 4b569d2a5..14917be6f 100644
--- a/tests/test_directconnect/test_directconnect.py
+++ b/tests/test_directconnect/test_directconnect.py
@@ -50,6 +50,56 @@ def test_describe_connections(client):
     assert len(resp["connections"]) == 1
 
 
+@mock_aws
+def test_connection_tags(client):
+    client.create_connection(
+        location="EqDC2",
+        bandwidth="10Gbps",
+        connectionName="TestConnection1",
+        requestMACSec=False,
+        tags=[
+            {
+                "key": "name",
+                "value": "Test Connection",
+            },
+            {
+                "key": "connection-id",
+                "value": "test-1",
+            },
+        ],
+    )
+    client.create_connection(
+        location="EqDC2",
+        bandwidth="10Gbps",
+        connectionName="TestConnection2",
+        requestMACSec=True,
+        tags=[
+            {
+                "key": "name",
+                "value": "Test Connection 2",
+            },
+            {
+                "key": "connection-id",
+                "value": "test-2",
+            },
+        ],
+    )
+    resp = client.describe_connections()
+    connections = resp["connections"]
+    assert len(connections) == 2
+
+    assert len(connections[0]["tags"]) == 2
+    assert connections[0]["tags"][0]["key"] == "name"
+    assert connections[0]["tags"][0]["value"] == "Test Connection"
+    assert connections[0]["tags"][1]["key"] == "connection-id"
+    assert connections[0]["tags"][1]["value"] == "test-1"
+
+    assert connections[1]["tags"][0]["key"] == "name"
+    assert connections[1]["tags"][0]["value"] == "Test Connection 2"
+    assert connections[1]["tags"][1]["key"] == "connection-id"
+    assert connections[1]["tags"][1]["value"] == "test-2"
+
+
 def test_delete_connection(client):
     connection = client.create_connection(
         location="EqDC2", bandwidth="10Gbps", connectionName="TestConnection"
@@ -214,3 +264,39 @@ def test_disassociate_mac_sec_key_lag(client):
     )
     lag_id = lag["lagId"]
     _test_disassociate_mac_sec_key_common(client=client, connection_id=lag_id)
+
+
+@mock_aws
+def test_tag_resource(client):
+    connection = client.create_connection(
+        location="EqDC2",
+        bandwidth="10Gbps",
+        connectionName="TestConnection1",
+        requestMACSec=False,
+    )
+    connection_id = connection["connectionId"]
+
+    client.tag_resource(
+        resourceArn=connection_id,
+        tags=[{"key": "t1", "value": "v1"}, {"key": "t2", "value": "v2"}],
+    )
+
+    expected = [{"key": "t1", "value": "v1"}, {"key": "t2", "value": "v2"}]
+    assert get_tags(connection_id, client) == expected
+    assert (
+        client.describe_connections(connectionId=connection_id)["connections"][0][
+            "tags"
+        ]
+        == expected
+    )
+
+    client.untag_resource(resourceArn=connection_id, tagKeys=["t1"])
+    assert get_tags(connection_id, client) == [{"key": "t2", "value": "v2"}]
+
+    client.untag_resource(resourceArn=connection_id, tagKeys=["t2"])
+    assert get_tags(connection_id, client) == []
+
+
+def get_tags(arn, client):
+    tags = client.describe_tags(resourceArns=[arn])
+    return tags["resourceTags"][0]["tags"]