Unify network_overlaps and is_unique_ip

Author: TuxPowered42

serveradmin/serverdb/models.py

@@ -101,24 +101,23 @@ def is_ip_address(ip_interface: Union[IPv4Interface, IPv6Interface]) -> None:
 def is_unique_ip(
     ip_interface: Union[IPv4Interface, IPv6Interface],
     server: "Server",
-    exclude_addr_types: list[str],
-    attribute_id: Optional[str] = None,
+    addr_types: list[str],
+    is_net: bool,
+    attribute_id: Optional[int] = None,
 ) -> None:
-    """Validate if IPv4/IPv6 address is unique
+    """Validate if IPv4/IPv6 interface is unique
 
     Raises a ValidationError if intern_ip or any other attribute of type inet
     with this ip_address already exists.
 
     :param ip_interface:
-    :param object_id:
+    :param server:
+    :param addr_types:
+    :param is_net:
     :param attribute_id:
     :return:
     """
 
-    # We avoid querying the duplicate hosts here and giving the user
-    # detailed information because checking with exists is cheaper than
-    # querying the server and this is a validation and should be fast.
-
     # Always exclude the current object_id from the query because we allow
     # duplication of data between the legacy (intern_ip, primary_ip6) and
     # the modern (ipv4, ipv6) attributes.
@@ -130,7 +129,18 @@ def is_unique_ip(
     duplicates = []
     object_id = server.server_id
 
-    # TODO: Make "aid" mandatory when intern_ip is gone.
+    if is_net:
+        servertype_inet_attr_q = Q(server__servertype=server.servertype_id)
+        servertype_intern_ip_q = Q(servertype=server.servertype_id)
+        inet_attr_q = Q(value__net_overlaps=ip_interface)
+        intern_ip_q = Q(intern_ip__net_overlaps=ip_interface)
+    else:
+        servertype_inet_attr_q = Q(True)
+        servertype_intern_ip_q = Q(True)
+        inet_attr_q = Q(value=ip_interface)
+        intern_ip_q = Q(intern_ip=ip_interface)
+
+    # TODO: Make attribute_id mandatory when intern_ip is gone.
     if attribute_id:
         object_attribute_condition = Q(server_id=object_id) | ~Q(
             attribute_id=attribute_id
@@ -139,14 +149,23 @@ def is_unique_ip(
         object_attribute_condition = Q(server_id=object_id)
 
     # TODO: Remove intern_ip.
-    for d in Server.objects.filter(intern_ip=ip_interface).exclude(
-        Q(servertype__ip_addr_type__in=exclude_addr_types) | Q(server_id=object_id)
+    for d in Server.objects.filter(
+        Q(
+            servertype_intern_ip_q
+            & intern_ip_q
+            & Q(servertype__ip_addr_type__in=addr_types)
+            & ~Q(server_id=object_id)
+        )
     ):
         duplicates.append(f"{d.hostname} (intern_ip)")
 
-    for d in ServerInetAttribute.objects.filter(value=ip_interface).exclude(
-        Q(server__servertype__ip_addr_type__in=exclude_addr_types)
-        | object_attribute_condition
+    for d in ServerInetAttribute.objects.filter(
+        Q(
+            servertype_inet_attr_q
+            & inet_attr_q
+            & Q(server__servertype__ip_addr_type__in=addr_types)
+        )
+        & ~object_attribute_condition
     ):
         duplicates.append(f"{d.server.hostname} ({d.attribute})")
 
@@ -187,43 +206,6 @@ def inet_to_python(obj: object) -> Union[IPv4Interface, IPv6Interface]:
         raise ValidationError(str(error))
 
 
-def network_overlaps(
-    ip_interface: Union[IPv4Interface, IPv6Interface],
-    servertype_id: str,
-    object_id: int,
-) -> None:
-    """Validate if network overlaps with other objects of the servertype_id
-
-    Raises a ValidationError if the ip network overlaps with any other existing
-    objects network of the given servertype.
-
-    :param ip_interface:
-    :param servertype_id:
-    :param object_id:
-    :return:
-    """
-
-    overlaps = (
-        Server.objects.filter(
-            servertype=servertype_id, intern_ip__net_overlaps=ip_interface
-        )
-        .exclude(server_id=object_id)
-        .exists()
-        or
-        # TODO: We should filter for attribute id here as well to have
-        # consistent bebaviour with ip_addr_type: host and is_unique.
-        ServerInetAttribute.objects.filter(
-            server__servertype=servertype_id, value__net_overlaps=ip_interface
-        )
-        .exclude(server_id=object_id)
-        .exists()
-    )
-    if overlaps:
-        raise ValidationError(
-            "{0} overlaps with network of another object".format(str(ip_interface))
-        )
-
-
 class Servertype(models.Model):
     servertype_id = models.CharField(
         max_length=32,
@@ -515,14 +497,12 @@ def clean(self):
 
             if ip_addr_type == "host":
                 is_ip_address(self.intern_ip)
-                is_unique_ip(self.intern_ip, self, ["network", "loadbalancer"])
-            elif ip_addr_type == "loadbalancer":
-                is_ip_address(self.intern_ip)
+                is_unique_ip(self.intern_ip, self, ["host"], False)
             elif ip_addr_type == "network":
                 is_network(self.intern_ip)
-                network_overlaps(
-                    self.intern_ip, self.servertype.servertype_id, self.server_id
-                )
+                is_unique_ip(self.intern_ip, self, ["network"], True)
+            elif ip_addr_type == "loadbalancer":
+                is_ip_address(self.intern_ip)
 
     def get_attributes(self, attribute):
         model = ServerAttribute.get_model(attribute.type)
@@ -748,16 +728,12 @@ def clean(self):
             )
         elif ip_addr_type == "host":
             is_ip_address(self.value)
-            is_unique_ip(
-                self.value, self.server, ["network", "loadbalancer"], self.attribute_id
-            )
-        elif ip_addr_type == "loadbalancer":
-            is_ip_address(self.value)
+            is_unique_ip(self.value, self.server, ["host"], False, self.attribute_id)
         elif ip_addr_type == "network":
             is_network(self.value)
-            network_overlaps(
-                self.value, self.server.servertype_id, self.server.server_id
-            )
+            is_unique_ip(self.value, self.server, ["network"], True, self.attribute_id)
+        elif ip_addr_type == "loadbalancer":
+            is_ip_address(self.value)
 
 
 class ServerMACAddressAttribute(ServerAttribute):