NAT-212/213: tox

gso/services/ipam.py

@@ -138,11 +138,18 @@ def assert_network_in_service(
 
 
 def find_networks(network_container: Optional[str] = "", network: Optional[str] = "", ip_version: int = 4):
-    """If network_container is not None, find all networks within the specified 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).
+    """Get all networks optinally filtering by a container or by a network.
+
+    Args:
+    ----
+    network_container (str, optional): container to filter by
+    network (str, optional): network to filter by
+    ip_version (int): 4 or 6
+
+    Returns:
+    -------
+    (list) all found networks mathing the args, which may be empty.
+
     """
     assert ip_version in [4, 6]
     oss = settings.load_oss_params()
@@ -158,7 +165,7 @@ def find_networks(network_container: Optional[str] = "", network: Optional[str]
         f"{wapi(infoblox_params)}/{endpoint}",
         params=params,
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
-        verify=False,
+        verify=False,  # noqa: S501
         timeout=REQUESTS_TIMEOUT
     )
     assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
@@ -205,7 +212,7 @@ def allocate_network_inner(
             json=req_payload,
             auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
             headers={"content-type": "application/json"},
-            verify=False,
+            verify=False,  # noqa: S501
             timeout=REQUESTS_TIMEOUT
         )
         if not match_error_code(response=r, error_code=IPAMErrors.CONTAINER_FULL):
@@ -226,7 +233,19 @@ def allocate_network_inner(
 
 
 def allocate_ipv4_network(service_type: str = "", comment: Optional[str] = "", extattrs: Optional[dict] = None) -> V4ServiceNetwork:
-    """Allocate IPv4 network within the container of the specified service type."""
+    """Allocate IPv4 network within the container of the specified service type.
+
+    Args:
+    ----
+    service_type (str): the name of the service type (e.g. "TRUNK")
+    comment (str, optional): a custom comment to write in the comment field in IPAM
+    extattrs (dict, optional): any extensible attributes to add in IPAM (e.g. "Site": {"value": "dummy"})
+
+    Returns:
+    -------
+    (V4ServiceNetwork): the allocated network
+
+    """
     if extattrs is None:
         extattrs = {}
     oss = settings.load_oss_params()
@@ -237,7 +256,19 @@ def allocate_ipv4_network(service_type: str = "", comment: Optional[str] = "", e
 
 
 def allocate_ipv6_network(service_type: str = "", comment: Optional[str] = "", extattrs: Optional[dict] = None) -> V6ServiceNetwork:
-    """Allocate IPv6 network within the container of the specified service type."""
+    """Allocate IPv6 network within the container of the specified service type.
+
+    Args:
+    ----
+    service_type (str): the name of the service type (e.g. "TRUNK")
+    comment (str, optional): a custom comment to write in the comment field in IPAM
+    extattrs (dict, optional): any extensible attributes to add in IPAM (e.g. "Site": {"value": "dummy"})
+
+    Returns:
+    -------
+    (V4ServiceNetwork): the allocated network
+
+    """
     if extattrs is None:
         extattrs = {}
     oss = settings.load_oss_params()
@@ -266,13 +297,21 @@ def allocate_networks(
 
 def find_next_available_ip(infoblox_params, network_ref: str = ""):
     """Find the next available IP address from a network given its ref.
-    Returns "NETWORK_FULL" if there's no space in the network.
-    Otherwise returns the next available IP address in the network.
+
+    Args:
+    ----
+    infoblox_params (settings.InfoBloxParams): infoblox params
+    network_ref (str): the network to find the next available ip, in InfoBlox reference format
+
+    Returns:
+    -------
+    (str): next available ip in the network, or "NETWORK_FULL" if there's no space in the network
+
     """
     r = requests.post(
         f"{wapi(infoblox_params)}/{network_ref}?_function=next_available_ip&num=1",
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
-        verify=False,
+        verify=False,  # noqa: S501
         timeout=REQUESTS_TIMEOUT
     )
 
@@ -286,7 +325,7 @@ def find_next_available_ip(infoblox_params, network_ref: str = ""):
     return received_ip[0]
 
 
-def allocate_host_inner(
+def allocate_host_inner(  # noqa: max-complexity=15
     hostname: str = "",
     addrs: Optional[Tuple] = None,
     networks: Optional[Tuple] = None,
@@ -294,14 +333,7 @@ def allocate_host_inner(
     dns_view: Optional[str] = "default",
     extattrs: Optional[dict] = None
 ) -> Union[HostAddresses, str]:
-    """If networks is not None, allocate host in those networks.
-    Otherwise if addrs is not None, allocate host with those addresses.
-    hostname parameter must be full name including domain name.
-    Return "IPV4_NETWORK_FULL" or "IPV6_NETWORK_FULL"
-    if couldn't allocate host due to requested network being full.
-    Return "IPV4_NETWORK_NOT_FOUND" or "IPV6_NETWORK_NOT_FOUND"
-    if couldn't allocate host due to requested network not existing.
-    """
+
     # TODO: should hostnames be unique
     # (i.e. fail if hostname already exists in this domain/service)?
     if cname_aliases is None:
@@ -313,6 +345,7 @@ def allocate_host_inner(
     assert oss.IPAM.INFOBLOX
     infoblox_params = oss.IPAM.INFOBLOX
 
+    # If networks is not None, allocate host in those networks.
     if networks:
         ipv4_network = networks[0]
         ipv6_network = networks[1]
@@ -320,6 +353,7 @@ def allocate_host_inner(
         assert ip_network_version(ipv6_network) == 6
 
         # Find the next available IP address in each network
+        # If requested error doesn't exist, return error
         network_info = find_networks(network=ipv4_network, ip_version=4)
         if len(network_info) != 1:
             return "IPV4_NETWORK_NOT_FOUND"
@@ -339,12 +373,14 @@ def allocate_host_inner(
             if ipv6_addr == "NETWORK_FULL":
                 return "IPV6_NETWORK_FULL"
 
+    # Otherwise if addrs is not None, allocate host with those addresses.
     else:
         ipv4_addr = addrs[0]
         ipv6_addr = addrs[1]
         assert ip_addr_version(ipv4_addr) == 4
         assert ip_addr_version(ipv6_addr) == 6
 
+    # hostname parameter must be full name including domain name.
     req_payload = {
         "ipv4addrs": [{"ipv4addr": ipv4_addr}],
         "ipv6addrs": [{"ipv6addr": ipv6_addr}],
@@ -358,7 +394,7 @@ def allocate_host_inner(
         f"{wapi(infoblox_params)}/record:host",
         json=req_payload,
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
-        verify=False,
+        verify=False,  # noqa: S501
         timeout=REQUESTS_TIMEOUT
     )
     assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
@@ -374,7 +410,7 @@ def allocate_host_inner(
                 f"{wapi(infoblox_params)}/record:cname",
                 json=cname_req_payload,
                 auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
-                verify=False,
+                verify=False,  # noqa: S501
                 timeout=REQUESTS_TIMEOUT
             )
             assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
@@ -383,29 +419,29 @@ def allocate_host_inner(
     return HostAddresses(v4=ipaddress.ip_address(ipv4_addr), v6=ipaddress.ip_address(ipv6_addr))
 
 
-def allocate_host(
+def allocate_host(  # noqa: max-complexity=15
     hostname: str = "",
     service_type: str = "",
     service_networks: Optional[ServiceNetworks] = None,
-    host_addresses: Optional[ServiceNetworks] = None,
+    host_addresses: Optional[HostAddresses] = None,
     cname_aliases: Optional[list] = None,
     extattrs: Optional[dict] = None,
 ) -> HostAddresses:
-    """Allocate host record with both IPv4 and IPv6 address, and respective DNS
-    A and AAAA records.
-    - If service_networks is provided, and it's in a valid container,
-    that one is used.
-    - If service_networks is not provided, and host_addresses is provided,
-    those specific addresses are used.
-    - If neither is provided:
-        - If service has configured containers, new ipv4 and ipv6 networks are
-        created and those are used. Note that in this case extattrs is for the
-        hosts and not for the networks.
-        - If service doesn't have configured containers and has configured
-        networks instead, the configured networks are used (they are filled up
-        in order of appearance in the configuration file).
-    The domain name is taken from the service type and appended to the
-    specified hostname.
+    """Allocate host record with both IPv4 and IPv6 address, and respective DNS A and AAAA records.
+
+    Args:
+    ----
+    hostname (str): hostname of the host (without domain name, which is taken from the service type)
+    service_type (str): the name of the service type (e.g. "TRUNK")
+    service_networks (ServiceNetworks, optional): ipv4 and ipv6 network to allocate host, must be in a valid container for the service type
+    host_addresses (HostAddresses, optional): ipv4 and ipv6 addresses to allocate host (service_networks has precedence)
+    cname_aliases (list, optional): to create cname records in addition to the host record
+    extattrs (dict, optional): any extensible attributes to add in IPAM (e.g. "Site": {"value": "dummy"})
+
+    Returns:
+    -------
+    (HostAddresses): ipv4 and ipv6 addresses of the allocated host
+
     """
     if cname_aliases is None:
         cname_aliases = []
@@ -432,6 +468,11 @@ def allocate_host(
     if cname_aliases:
         cname_aliases = [alias + domain_name for alias in cname_aliases]
 
+    # When neither service_networks not host_addresses are provided:
+    # If service has configured containers, new ipv4 and ipv6 networks are created and those are used.
+    # Note that in this case extattrs is for the hosts and not for the networks.
+    # If service doesn't have configured containers and has configured networks instead, the configured
+    # networks are used (they are filled up in order of appearance in the configuration file).
     if not service_networks and not host_addresses:
         if oss_ipv4_containers and oss_ipv6_containers:
             # This service has configured containers.
@@ -555,7 +596,7 @@ def delete_network(
     r = requests.delete(
         f'{wapi(infoblox_params)}/{network_info[0]["_ref"]}',
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
-        verify=False,
+        verify=False,  # noqa: S501
         timeout=REQUESTS_TIMEOUT
     )
     assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
@@ -576,6 +617,7 @@ def delete_host(
     service_type: str = ""
 ) -> HostAddresses:
     """Delete host record and associated CNAME records.
+
     All arguments passed to this function must match together a host record in
     IPAM, and all CNAME records associated to it must also be passed exactly.
     """
@@ -617,7 +659,7 @@ def delete_host(
             "view": dns_view,
         },
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
-        verify=False,
+        verify=False,  # noqa: S501
         timeout=REQUESTS_TIMEOUT
     )
     host_data = r.json()
@@ -633,7 +675,7 @@ def delete_host(
             "view": dns_view,
         },
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
-        verify=False,
+        verify=False,  # noqa: S501
         timeout=REQUESTS_TIMEOUT
     )
     cname_data = r.json()
@@ -645,7 +687,7 @@ def delete_host(
     r = requests.delete(
         f"{wapi(infoblox_params)}/{host_ref}",
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
-        verify=False,
+        verify=False,  # noqa: S501
         timeout=REQUESTS_TIMEOUT
     )
     assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
@@ -656,7 +698,7 @@ def delete_host(
         r = requests.delete(
             f"{wapi(infoblox_params)}/{cname_ref}",
             auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
-            verify=False,
+            verify=False,  # noqa: S501
             timeout=REQUESTS_TIMEOUT
         )
         assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
@@ -669,8 +711,8 @@ def validate_network(
     network: ipaddress.ip_network = None
 ) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
     """Validate IPv4 or IPv6 network.
-    Check if the specified network exist, and, if it does,
-    check if its comment field contains gso_subscription_id.
+
+    Check if the specified network exist, and, if it does, check if its comment field contains gso_subscription_id.
     Returns the network if validation successful.
     """
     assert network, "No network specified to validate."
@@ -693,6 +735,7 @@ def validate_host(
     service_type: str = ""
 ) -> HostAddresses:
     """Validate host.
+
     Check if all arguments passed to this function match together a host record in
     IPAM, and all CNAME records associated to it also match exactly.
     Returns the host if validation successful.
@@ -721,7 +764,7 @@ def validate_host(
             "view": dns_view,
         },
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
-        verify=False,
+        verify=False,  # noqa: S501
         timeout=REQUESTS_TIMEOUT
     )
     host_data = r.json()
@@ -736,7 +779,7 @@ def validate_host(
             "view": dns_view,
         },
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
-        verify=False,
+        verify=False,  # noqa: S501
         timeout=REQUESTS_TIMEOUT
     )
     cname_data = r.json()