diff --git a/docs/docs/services/networkmanager.rst b/docs/docs/services/networkmanager.rst index a66a4a68d53d..ffa82b961648 100644 --- a/docs/docs/services/networkmanager.rst +++ b/docs/docs/services/networkmanager.rst @@ -25,10 +25,10 @@ networkmanager - [ ] create_connect_peer - [ ] create_connection - [X] create_core_network -- [ ] create_device +- [X] create_device - [X] create_global_network -- [ ] create_link -- [ ] create_site +- [X] create_link +- [X] create_site - [ ] create_site_to_site_vpn_attachment - [ ] create_transit_gateway_peering - [ ] create_transit_gateway_route_table_attachment @@ -38,12 +38,12 @@ networkmanager - [ ] delete_connection - [X] delete_core_network - [ ] delete_core_network_policy_version -- [ ] delete_device +- [X] delete_device - [ ] delete_global_network -- [ ] delete_link +- [X] delete_link - [ ] delete_peering - [ ] delete_resource_policy -- [ ] delete_site +- [X] delete_site - [ ] deregister_transit_gateway - [X] describe_global_networks - [ ] disassociate_connect_peer @@ -60,9 +60,9 @@ networkmanager - [ ] get_core_network_change_set - [ ] get_core_network_policy - [ ] get_customer_gateway_associations -- [ ] get_devices +- [X] get_devices - [ ] get_link_associations -- [ ] get_links +- [X] get_links - [ ] get_network_resource_counts - [ ] get_network_resource_relationships - [ ] get_network_resources @@ -71,7 +71,7 @@ networkmanager - [ ] get_resource_policy - [ ] get_route_analysis - [ ] get_site_to_site_vpn_attachment -- [ ] get_sites +- [X] get_sites - [ ] get_transit_gateway_connect_peer_associations - [ ] get_transit_gateway_peering - [ ] get_transit_gateway_registrations diff --git a/moto/networkmanager/models.py b/moto/networkmanager/models.py index d0e87e6b0b0f..e8eee54a349e 100644 --- a/moto/networkmanager/models.py +++ b/moto/networkmanager/models.py @@ -9,7 +9,7 @@ from moto.utilities.paginator import paginate from moto.utilities.utils import PARTITION_NAMES -from .exceptions import ResourceNotFound +from .exceptions import ResourceNotFound, ValidationError PAGINATION_MODEL = { "describe_global_networks": { @@ -24,6 +24,24 @@ "limit_default": 100, "unique_attribute": "core_network_arn", }, + "get_sites": { + "input_token": "next_token", + "limit_key": "max_results", + "limit_default": 100, + "unique_attribute": "site_arn", + }, + "get_links": { + "input_token": "next_token", + "limit_key": "max_results", + "limit_default": 100, + "unique_attribute": "link_arn", + }, + "get_devices": { + "input_token": "next_token", + "limit_key": "max_results", + "limit_default": 100, + "unique_attribute": "device_arn", + }, } @@ -90,6 +108,135 @@ def to_dict(self) -> Dict[str, Any]: } +class Site(BaseModel): + def __init__( + self, + account_id: str, + partition: str, + global_network_id: str, + description: Optional[str], + location: Optional[Dict[str, Any]], + tags: Optional[List[Dict[str, str]]], + ): + self.global_network_id = global_network_id + self.description = description + self.location = location + self.tags = tags or [] + self.site_id = "site-" + "".join(mock_random.get_random_hex(18)) + self.site_arn = ( + f"arn:{partition}:networkmanager:{account_id}:site/{self.site_id}" + ) + self.created_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") + self.state = "PENDING" + + def to_dict(self) -> Dict[str, Any]: + return { + "SiteId": self.site_id, + "SiteArn": self.site_arn, + "GlobalNetworkId": self.global_network_id, + "Description": self.description, + "Location": self.location, + "Tags": self.tags, + "State": self.state, + "CreatedAt": self.created_at, + } + + +class Link(BaseModel): + def __init__( + self, + account_id: str, + partition: str, + global_network_id: str, + description: Optional[str], + type: Optional[str], + bandwidth: Dict[str, int], + provider: Optional[str], + site_id: str, + tags: Optional[List[Dict[str, str]]], + ): + self.global_network_id = global_network_id + self.description = description + self.type = type + self.bandwidth = bandwidth + self.provider = provider + self.site_id = site_id + self.tags = tags or [] + self.link_id = "link-" + "".join(mock_random.get_random_hex(18)) + self.link_arn = ( + f"arn:{partition}:networkmanager:{account_id}:link/{self.link_id}" + ) + self.created_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") + self.state = "PENDING" + + def to_dict(self) -> Dict[str, Any]: + return { + "LinkId": self.link_id, + "LinkArn": self.link_arn, + "GlobalNetworkId": self.global_network_id, + "Description": self.description, + "Type": self.type, + "Bandwidth": self.bandwidth, + "Provider": self.provider, + "SiteId": self.site_id, + "Tags": self.tags, + "State": self.state, + "CreatedAt": self.created_at, + } + + +class Device(BaseModel): + def __init__( + self, + account_id: str, + partition: str, + global_network_id: str, + aws_location: Optional[Dict[str, str]], + description: Optional[str], + type: Optional[str], + vendor: Optional[str], + model: Optional[str], + serial_number: Optional[str], + location: Optional[Dict[str, str]], + site_id: Optional[str], + tags: Optional[List[Dict[str, str]]], + ): + self.global_network_id = global_network_id + self.aws_location = aws_location + self.description = description + self.type = type + self.vendor = vendor + self.model = model + self.serial_number = serial_number + self.location = location + self.site_id = site_id + self.tags = tags or [] + self.device_id = "device-" + "".join(mock_random.get_random_hex(18)) + self.device_arn = ( + f"arn:{partition}:networkmanager:{account_id}:device/{self.device_id}" + ) + self.created_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") + self.state = "PENDING" + + def to_dict(self) -> Dict[str, Any]: + return { + "DeviceId": self.device_id, + "DeviceArn": self.device_arn, + "GlobalNetworkId": self.global_network_id, + "AWSLocation": self.aws_location, + "Description": self.description, + "Type": self.type, + "Vendor": self.vendor, + "Model": self.model, + "SerialNumber": self.serial_number, + "Location": self.location, + "SiteId": self.site_id, + "Tags": self.tags, + "State": self.state, + "CreatedAt": self.created_at, + } + + class NetworkManagerBackend(BaseBackend): """Implementation of NetworkManager APIs.""" @@ -97,6 +244,32 @@ def __init__(self, region_name: str, account_id: str) -> None: super().__init__(region_name, account_id) self.global_networks: Dict[str, GlobalNetwork] = {} self.core_networks: Dict[str, CoreNetwork] = {} + self.sites: Dict[str, Dict[str, Site]] = {} + self.links: Dict[str, Dict[str, Link]] = {} + self.devices: Dict[str, Dict[str, Device]] = {} + + def _get_resource_from_arn(self, arn: str) -> Any: + resources_types: Dict[str, Dict[str, Any]] = { + "core-network": self.core_networks, + "global-network": self.global_networks, + "site": self.sites, + "link": self.links, + "device": self.devices, + } + try: + target_resource, target_name = arn.split(":")[-1].split("/") + resources = resources_types.get(target_resource, {}) + # Flatten the nested dictionary stores + if target_resource not in ["core-network", "global-network"]: + resources = { + k: v + for inner_dict in resources.values() + for k, v in inner_dict.items() + } + resource = resources.get(target_name) # type: ignore + except (KeyError, ValueError, AttributeError): + raise ResourceNotFound(arn) + return resource def create_global_network( self, @@ -111,6 +284,10 @@ def create_global_network( ) gnw_id = global_network.global_network_id self.global_networks[gnw_id] = global_network + # Create empty dict for resources + self.sites[gnw_id] = {} + self.links[gnw_id] = {} + self.devices[gnw_id] = {} return global_network def create_core_network( @@ -165,18 +342,6 @@ def get_core_network(self, core_network_id: str) -> CoreNetwork: core_network = self.core_networks[core_network_id] return core_network - def _get_resource_from_arn(self, arn: str) -> Any: - resources = { - "core-network": self.core_networks, - "global-network": self.global_networks, - } - try: - target_resource, target_name = arn.split(":")[-1].split("/") - resource = resources.get(target_resource).get(target_name) # type: ignore - except (KeyError, ValueError): - raise ResourceNotFound(arn) - return resource - @paginate(pagination_model=PAGINATION_MODEL) def describe_global_networks( self, global_network_ids: List[str] @@ -195,6 +360,177 @@ def describe_global_networks( queried_global_networks.append(global_network) return queried_global_networks + def create_site( + self, + global_network_id: str, + description: Optional[str], + location: Optional[Dict[str, str]], + tags: Optional[List[Dict[str, str]]], + ) -> Site: + # check if global network exists + if global_network_id not in self.global_networks: + raise ResourceNotFound(global_network_id) + site = Site( + global_network_id=global_network_id, + description=description, + location=location, + tags=tags, + account_id=self.account_id, + partition=self.partition, + ) + site_id = site.site_id + self.sites[global_network_id][site_id] = site + return site + + def delete_site(self, global_network_id: str, site_id: str) -> Site: + if global_network_id not in self.global_networks: + raise ResourceNotFound(global_network_id) + if global_network_id not in self.sites: + raise ResourceNotFound(site_id) + gn_sites = self.sites[global_network_id] + site = gn_sites.pop(site_id) + site.state = "DELETING" + return site + + @paginate(pagination_model=PAGINATION_MODEL) + def get_sites(self, global_network_id: str, site_ids: List[str]) -> List[Site]: + if global_network_id not in self.global_networks: + raise ValidationError("Incorrect input.") + gn_sites = self.sites.get(global_network_id) or {} + queried = [] + if not site_ids: + queried = list(gn_sites.values()) + else: + for id in site_ids: + if id in gn_sites: + q = gn_sites[id] + queried.append(q) + + return queried + + def create_link( + self, + global_network_id: str, + description: Optional[str], + type: Optional[str], + bandwidth: Dict[str, Any], + provider: Optional[str], + site_id: str, + tags: Optional[List[Dict[str, str]]], + ) -> Link: + # check if global network exists + if global_network_id not in self.global_networks: + raise ResourceNotFound(global_network_id) + link = Link( + global_network_id=global_network_id, + description=description, + type=type, + bandwidth=bandwidth, + provider=provider, + site_id=site_id, + tags=tags, + account_id=self.account_id, + partition=self.partition, + ) + self.links[global_network_id][link.link_id] = link + return link + + @paginate(pagination_model=PAGINATION_MODEL) + def get_links( + self, + global_network_id: str, + link_ids: List[str], + site_id: str, + type: str, + provider: str, + ) -> List[Link]: + if global_network_id not in self.global_networks: + raise ValidationError("Incorrect input.") + # TODO: Implement filtering by site_id, type, provider + gn_links = self.links.get(global_network_id) or {} + queried = [] + if not link_ids: + queried = list(gn_links.values()) + else: + for id in link_ids: + if id in gn_links: + q = gn_links[id] + queried.append(q) + return queried + + def delete_link(self, global_network_id: str, link_id: str) -> Link: + try: + link = self.links[global_network_id].pop(link_id) + except KeyError: + raise ResourceNotFound(link_id) + link.state = "DELETING" + return link + + def create_device( + self, + global_network_id: str, + aws_location: Optional[Dict[str, str]], + description: Optional[str], + type: Optional[str], + vendor: Optional[str], + model: Optional[str], + serial_number: Optional[str], + location: Optional[Dict[str, str]], + site_id: Optional[str], + tags: Optional[List[Dict[str, str]]], + ) -> Device: + # check if global network exists + if global_network_id not in self.global_networks: + raise ResourceNotFound(global_network_id) + device = Device( + global_network_id=global_network_id, + aws_location=aws_location, + description=description, + type=type, + vendor=vendor, + model=model, + serial_number=serial_number, + location=location, + site_id=site_id, + tags=tags, + account_id=self.account_id, + partition=self.partition, + ) + self.devices[global_network_id][device.device_id] = device + return device + + @paginate(pagination_model=PAGINATION_MODEL) + def get_devices( + self, global_network_id: str, device_ids: List[str], site_id: Optional[str] + ) -> List[Device]: + if global_network_id not in self.global_networks: + raise ValidationError("Incorrect input.") + # TODO: Implement filtering by site_id + gn_devices = self.devices.get(global_network_id) or {} + queried = [] + if not device_ids: + queried = list(gn_devices.values()) + else: + for id in device_ids: + if id in gn_devices: + q = gn_devices[id] + queried.append(q) + + return queried + + def delete_device(self, global_network_id: str, device_id: str) -> Device: + try: + device = self.devices[global_network_id].pop(device_id) + except KeyError: + raise ResourceNotFound(device_id) + device.state = "DELETING" + return device + + def list_tags_for_resource(self, resource_arn: str) -> List[Dict[str, str]]: + resource = self._get_resource_from_arn(resource_arn) + tag_list = resource.tags + return tag_list + networkmanager_backends = BackendDict( NetworkManagerBackend, diff --git a/moto/networkmanager/responses.py b/moto/networkmanager/responses.py index b8ccfbe1e998..d64cf8e1ac5f 100644 --- a/moto/networkmanager/responses.py +++ b/moto/networkmanager/responses.py @@ -109,3 +109,150 @@ def describe_global_networks(self) -> str: return json.dumps( dict(GlobalNetworks=list_global_networks, nextToken=next_token) ) + + def create_site(self) -> str: + params = json.loads(self.body) + global_network_id = unquote(self.path.split("/")[-2]) + description = params.get("Description") + location = params.get("Location") + tags = params.get("Tags") + site = self.networkmanager_backend.create_site( + global_network_id=global_network_id, + description=description, + location=location, + tags=tags, + ) + return json.dumps(dict(Site=site.to_dict())) + + def delete_site(self) -> str: + global_network_id = unquote(self.path.split("/")[-3]) + site_id = unquote(self.path.split("/")[-1]) + site = self.networkmanager_backend.delete_site( + global_network_id=global_network_id, + site_id=site_id, + ) + return json.dumps(dict(Site=site.to_dict())) + + def get_sites(self) -> str: + params = self._get_params() + global_network_id = unquote(self.path.split("/")[-2]) + site_ids = self.querystring.get("siteIds") + max_results = params.get("MaxResults") + next_token = params.get("NextToken") + sites, next_token = self.networkmanager_backend.get_sites( + global_network_id=global_network_id, + site_ids=site_ids, + max_results=max_results, + next_token=next_token, + ) + list_sites = [site.to_dict() for site in sites] + return json.dumps(dict(Sites=list_sites, nextToken=next_token)) + + def create_link(self) -> str: + params = json.loads(self.body) + global_network_id = unquote(self.path.split("/")[-2]) + description = params.get("Description") + type = params.get("Type") + bandwidth = params.get("Bandwidth") + provider = params.get("Provider") + site_id = params.get("SiteId") + tags = params.get("Tags") + link = self.networkmanager_backend.create_link( + global_network_id=global_network_id, + description=description, + type=type, + bandwidth=bandwidth, + provider=provider, + site_id=site_id, + tags=tags, + ) + return json.dumps(dict(Link=link.to_dict())) + + def get_links(self) -> str: + params = self._get_params() + global_network_id = unquote(self.path.split("/")[-2]) + link_ids = self.querystring.get("linkIds") + site_id = params.get("SiteId") + type = params.get("Type") + provider = params.get("Provider") + max_results = params.get("MaxResults") + next_token = params.get("NextToken") + links, next_token = self.networkmanager_backend.get_links( + global_network_id=global_network_id, + link_ids=link_ids, + site_id=site_id, + type=type, + provider=provider, + max_results=max_results, + next_token=next_token, + ) + list_links = [link.to_dict() for link in links] + return json.dumps(dict(Links=list_links, nextToken=next_token)) + + def delete_link(self) -> str: + global_network_id = unquote(self.path.split("/")[-3]) + link_id = unquote(self.path.split("/")[-1]) + link = self.networkmanager_backend.delete_link( + global_network_id=global_network_id, + link_id=link_id, + ) + return json.dumps(dict(Link=link.to_dict())) + + def create_device(self) -> str: + params = json.loads(self.body) + global_network_id = unquote(self.path.split("/")[-2]) + aws_location = params.get("AWSLocation") + description = params.get("Description") + type = params.get("Type") + vendor = params.get("Vendor") + model = params.get("Model") + serial_number = params.get("SerialNumber") + location = params.get("Location") + site_id = params.get("SiteId") + tags = params.get("Tags") + device = self.networkmanager_backend.create_device( + global_network_id=global_network_id, + aws_location=aws_location, + description=description, + type=type, + vendor=vendor, + model=model, + serial_number=serial_number, + location=location, + site_id=site_id, + tags=tags, + ) + return json.dumps(dict(Device=device.to_dict())) + + def get_devices(self) -> str: + params = self._get_params() + global_network_id = unquote(self.path.split("/")[-2]) + device_ids = self.querystring.get("deviceIds") + site_id = params.get("SiteId") + max_results = params.get("MaxResults") + next_token = params.get("NextToken") + devices, next_token = self.networkmanager_backend.get_devices( + global_network_id=global_network_id, + device_ids=device_ids, + site_id=site_id, + max_results=max_results, + next_token=next_token, + ) + list_devices = [device.to_dict() for device in devices] + return json.dumps(dict(Devices=list_devices, nextToken=next_token)) + + def delete_device(self) -> str: + global_network_id = unquote(self.path.split("/")[-3]) + device_id = unquote(self.path.split("/")[-1]) + device = self.networkmanager_backend.delete_device( + global_network_id=global_network_id, + device_id=device_id, + ) + return json.dumps(dict(Device=device.to_dict())) + + def list_tags_for_resource(self) -> str: + resource_arn = unquote(self.path.split("/tags/")[-1]) + tag_list = self.networkmanager_backend.list_tags_for_resource( + resource_arn=resource_arn, + ) + return json.dumps(dict(TagList=tag_list)) diff --git a/moto/networkmanager/urls.py b/moto/networkmanager/urls.py index 7247a03522dd..4badd3d9f505 100644 --- a/moto/networkmanager/urls.py +++ b/moto/networkmanager/urls.py @@ -9,9 +9,15 @@ url_paths = { "{0}/$": NetworkManagerResponse.dispatch, "{0}/global-networks$": NetworkManagerResponse.dispatch, + "{0}/global-networks/(?P[^/.]+)$": NetworkManagerResponse.dispatch, + "{0}/global-networks/(?P[^/.]+)/sites$": NetworkManagerResponse.dispatch, + "{0}/global-networks/(?P[^/.]+)/sites/(?P[^/.]+)$": NetworkManagerResponse.dispatch, + "{0}/global-networks/(?P[^/.]+)/links$": NetworkManagerResponse.dispatch, + "{0}/global-networks/(?P[^/.]+)/links/(?P[^/.]+)$": NetworkManagerResponse.dispatch, + "{0}/global-networks/(?P[^/.]+)/devices$": NetworkManagerResponse.dispatch, + "{0}/global-networks/(?P[^/.]+)/devices/(?P[^/.]+)$": NetworkManagerResponse.dispatch, "{0}/core-networks$": NetworkManagerResponse.dispatch, "{0}/core-networks/(?P[^/.]+)$": NetworkManagerResponse.dispatch, - "{0}/global-networks/(?P[^/.]+)$": NetworkManagerResponse.dispatch, "{0}/tags$": NetworkManagerResponse.dispatch, "{0}/tags/(?P[^/.]+)$": NetworkManagerResponse.dispatch, "{0}/tags/(?P[^/]+)/(?P[^/]+)$": NetworkManagerResponse.dispatch, diff --git a/tests/test_networkmanager/test_networkmanager.py b/tests/test_networkmanager/test_networkmanager.py index 6ebed4a0fabb..0a35d08c55b9 100644 --- a/tests/test_networkmanager/test_networkmanager.py +++ b/tests/test_networkmanager/test_networkmanager.py @@ -1,6 +1,7 @@ """Unit tests for networkmanager-supported APIs.""" import boto3 +import pytest from moto import mock_aws from tests import DEFAULT_ACCOUNT_ID @@ -9,12 +10,10 @@ # http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html +@mock_aws def create_global_network(client) -> str: return client.create_global_network( Description="Test global network", - Tags=[ - {"Key": "Name", "Value": "TestNetwork"}, - ], )["GlobalNetwork"]["GlobalNetworkId"] @@ -84,6 +83,10 @@ def test_delete_core_network(): @mock_aws def test_tag_resource(): + test_tags = [ + {"Key": "Moto", "Value": "TestTag"}, + {"Key": "Owner", "Value": "Alice"}, + ] client = boto3.client("networkmanager") gn_id = create_global_network(client) cn = client.create_core_network(GlobalNetworkId=gn_id)["CoreNetwork"] @@ -91,28 +94,39 @@ def test_tag_resource(): # Check tagging core-network resp = client.tag_resource( ResourceArn=cn["CoreNetworkArn"], - Tags=[{"Key": "Test", "Value": "TestValue-Core"}], + Tags=test_tags, ) assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200 updated_cn = client.get_core_network(CoreNetworkId=cn["CoreNetworkId"])[ "CoreNetwork" ] - assert updated_cn["Tags"] == [{"Key": "Test", "Value": "TestValue-Core"}] + assert updated_cn["Tags"] == test_tags # Check tagging global-network gn_arn = client.describe_global_networks()["GlobalNetworks"][0]["GlobalNetworkArn"] - resp = client.tag_resource( - ResourceArn=gn_arn, Tags=[{"Key": "Test", "Value": "TestValue-Global"}] - ) + resp = client.tag_resource(ResourceArn=gn_arn, Tags=test_tags) assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200 updated_gn = client.describe_global_networks(GlobalNetworkIds=[gn_id])[ "GlobalNetworks" ][0] - assert len(updated_gn["Tags"]) == 2 - assert updated_gn["Tags"] == [ - {"Key": "Name", "Value": "TestNetwork"}, - {"Key": "Test", "Value": "TestValue-Global"}, - ] + assert updated_gn["Tags"] == test_tags + + # Check tagging site + site = client.create_site( + GlobalNetworkId=gn_id, + Description="Test site", + Location={ + "Address": "123 Main St", + "Latitude": "47.6062", + "Longitude": "122.3321", + }, + )["Site"] + resp = client.tag_resource(ResourceArn=site["SiteArn"], Tags=test_tags) + assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200 + updated_site = client.get_sites(GlobalNetworkId=gn_id, SiteIds=[site["SiteId"]])[ + "Sites" + ][0] + assert updated_site["Tags"] == test_tags @mock_aws @@ -191,3 +205,329 @@ def test_describe_global_networks(): 0 ] assert gn["GlobalNetworkId"] == g_id + + +@mock_aws +def test_create_site(): + client = boto3.client("networkmanager") + gn_id = create_global_network(client) + site = client.create_site( + GlobalNetworkId=gn_id, + Description="Test site", + Location={ + "Address": "123 Main St", + "Latitude": "47.6062", + "Longitude": "122.3321", + }, + Tags=[ + {"Key": "Name", "Value": "TestSite"}, + ], + )["Site"] + assert site["GlobalNetworkId"] == gn_id + assert site["Description"] == "Test site" + assert len(site["Tags"]) == 1 + + +@mock_aws +def test_delete_site(): + client = boto3.client("networkmanager") + gn_id = create_global_network(client) + site_id = client.create_site( + GlobalNetworkId=gn_id, Description="Test site to be deleted" + )["Site"]["SiteId"] + + resp = client.delete_site(GlobalNetworkId=gn_id, SiteId=site_id) + assert resp["Site"]["State"] == "DELETING" + + +@mock_aws +def test_get_sites(): + NUM_SITES = 4 + NUM_TO_TEST = 2 + client = boto3.client("networkmanager") + gn_id = create_global_network(client) + site_ids = [] + for i in range(NUM_SITES): + site_id = client.create_site( + GlobalNetworkId=gn_id, + Description="Test site #{i}", + )["Site"]["SiteId"] + site_ids.append(site_id) + sites_to_get = site_ids[0:NUM_TO_TEST] + resp = client.get_sites(GlobalNetworkId=gn_id, SiteIds=sites_to_get)["Sites"] + assert len(resp) == NUM_TO_TEST + + # Check each site by ID + for site in resp: + assert site["GlobalNetworkId"] == gn_id + assert site["SiteId"] in sites_to_get + + # Check all sites + all_sites = client.get_sites(GlobalNetworkId=gn_id)["Sites"] + assert len(all_sites) == NUM_SITES + + # Check invalid resource id returns empty list + resp = client.get_sites(GlobalNetworkId=gn_id, SiteIds=["invalid-id"]) + assert len(resp["Sites"]) == 0 + + +@mock_aws +def test_create_link(): + client = boto3.client("networkmanager") + gn_id = create_global_network(client) + link = client.create_link( + GlobalNetworkId=gn_id, + Description="Test link", + Type="AWS", + Bandwidth={"UploadSpeed": 100, "DownloadSpeed": 100}, + Provider="AWS", + SiteId="site-id", + Tags=[ + {"Key": "Name", "Value": "TestLink"}, + ], + )["Link"] + assert link["GlobalNetworkId"] == gn_id + assert link["Description"] == "Test link" + assert link["Type"] == "AWS" + assert link["Provider"] == "AWS" + assert link["SiteId"] == "site-id" + assert len(link["Tags"]) == 1 + + +@mock_aws +def test_get_links(): + NUM_LINKS = 4 + NUM_TO_TEST = 2 + client = boto3.client("networkmanager") + gn_id = create_global_network(client) + ids = [] + for i in range(NUM_LINKS): + id = client.create_link( + GlobalNetworkId=gn_id, + SiteId="site-id", + Description="Test link #{i}", + Bandwidth={"UploadSpeed": 100, "DownloadSpeed": 100}, + )["Link"]["LinkId"] + ids.append(id) + resources_to_get = [id for id in ids[0:NUM_TO_TEST]] + resp = client.get_links(GlobalNetworkId=gn_id, LinkIds=resources_to_get)["Links"] + assert len(resp) == NUM_TO_TEST + + # Check all links + all_links = client.get_links(GlobalNetworkId=gn_id)["Links"] + assert len(all_links) == NUM_LINKS + + # Check invalid resource id returns empty list + resp = client.get_links(GlobalNetworkId=gn_id, LinkIds=["invalid-id"]) + assert len(resp["Links"]) == 0 + + +@mock_aws +def test_delete_link(): + client = boto3.client("networkmanager") + gn_id = create_global_network(client) + link_id = client.create_link( + GlobalNetworkId=gn_id, + SiteId="site-id", + Description="Test link to delete", + Bandwidth={"UploadSpeed": 100, "DownloadSpeed": 100}, + )["Link"]["LinkId"] + + resp = client.delete_link(GlobalNetworkId=gn_id, LinkId=link_id)["Link"] + assert resp["State"] == "DELETING" + assert resp["LinkId"] == link_id + + +@mock_aws +def test_create_device(): + client = boto3.client("networkmanager") + gn_id = create_global_network(client) + device = client.create_device( + GlobalNetworkId=gn_id, + AWSLocation={ + "Zone": "us-west-2a", + "SubnetArn": "subnet-arn", + }, + Description="Test device", + )["Device"] + assert device["GlobalNetworkId"] == gn_id + + +@mock_aws +def test_get_devices(): + NUM_DEVICES = 4 + NUM_TO_TEST = 2 + client = boto3.client("networkmanager") + gn_id = create_global_network(client) + ids = [] + for i in range(NUM_DEVICES): + id = client.create_device( + GlobalNetworkId=gn_id, + AWSLocation={ + "Zone": "us-east-1", + "SubnetArn": "subnet-arn", + }, + Description=f"Test device #{i}", + )["Device"]["DeviceId"] + ids.append(id) + resources_to_get = [id for id in ids[0:NUM_TO_TEST]] + resp = client.get_devices(GlobalNetworkId=gn_id, DeviceIds=resources_to_get)[ + "Devices" + ] + assert len(resp) == NUM_TO_TEST + + # Check all devices + all_devices = client.get_devices(GlobalNetworkId=gn_id)["Devices"] + assert len(all_devices) == NUM_DEVICES + + # Check invalid resource id returns empty list + resp = client.get_devices(GlobalNetworkId=gn_id, DeviceIds=["invalid-id"]) + assert len(resp["Devices"]) == 0 + + +@mock_aws +def test_delete_device(): + client = boto3.client("networkmanager") + gn_id = create_global_network(client) + device_id = client.create_device( + GlobalNetworkId=gn_id, + Description="Test device to delete", + AWSLocation={ + "Zone": "us-west-2a", + "SubnetArn": "subnet-arn", + }, + )["Device"]["DeviceId"] + + resp = client.delete_device(GlobalNetworkId=gn_id, DeviceId=device_id)["Device"] + assert resp["State"] == "DELETING" + + # Check that the device is deleted + devices = client.get_devices(GlobalNetworkId=gn_id)["Devices"] + assert len(devices) == 0 + + +@mock_aws +def test_list_tags_for_resource(): + sample_tags = [ + {"Key": "Moto", "Value": "TestTag"}, + {"Key": "Owner", "Value": "Alice"}, + ] + client = boto3.client("networkmanager") + + # Global Network + g_network = client.create_global_network( + Description="Test global network", Tags=sample_tags + )["GlobalNetwork"] + gn_arn = g_network["GlobalNetworkArn"] + gn_id = g_network["GlobalNetworkId"] + resp = client.list_tags_for_resource(ResourceArn=gn_arn) + assert resp["TagList"] == sample_tags + + # Core Network + cn_arn = client.create_core_network( + Description="Test core network", GlobalNetworkId=gn_id, Tags=sample_tags + )["CoreNetwork"]["CoreNetworkArn"] + resp = client.list_tags_for_resource(ResourceArn=cn_arn) + assert resp["TagList"] == sample_tags + + # Site + site_arn = client.create_site( + GlobalNetworkId=gn_id, Description="Test site", Tags=sample_tags + )["Site"]["SiteArn"] + resp = client.list_tags_for_resource(ResourceArn=site_arn) + assert resp["TagList"] == sample_tags + + # Link + link_arn = client.create_link( + GlobalNetworkId=gn_id, + SiteId="site-id", + Description="Test link", + Bandwidth={"UploadSpeed": 100, "DownloadSpeed": 100}, + Tags=sample_tags, + )["Link"]["LinkArn"] + resp = client.list_tags_for_resource(ResourceArn=link_arn) + assert resp["TagList"] == sample_tags + + # Device + device_arn = client.create_device( + GlobalNetworkId=gn_id, + Description="Test device", + AWSLocation={ + "Zone": "us-west-2a", + "SubnetArn": "subnet-arn", + }, + Tags=sample_tags, + )["Device"]["DeviceArn"] + resp = client.list_tags_for_resource(ResourceArn=device_arn) + assert resp["TagList"] == sample_tags + + +# Exception testing + + +@mock_aws +def test_device_exceptions(): + client = boto3.client("networkmanager") + gn_id = create_global_network(client) + device_id = client.create_device( + GlobalNetworkId=gn_id, + Description="Test device", + AWSLocation={ + "Zone": "us-west-2a", + "SubnetArn": "subnet-arn", + }, + )["Device"]["DeviceId"] + + # Test invalid global_network_id for create resource + with pytest.raises(Exception): + client.create_device( + GlobalNetworkId="invalid-global-network-id", + AWSLocation={ + "Zone": "us-west-2a", + "SubnetArn": "subnet-arn", + }, + Description="Test device", + ) + + # Test invalid global_network_id for get + with pytest.raises(Exception): + client.get_devices( + GlobalNetworkId="invalid-global-network-id", DeviceIds=[device_id] + ) + + +@mock_aws +def test_site_exceptions(): + client = boto3.client("networkmanager") + + # Test invalid global_network_id for create resource + with pytest.raises(Exception): + client.create_site( + GlobalNetworkId="invalid-global-network-id", Description="Test site" + ) + + # Test invalid global_network_id for get + with pytest.raises(Exception): + client.get_devices( + GlobalNetworkId="invalid-global-network-id", SiteIds=["site-id"] + ) + + +@mock_aws +def test_link_exceptions(): + client = boto3.client("networkmanager") + + # Test invalid global_network_id for create resource + with pytest.raises(Exception): + client.create_link( + GlobalNetworkId="invalid-global-network-id", + SiteId="site-id", + Description="Test link", + Bandwidth={"UploadSpeed": 100, "DownloadSpeed": 100}, + ) + + # Test invalid global_network_id for get + with pytest.raises(Exception): + client.get_links( + GlobalNetworkId="invalid-global-network-id", LinkIds=["link-id"] + )