NAT-200: delete network in ipam

gso/services/_ipam.py

@@ -172,9 +172,9 @@ def _allocate_network(
     assert 'network' in r.json()
     allocated_network = r.json()['network']
     if ip_version == 4:
-        return V4ServiceNetwork(v4=allocated_network)
+        return V4ServiceNetwork(v4=ipaddress.ip_network(allocated_network))
     else:
-        return V6ServiceNetwork(v6=allocated_network)
+        return V6ServiceNetwork(v6=ipaddress.ip_network(allocated_network))
 
 
 def allocate_service_ipv4_network(service_type, comment="", extattrs={}
@@ -495,6 +495,72 @@ def allocate_service_host(hostname=None,
     return host
 
 
+def delete_service_network(network, service_type=None
+                           ) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
+    """
+    Delete IPv4 or IPv6 network by CIDR.
+    """
+    oss = settings.load_oss_params()
+    assert oss.IPAM
+    ipam_params = oss.IPAM
+    assert ipam_params.INFOBLOX
+    infoblox_params = ipam_params.INFOBLOX
+
+    assert hasattr(ipam_params, service_type) \
+        and service_type != 'INFOBLOX', "Invalid service type."
+
+    ip_version = _ip_network_version(network)
+    ipnetwork = ipaddress.ip_network(network)
+
+    # Ensure that the network to be deleted is under the service type.
+    # Otherwise user is not allowed to delete it
+    if ip_version == 4:
+        oss_ipv4_containers = getattr(ipam_params, service_type).V4.containers
+        oss_ipv4_networks = getattr(ipam_params, service_type).V4.networks
+        if oss_ipv4_containers:
+            assert any(ipnetwork.subnet_of(oss_ipv4_container)
+                       for oss_ipv4_container in oss_ipv4_containers), \
+                       "Can't delete: network doesn't belong to service type."
+        else:
+            assert ipnetwork in oss_ipv4_networks, \
+                "Can't delete: network doesn't belong to service type."
+
+    else:
+        oss_ipv6_containers = getattr(ipam_params, service_type).V6.containers
+        oss_ipv6_networks = getattr(ipam_params, service_type).V6.networks
+        if oss_ipv6_containers:
+            assert any(ipnetwork.subnet_of(oss_ipv6_container)
+                       for oss_ipv6_container in oss_ipv6_containers), \
+                       "Can't delete: network doesn't belong to service type."
+        else:
+            assert ipnetwork in oss_ipv6_networks, \
+                "Can't delete: network doesn't belong to service type."
+
+    network_info = _find_networks(network=network, ip_version=ip_version)
+    assert len(network_info) == 1, "Network does not exist."
+    assert '_ref' in network_info[0]
+
+    r = requests.delete(
+        f'{_wapi(infoblox_params)}/{network_info[0]["_ref"]}',
+        auth=HTTPBasicAuth(infoblox_params.username,
+                           infoblox_params.password),
+        verify=False
+    )
+    assert r.status_code >= 200 and r.status_code < 300, \
+        f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
+
+    # Extract ipv4/ipv6 address from the network reference obtained in the
+    # response
+    r_text = r.text
+    print(r_text)
+    network_address = ipaddress.ip_network(
+        r_text.rsplit("/", 1)[0].split(":")[1].replace("%3A", ":"))
+    if ip_version == 4:
+        return V4ServiceNetwork(v4=ipaddress.ip_network(network_address))
+    else:
+        return V6ServiceNetwork(v6=ipaddress.ip_network(network_address))
+
+
 """
 Below methods are not used for supported outside calls
 """
@@ -556,43 +622,6 @@ def _get_network_capacity(network=None):
     return utilization
 
 
-def _delete_network(network) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
-    """
-    Delete IPv4 or IPv6 network by CIDR.
-    """
-    # TODO: should we check that there are no hosts in this network before
-    # deleting? Deleting a network deletes the hosts in it, but not the
-    # associated DNS records.
-    oss = settings.load_oss_params()
-    assert oss.IPAM.INFOBLOX
-    infoblox_params = oss.IPAM.INFOBLOX
-
-    ip_version = _ip_network_version(network)
-
-    network_info = _find_networks(network=network, ip_version=ip_version)
-    assert len(network_info) == 1, "Network does not exist."
-    assert '_ref' in network_info[0]
-
-    r = requests.delete(
-        f'{_wapi(infoblox_params)}/{network_info[0]["_ref"]}',
-        auth=HTTPBasicAuth(infoblox_params.username,
-                           infoblox_params.password),
-        verify=False
-    )
-    assert r.status_code >= 200 and r.status_code < 300, \
-        f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
-
-    # Extract ipv4/ipv6 address from the network reference obtained in the
-    # response
-    r_json = r.json()
-    network_address = ipaddress.ip_network(
-        r_json.rsplit("/", 1)[0].split(":")[1].replace("%3A", ":"))
-    if ip_version == 4:
-        return V4ServiceNetwork(v4=network_address)
-    else:
-        return V6ServiceNetwork(v6=network_address)
-
-
 def _delete_host_by_ip(addr) -> Union[V4HostAddress, V6HostAddress]:
     """
     Delete IPv4 or IPv6 host by its address.

gso/services/ipam.py

 import ipaddress
 from pydantic import BaseSettings
+from typing import Union
 
 from gso.services import _ipam
 
@@ -57,6 +58,14 @@ def new_service_host(hostname,
         extattrs=extattrs)
 
 
+def delete_service_network(network, service_type
+                           ) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
+    return _ipam.delete_service_network(
+        network=network,
+        service_type=service_type
+    )
+
+
 if __name__ == '__main__':
     # sample call flow to allocate two loopback interfaces and a trunk service
     # new_service_host can be called passing networks, addresses, or nothing.
@@ -74,11 +83,11 @@ if __name__ == '__main__':
     # loA_v4_host_address = ipaddress.ip_address('10.255.255.0')
     # loA_v6_host_address = ipaddress.ip_address('dead:beef::0')
     # loA_host_addresses = HostAddresses(v4=loA_v4_host_address,
-                                       # v6=loA_v6_host_address)
+    #                                    v6=loA_v6_host_address)
     # new_service_host(hostname=hostname_A+"_LO",
-                     # host_addresses=loA_host_addresses,
-                     # cname_aliases=["alias1.hA", "alias2.hA"],
-                     # service_type='LO')
+    #                  host_addresses=loA_host_addresses,
+    #                  cname_aliases=["alias1.hA", "alias2.hA"],
+    #                  service_type='LO')
     new_service_host(hostname=hostname_A+"_LO",
                      cname_aliases=["alias1.hA", "alias2.hA"],
                      service_type='LO')

test/test_ipam.py

@@ -70,7 +70,6 @@ def test_new_service_host(data_config_filename):
                 "network_view": "default"
             }
         ]
-
     )
 
     responses.add(
@@ -83,7 +82,6 @@ def test_new_service_host(data_config_filename):
                 "network_view": "default"
             }
         ]
-
     )
 
     responses.add(
@@ -194,3 +192,97 @@ def test_new_service_host(data_config_filename):
             )
         )
         assert service_hosts is None
+
+
+@responses.activate
+def test_delete_service_network(data_config_filename):
+
+    responses.add(
+        method=responses.GET,
+        url=re.compile(r'.*/wapi.*/network.*10.255.255.0.*'),
+        json=[
+            {
+                "_ref": "network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.0/26/default",   # noqa: E501
+                "network": "10.255.255.0/26",
+                "network_view": "default"
+            }
+        ]
+    )
+
+    responses.add(
+        method=responses.GET,
+        url=re.compile(r'.*/wapi.*/network.*10.255.255.20.*'),
+        json=[
+            {
+                "_ref": "network/ZG5zLm5Gd0VHQkRQUjMzLjMwNzIuMzE1LzAyLzI:100.255.255.20/32/default",   # noqa: E501
+                "network": "100.255.255.20/32",
+                "network_view": "default"
+            }
+        ]
+    )
+
+    responses.add(
+        method=responses.GET,
+        url=re.compile(r'.*/wapi.*/ipv6network.*dead.*beef.*'),
+        json=[
+            {
+                "_ref": "ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default",   # noqa: E501
+                "network": "dead:beef::18/128",
+                "network_view": "default"
+            }
+        ]
+    )
+
+    responses.add(
+        method=responses.GET,
+        url=re.compile(r'.*/wapi.*/ipv6network.*beef.*dead.*'),
+        json=[
+            {
+                "_ref": "ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:beef%3Adead%3A%3A18/128/default",   # noqa: E501
+                "network": "beef:dead::18/128",
+                "network_view": "default"
+            }
+        ]
+    )
+
+    responses.add(
+        method=responses.DELETE,
+        url=re.compile(r'.*/wapi.*/network.*10.255.255.0.*'),
+        body="network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.0/26/default"   # noqa: E501
+    )
+
+    responses.add(
+        method=responses.DELETE,
+        url=re.compile(r'.*/wapi.*/network.*100.255.255.*'),
+        body="network/ZG5zLm5Gd0VHQkRQUjMzLjMwNzIuMzE1LzAyLzI:100.255.255.20/32/default"   # noqa: E501
+    )
+
+    responses.add(
+        method=responses.DELETE,
+        url=re.compile(r'.*/wapi.*/ipv6network.*dead.*beef.*'),
+        body="ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default"   # noqa: E501
+    )
+
+    responses.add(
+        method=responses.DELETE,
+        url=re.compile(r'.*/wapi.*/ipv6network.*beef.*dead.*'),
+        body="ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:beef%3Adead%3A%3A18/128/default"   # noqa: E501
+    )
+
+    service_network = ipam.delete_service_network(network='10.255.255.0/26', service_type='LO')   # noqa: E501
+    assert service_network == ipam.V4ServiceNetwork(
+        v4=ipaddress.ip_network('10.255.255.0/26')
+    )
+
+    with pytest.raises(AssertionError):
+        service_network = ipam.delete_service_network(network='10.255.255.20/32', service_type='LO')   # noqa: E501
+        assert service_network is None
+
+    service_network = ipam.delete_service_network(network='dead:beef::18/128', service_type='TRUNK')   # noqa: E501
+    assert service_network == ipam.V6ServiceNetwork(
+        v6=ipaddress.ip_network('dead:beef::18/128')
+    )
+
+    with pytest.raises(AssertionError):
+        service_network = ipam.delete_service_network(network='beef:dead::18/128', service_type='TRUNK')   # noqa: E501
+        assert service_network is None