diff --git a/gso/services/_ipam.py b/gso/services/_ipam.py
index 84d9d7e2a7f30525a375d1f92f1e4128665b42ee..32591d0d33db773fc28cc0b7001bc0a092139675 100644
--- a/gso/services/_ipam.py
+++ b/gso/services/_ipam.py
@@ -87,6 +87,8 @@ def _find_networks(network_container=None, network=None, ip_version=4):
container.
Otherwise, if network is not None, find the specified network.
Otherwise find all networks.
+ A list of all found networks is returned (an HTTP 200 code
+ may be returned with an empty list.)
"""
assert ip_version in [4, 6]
oss = settings.load_oss_params()
@@ -105,7 +107,6 @@ def _find_networks(network_container=None, network=None, ip_version=4):
infoblox_params.password),
verify=False
)
- # TODO: propagate "network not found" error to caller
assert r.status_code >= 200 and r.status_code < 300, \
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
return r.json()
@@ -177,7 +178,7 @@ def _allocate_network(
return V6ServiceNetwork(v6=ipaddress.ip_network(allocated_network))
-def allocate_service_ipv4_network(service_type, comment="", extattrs={}
+def allocate_service_ipv4_network(service_type='', comment="", extattrs={}
) -> V4ServiceNetwork:
"""
Allocate IPv4 network within the container of the specified service type.
@@ -194,7 +195,7 @@ def allocate_service_ipv4_network(service_type, comment="", extattrs={}
extattrs)
-def allocate_service_ipv6_network(service_type, comment="", extattrs={}
+def allocate_service_ipv6_network(service_type='', comment="", extattrs={}
) -> V6ServiceNetwork:
"""
Allocate IPv6 network within the container of the specified service type.
@@ -211,7 +212,7 @@ def allocate_service_ipv6_network(service_type, comment="", extattrs={}
extattrs)
-def _find_next_available_ip(infoblox_params, network_ref):
+def _find_next_available_ip(infoblox_params, network_ref=''):
"""
Find the next available IP address from a network given its ref.
Returns "NETWORK_FULL" if there's no space in the network.
@@ -236,10 +237,10 @@ def _find_next_available_ip(infoblox_params, network_ref):
return received_ip[0]
-def _allocate_host(hostname=None,
+def _allocate_host(hostname='',
addrs=None,
networks=None,
- cname_aliases=None,
+ cname_aliases=[],
dns_view="default",
extattrs={}
) -> Union[HostAddresses, str]:
@@ -265,15 +266,15 @@ def _allocate_host(hostname=None,
# Find the next available IP address in each network
network_info = _find_networks(network=ipv4_network, ip_version=4)
- assert len(network_info) == 1, \
- "IPv4 Network does not exist. Create it first."
+ if len(network_info) != 1:
+ return "IPV4_NETWORK_NOT_FOUND"
assert '_ref' in network_info[0]
ipv4_addr = _find_next_available_ip(infoblox_params,
network_info[0]["_ref"])
network_info = _find_networks(network=ipv6_network, ip_version=6)
- assert len(network_info) == 1, \
- "IPv6 Network does not exist. Create it first."
+ if len(network_info) != 1:
+ return "IPV6_NETWORK_NOT_FOUND"
assert '_ref' in network_info[0]
ipv6_addr = _find_next_available_ip(infoblox_params,
network_info[0]["_ref"])
@@ -346,8 +347,8 @@ def _allocate_host(hostname=None,
v6=ipaddress.ip_address(ipv6_addr))
-def allocate_service_host(hostname=None,
- service_type=None,
+def allocate_service_host(hostname='',
+ service_type='',
service_networks: ServiceNetworks = None,
host_addresses: HostAddresses = None,
cname_aliases=None,
@@ -415,9 +416,9 @@ def allocate_service_host(hostname=None,
ipv4_network = str(oss_ipv4_networks[0])
ipv6_network = str(oss_ipv6_networks[0])
+ ipv4_network_index = 0
+ ipv6_network_index = 0
while True:
- ipv4_network_index = 0
- ipv6_network_index = 0
network_tuple = (ipv4_network, ipv6_network)
host = _allocate_host(hostname=hostname+domain_name,
networks=network_tuple,
@@ -425,15 +426,19 @@ def allocate_service_host(hostname=None,
dns_view=dns_view,
extattrs=extattrs)
- if "NETWORK_FULL" not in host:
+ if "NETWORK_FULL" not in host and "NETWORK_NOT_FOUND" not in host:
break
elif "IPV4" in host:
ipv4_network_index += 1
+ assert oss_ipv4_networks, \
+ "No available space in any IPv4 network for this service."
assert ipv4_network_index < len(oss_ipv4_networks), \
"No available space in any IPv4 network for this service."
ipv4_network = str(oss_ipv4_networks[ipv4_network_index])
- else: # IPV6 in host
+ else: # "IPV6" in host
ipv6_network_index += 1
+ assert oss_ipv6_networks, \
+ "No available space in any IPv6 network for this service."
assert ipv6_network_index < len(oss_ipv6_networks), \
"No available space in any IPv6 network for this service."
ipv6_network = str(oss_ipv6_networks[ipv6_network_index])
@@ -462,7 +467,10 @@ def allocate_service_host(hostname=None,
dns_view=dns_view,
extattrs=extattrs
)
- assert "NETWORK_FULL" not in host
+ assert "NETWORK_FULL" not in host, \
+ "Network is full."
+ assert "NETWORK_NOT_FOUND" not in host, \
+ "Network does not exist. Create it first."
elif host_addresses:
# IPv4
@@ -495,7 +503,7 @@ def allocate_service_host(hostname=None,
return host
-def delete_service_network(network, service_type=None
+def delete_service_network(ipnetwork=None, service_type=''
) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
"""
Delete IPv4 or IPv6 network by CIDR.
@@ -509,8 +517,8 @@ def delete_service_network(network, service_type=None
assert hasattr(ipam_params, service_type) \
and service_type != 'INFOBLOX', "Invalid service type."
+ network = str(ipnetwork)
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
@@ -561,6 +569,75 @@ def delete_service_network(network, service_type=None
return V6ServiceNetwork(v6=ipaddress.ip_network(network_address))
+# def delete_service_host(
+ # hostname='',
+ # host_addresses: HostAddresses = None,
+ # cname_aliases=[],
+ # service_type=''
+# ) -> Union[V4HostAddress, V6HostAddress]:
+ # """
+ # Delete IPv4 or IPv6 host by its address.
+ # """
+ # oss = settings.load_oss_params()
+ # assert oss.IPAM.INFOBLOX
+ # infoblox_params = oss.IPAM.INFOBLOX
+
+ # ip_version = _ip_addr_version(addr)
+ # ip_param = 'ipv4addr' if ip_version == 4 else 'ipv6addr'
+
+ # # Find host record reference
+ # r = requests.get(
+ # f'{_wapi(infoblox_params)}/record:host',
+ # params={ip_param: addr},
+ # auth=HTTPBasicAuth(infoblox_params.username,
+ # infoblox_params.password),
+ # verify=False
+ # )
+ # host_data = r.json()
+ # assert len(host_data) == 1, "Host does not exist."
+ # assert '_ref' in host_data[0]
+ # host_ref = host_data[0]['_ref']
+
+ # # Delete it
+ # r = requests.delete(
+ # f'{_wapi(infoblox_params)}/{host_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}"
+
+ # # Also find and delete the associated dns a/aaaa record
+ # endpoint = 'record:a' if ip_version == 4 else 'record:aaaa'
+
+ # r = requests.get(
+ # f'{_wapi(infoblox_params)}/{endpoint}',
+ # params={ip_param: addr},
+ # auth=HTTPBasicAuth(infoblox_params.username,
+ # infoblox_params.password),
+ # verify=False
+ # )
+ # dns_data = r.json()
+ # assert len(dns_data) == 1, "DNS record does not exist."
+ # assert '_ref' in dns_data[0]
+ # dns_ref = dns_data[0]['_ref']
+
+ # r = requests.delete(
+ # f'{_wapi(infoblox_params)}/{dns_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}"
+
+ # if ip_version == 4:
+ # return V4HostAddress(v4=addr)
+ # else:
+ # return V6HostAddress(v6=addr)
+
+
"""
Below methods are not used for supported outside calls
"""
@@ -622,70 +699,6 @@ def _get_network_capacity(network=None):
return utilization
-def _delete_host_by_ip(addr) -> Union[V4HostAddress, V6HostAddress]:
- """
- Delete IPv4 or IPv6 host by its address.
- """
- oss = settings.load_oss_params()
- assert oss.IPAM.INFOBLOX
- infoblox_params = oss.IPAM.INFOBLOX
-
- ip_version = _ip_addr_version(addr)
- ip_param = 'ipv4addr' if ip_version == 4 else 'ipv6addr'
-
- # Find host record reference
- r = requests.get(
- f'{_wapi(infoblox_params)}/record:host',
- params={ip_param: addr},
- auth=HTTPBasicAuth(infoblox_params.username,
- infoblox_params.password),
- verify=False
- )
- host_data = r.json()
- assert len(host_data) == 1, "Host does not exist."
- assert '_ref' in host_data[0]
- host_ref = host_data[0]['_ref']
-
- # Delete it
- r = requests.delete(
- f'{_wapi(infoblox_params)}/{host_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}"
-
- # Also find and delete the associated dns a/aaaa record
- endpoint = 'record:a' if ip_version == 4 else 'record:aaaa'
-
- r = requests.get(
- f'{_wapi(infoblox_params)}/{endpoint}',
- params={ip_param: addr},
- auth=HTTPBasicAuth(infoblox_params.username,
- infoblox_params.password),
- verify=False
- )
- dns_data = r.json()
- assert len(dns_data) == 1, "DNS record does not exist."
- assert '_ref' in dns_data[0]
- dns_ref = dns_data[0]['_ref']
-
- r = requests.delete(
- f'{_wapi(infoblox_params)}/{dns_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}"
-
- if ip_version == 4:
- return V4HostAddress(v4=addr)
- else:
- return V6HostAddress(v6=addr)
-
-
def _get_network_usage_status(network):
"""
Get status and usage fields of all hosts in the specified ipv4 or ipv6
diff --git a/gso/services/ipam.py b/gso/services/ipam.py
index cd8d2588d8a7953a6721820370aebd189a93c736..5e8273027983a2a633e72da4272658887dcda36d 100644
--- a/gso/services/ipam.py
+++ b/gso/services/ipam.py
@@ -31,7 +31,7 @@ class HostAddresses(BaseSettings):
v6: ipaddress.IPv6Address
-def new_service_networks(service_type,
+def new_service_networks(service_type='',
comment="",
extattrs={}) -> ServiceNetworks:
v4_service_network = _ipam.allocate_service_ipv4_network(
@@ -44,7 +44,7 @@ def new_service_networks(service_type,
def new_service_host(hostname,
- service_type,
+ service_type='',
service_networks: ServiceNetworks = None,
host_addresses: HostAddresses = None,
cname_aliases=None,
@@ -58,14 +58,24 @@ def new_service_host(hostname,
extattrs=extattrs)
-def delete_service_network(network, service_type
- ) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
+def delete_service_network(
+ network: ipaddress.ip_network = None, service_type=''
+) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
return _ipam.delete_service_network(
- network=network,
+ ipnetwork=network,
service_type=service_type
)
+def delete_service_host(
+ hostname='',
+ host_addresses: HostAddresses = None,
+ cname_aliases=[],
+ service_type=''
+) -> HostAddresses:
+ return None
+
+
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.
diff --git a/test/test_ipam.py b/test/test_ipam.py
index e839c00ae6db7e2a2d81ddb4264c622f140a43c7..2a2659c04ab6d179174fbb7c22fa9e4633172946 100644
--- a/test/test_ipam.py
+++ b/test/test_ipam.py
@@ -86,7 +86,7 @@ def test_new_service_host(data_config_filename):
responses.add(
method=responses.GET,
- url=re.compile(r'.*/wapi.*/ipv6network.*'),
+ url=re.compile(r'.*/wapi.*/ipv6network.*dead.*beef.*'),
json=[
{
"_ref": "ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default", # noqa: E501
@@ -96,6 +96,12 @@ def test_new_service_host(data_config_filename):
]
)
+ responses.add(
+ method=responses.GET,
+ url=re.compile(r'.*/wapi.*/ipv6network.*beef.*dead.*'),
+ json=[]
+ )
+
responses.add(
method=responses.POST,
url=re.compile(r'.*/wapi.*/network/.*10.255.255.*?_function=next_available_ip&num=1$'), # noqa: E501
@@ -193,6 +199,18 @@ def test_new_service_host(data_config_filename):
)
assert service_hosts is None
+ # test host creation that should return a network not exist error
+ with pytest.raises(AssertionError):
+ service_hosts = ipam.new_service_host(
+ hostname='test',
+ service_type='TRUNK',
+ service_networks=ipam.ServiceNetworks(
+ v4=ipaddress.ip_network('10.255.255.20/32'),
+ v6=ipaddress.ip_network('beef:dead::18/128')
+ )
+ )
+ assert service_hosts is None
+
@responses.activate
def test_delete_service_network(data_config_filename):
@@ -269,20 +287,31 @@ def test_delete_service_network(data_config_filename):
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
+ service_network = ipam.delete_service_network(
+ network=ipaddress.ip_network('10.255.255.0/26'), service_type='LO'
+ )
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
+ service_network = ipam.delete_service_network(
+ network=ipaddress.ip_network('10.255.255.20/32'),
+ service_type='LO'
+ )
assert service_network is None
- service_network = ipam.delete_service_network(network='dead:beef::18/128', service_type='TRUNK') # noqa: E501
+ service_network = ipam.delete_service_network(
+ network=ipaddress.ip_network('dead:beef::18/128'),
+ service_type='TRUNK'
+ )
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
+ service_network = ipam.delete_service_network(
+ network=ipaddress.ip_network('beef:dead::18/128'),
+ service_type='TRUNK'
+ )
assert service_network is None