Feature: Add NetworkManager resource support

moto/networkmanager/models.py

@@ -1,12 +1,11 @@
 """NetworkManagerBackend class with methods for supported APIs."""
 
-import random
 from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
-from moto.ec2.utils import HEX_CHARS
+from moto.moto_api._internal import mock_random
 from moto.utilities.paginator import paginate
 from moto.utilities.utils import PARTITION_NAMES
 
@@ -32,15 +31,16 @@ class GlobalNetwork(BaseModel):
     def __init__(
         self,
         account_id: str,
+        partition: str,
         description: Optional[str],
         tags: Optional[List[Dict[str, str]]],
     ):
         self.description = description
         self.tags = tags or []
         self.global_network_id = "global-network-" + "".join(
-            random.choice(HEX_CHARS) for _ in range(18)
+            mock_random.get_random_hex(18)
         )
-        self.global_network_arn = f"arn:aws:networkmanager:{account_id}:global-network/{self.global_network_id}"
+        self.global_network_arn = f"arn:{partition}:networkmanager:{account_id}:global-network/{self.global_network_id}"
         self.created_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
         self.state = "PENDING"
 
@@ -59,6 +59,7 @@ class CoreNetwork(BaseModel):
     def __init__(
         self,
         account_id: str,
+        partition: str,
         global_network_id: str,
         description: Optional[str],
         tags: Optional[List[Dict[str, str]]],
@@ -70,12 +71,8 @@ def __init__(
         self.tags = tags or []
         self.policy_document = policy_document
         self.client_token = client_token
-        self.core_network_id = "core-network-" + "".join(
-            random.choice(HEX_CHARS) for _ in range(18)
-        )
-        self.core_network_arn = (
-            f"arn:aws:networkmanager:{account_id}:core-network/{self.core_network_id}"
-        )
+        self.core_network_id = "core-network-" + "".join(mock_random.get_random_hex(18))
+        self.core_network_arn = f"arn:{partition}:networkmanager:{account_id}:core-network/{self.core_network_id}"
 
         self.created_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
         self.state = "PENDING"
@@ -101,8 +98,6 @@ def __init__(self, region_name: str, account_id: str) -> None:
         self.global_networks: Dict[str, GlobalNetwork] = {}
         self.core_networks: Dict[str, CoreNetwork] = {}
 
-    # add methods from here
-
     def create_global_network(
         self,
         description: Optional[str],
@@ -112,6 +107,7 @@ def create_global_network(
             description=description,
             tags=tags,
             account_id=self.account_id,
+            partition=self.partition,
         )
         gnw_id = global_network.global_network_id
         self.global_networks[gnw_id] = global_network
@@ -136,6 +132,7 @@ def create_core_network(
             policy_document=policy_document,
             client_token=client_token,
             account_id=self.account_id,
+            partition=self.partition,
         )
         cnw_id = core_network.core_network_id
         self.core_networks[cnw_id] = core_network

moto/networkmanager/responses.py

@@ -55,7 +55,7 @@ def delete_core_network(self) -> str:
     def tag_resource(self) -> TYPE_RESPONSE:
         params = json.loads(self.body)
         tags = params.get("Tags")
-        resource_arn = unquote(self.path.split("/")[-1])
+        resource_arn = unquote(self.path.split("/tags/")[-1])
 
         self.networkmanager_backend.tag_resource(
             resource_arn=resource_arn,
@@ -66,7 +66,7 @@ def tag_resource(self) -> TYPE_RESPONSE:
     def untag_resource(self) -> TYPE_RESPONSE:
         params = self._get_params()
         tag_keys = params.get("tagKeys")
-        resource_arn = unquote(self.path.split("/")[-1])
+        resource_arn = unquote(self.path.split("/tags/")[-1])
         self.networkmanager_backend.untag_resource(
             resource_arn=resource_arn,
             tag_keys=tag_keys,

moto/networkmanager/urls.py

@@ -14,4 +14,5 @@
     "{0}/global-networks/(?P<networkid>[^/.]+)$": NetworkManagerResponse.dispatch,
     "{0}/tags$": NetworkManagerResponse.dispatch,
     "{0}/tags/(?P<resourcearn>[^/.]+)$": NetworkManagerResponse.dispatch,
+    "{0}/tags/(?P<arn_prefix>[^/]+)/(?P<resource_id>[^/]+)$": NetworkManagerResponse.dispatch,
 }

tests/test_networkmanager/test_networkmanager.py

@@ -3,6 +3,7 @@
 import boto3
 
 from moto import mock_aws
+from tests import DEFAULT_ACCOUNT_ID
 
 # See our Development Tips on writing tests for hints on how to write good tests:
 # http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html
@@ -28,6 +29,10 @@ def test_create_global_network():
     )
 
     global_network = resp["GlobalNetwork"]
+    assert (
+        global_network["GlobalNetworkArn"]
+        == f"arn:aws:networkmanager:{DEFAULT_ACCOUNT_ID}:global-network/{global_network['GlobalNetworkId']}"
+    )
     assert global_network["Description"] == "Test global network"
     assert global_network["Tags"] == [{"Key": "Name", "Value": "TestNetwork"}]
     assert global_network["State"] == "PENDING"
@@ -55,6 +60,10 @@ def test_create_core_network():
     )
 
     core_network = resp["CoreNetwork"]
+    assert (
+        core_network["CoreNetworkArn"]
+        == f"arn:aws:networkmanager:{DEFAULT_ACCOUNT_ID}:core-network/{core_network['CoreNetworkId']}"
+    )
     assert core_network["GlobalNetworkId"] == global_network_id
     assert core_network["Description"] == "Test core network"
     assert len(core_network["Tags"]) == 1