Skip to content

Commit 6ba4032

Browse files
committed
Unify network_overlaps and is_unique_ip
Use a single function is_unique_ip() for host and network inet address types. This means networks just like hosts will allow duplicate IP address as long as they are on a different attribute. Update tests to take into account that change, ensure that the duplicate inet attributes are allowed between different inet attribute types independent of order of server creation.
1 parent 7b671cd commit 6ba4032

File tree

2 files changed

+51
-89
lines changed

2 files changed

+51
-89
lines changed

serveradmin/serverdb/models.py

Lines changed: 43 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -101,24 +101,23 @@ def is_ip_address(ip_interface: Union[IPv4Interface, IPv6Interface]) -> None:
101101
def is_unique_ip(
102102
ip_interface: Union[IPv4Interface, IPv6Interface],
103103
server: "Server",
104-
exclude_addr_types: list[str],
105-
attribute_id: Optional[str] = None,
104+
addr_types: list[str],
105+
is_net: bool,
106+
attribute_id: Optional[int] = None,
106107
) -> None:
107-
"""Validate if IPv4/IPv6 address is unique
108+
"""Validate if IPv4/IPv6 interface is unique
108109
109110
Raises a ValidationError if intern_ip or any other attribute of type inet
110111
with this ip_address already exists.
111112
112113
:param ip_interface:
113-
:param object_id:
114+
:param server:
115+
:param addr_types:
116+
:param is_net:
114117
:param attribute_id:
115118
:return:
116119
"""
117120

118-
# We avoid querying the duplicate hosts here and giving the user
119-
# detailed information because checking with exists is cheaper than
120-
# querying the server and this is a validation and should be fast.
121-
122121
# Always exclude the current object_id from the query because we allow
123122
# duplication of data between the legacy (intern_ip, primary_ip6) and
124123
# the modern (ipv4, ipv6) attributes.
@@ -130,23 +129,41 @@ def is_unique_ip(
130129
duplicates = []
131130
object_id = server.server_id
132131

133-
# TODO: Make "aid" mandatory when intern_ip is gone.
132+
if is_net:
133+
servertype_inet_attr_q = Q(server__servertype=server.servertype_id)
134+
servertype_intern_ip_q = Q(servertype=server.servertype_id)
135+
inet_attr_q = Q(value__net_overlaps=ip_interface)
136+
intern_ip_q = Q(intern_ip__net_overlaps=ip_interface)
137+
else:
138+
servertype_inet_attr_q = Q()
139+
servertype_intern_ip_q = Q()
140+
inet_attr_q = Q(value=ip_interface)
141+
intern_ip_q = Q(intern_ip=ip_interface)
142+
143+
# TODO: Make attribute_id mandatory when intern_ip is gone.
134144
if attribute_id:
135-
object_attribute_condition = Q(server_id=object_id) | ~Q(
136-
attribute_id=attribute_id
137-
)
145+
object_attribute_q = Q(server_id=object_id) | ~Q(attribute_id=attribute_id)
138146
else:
139-
object_attribute_condition = Q(server_id=object_id)
147+
object_attribute_q = Q(server_id=object_id)
140148

141149
# TODO: Remove intern_ip.
142-
for d in Server.objects.filter(intern_ip=ip_interface).exclude(
143-
Q(servertype__ip_addr_type__in=exclude_addr_types) | Q(server_id=object_id)
150+
for d in Server.objects.filter(
151+
Q(
152+
servertype_intern_ip_q # Servertype
153+
& intern_ip_q # IP address
154+
& Q(servertype__ip_addr_type__in=addr_types) # IP address type
155+
& ~Q(server_id=object_id) # Self-server
156+
)
144157
):
145158
duplicates.append(f"{d.hostname} (intern_ip)")
146159

147-
for d in ServerInetAttribute.objects.filter(value=ip_interface).exclude(
148-
Q(server__servertype__ip_addr_type__in=exclude_addr_types)
149-
| object_attribute_condition
160+
for d in ServerInetAttribute.objects.filter(
161+
Q(
162+
servertype_inet_attr_q # Servertype
163+
& inet_attr_q # IP address
164+
& Q(server__servertype__ip_addr_type__in=addr_types) # IP address type
165+
& ~object_attribute_q # Self-server
166+
)
150167
):
151168
duplicates.append(f"{d.server.hostname} ({d.attribute})")
152169

@@ -187,43 +204,6 @@ def inet_to_python(obj: object) -> Union[IPv4Interface, IPv6Interface]:
187204
raise ValidationError(str(error))
188205

189206

190-
def network_overlaps(
191-
ip_interface: Union[IPv4Interface, IPv6Interface],
192-
servertype_id: str,
193-
object_id: int,
194-
) -> None:
195-
"""Validate if network overlaps with other objects of the servertype_id
196-
197-
Raises a ValidationError if the ip network overlaps with any other existing
198-
objects network of the given servertype.
199-
200-
:param ip_interface:
201-
:param servertype_id:
202-
:param object_id:
203-
:return:
204-
"""
205-
206-
overlaps = (
207-
Server.objects.filter(
208-
servertype=servertype_id, intern_ip__net_overlaps=ip_interface
209-
)
210-
.exclude(server_id=object_id)
211-
.exists()
212-
or
213-
# TODO: We should filter for attribute id here as well to have
214-
# consistent bebaviour with ip_addr_type: host and is_unique.
215-
ServerInetAttribute.objects.filter(
216-
server__servertype=servertype_id, value__net_overlaps=ip_interface
217-
)
218-
.exclude(server_id=object_id)
219-
.exists()
220-
)
221-
if overlaps:
222-
raise ValidationError(
223-
"{0} overlaps with network of another object".format(str(ip_interface))
224-
)
225-
226-
227207
class Servertype(models.Model):
228208
servertype_id = models.CharField(
229209
max_length=32,
@@ -515,14 +495,12 @@ def clean(self):
515495

516496
if ip_addr_type == "host":
517497
is_ip_address(self.intern_ip)
518-
is_unique_ip(self.intern_ip, self, ["network", "loadbalancer"])
519-
elif ip_addr_type == "loadbalancer":
520-
is_ip_address(self.intern_ip)
498+
is_unique_ip(self.intern_ip, self, ["host"], False)
521499
elif ip_addr_type == "network":
522500
is_network(self.intern_ip)
523-
network_overlaps(
524-
self.intern_ip, self.servertype.servertype_id, self.server_id
525-
)
501+
is_unique_ip(self.intern_ip, self, ["network"], True)
502+
elif ip_addr_type == "loadbalancer":
503+
is_ip_address(self.intern_ip)
526504

527505
def get_attributes(self, attribute):
528506
model = ServerAttribute.get_model(attribute.type)
@@ -748,16 +726,12 @@ def clean(self):
748726
)
749727
elif ip_addr_type == "host":
750728
is_ip_address(self.value)
751-
is_unique_ip(
752-
self.value, self.server, ["network", "loadbalancer"], self.attribute_id
753-
)
754-
elif ip_addr_type == "loadbalancer":
755-
is_ip_address(self.value)
729+
is_unique_ip(self.value, self.server, ["host"], False, self.attribute_id)
756730
elif ip_addr_type == "network":
757731
is_network(self.value)
758-
network_overlaps(
759-
self.value, self.server.servertype_id, self.server.server_id
760-
)
732+
is_unique_ip(self.value, self.server, ["network"], True, self.attribute_id)
733+
elif ip_addr_type == "loadbalancer":
734+
is_ip_address(self.value)
761735

762736

763737
class ServerMACAddressAttribute(ServerAttribute):

serveradmin/serverdb/tests/test_ip_addr_type.py

Lines changed: 8 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -211,16 +211,17 @@ def test_server_with_duplicate_inet_different_attrs(self):
211211
other_attribute["ip_config_new"] = "10.0.0.2/32"
212212
self.assertIsNone(other_attribute.commit(user=User.objects.first()))
213213

214-
def test_server_with_duplicate_inet_for_loadbalancer(self):
215-
server = self._get_server("loadbalancer")
214+
def test_server_with_duplicate_inet_ip(self):
215+
server = self._get_server("host")
216216
server["intern_ip"] = "10.0.0.1/32"
217+
server["ip_config"] = "10.0.0.2/32"
217218
server.commit(user=User.objects.first())
218219

219-
duplicate = self._get_server("host")
220+
# Duplicate attribute is allowed because of different inet attribute type
221+
duplicate = self._get_server("loadbalancer")
220222
duplicate["intern_ip"] = "10.0.0.2/32"
221223
duplicate["ip_config"] = "10.0.0.1/32"
222-
with self.assertRaises(ValidationError):
223-
duplicate.commit(user=User.objects.first())
224+
self.assertIsNone(duplicate.commit(user=User.objects.first()))
224225

225226
def test_change_server_hostname(self):
226227
server = self._get_server("host")
@@ -309,12 +310,11 @@ def test_server_with_duplicate_inet_ip(self):
309310
first["ip_config"] = "10.0.0.2/32"
310311
first.commit(user=User.objects.first())
311312

312-
# Test "cross" duplicate attribute is denied
313+
# Duplicate attribute is allowed because of different inet attribute type
313314
duplicate = self._get_server("host")
314315
duplicate["intern_ip"] = "10.0.0.2/32"
315316
duplicate["ip_config"] = "10.0.0.1/32"
316-
with self.assertRaises(ValidationError):
317-
duplicate.commit(user=User.objects.first())
317+
self.assertIsNone(duplicate.commit(user=User.objects.first()))
318318

319319
def test_server_with_duplicate_inet_different_attrs(self):
320320
server = self._get_server("loadbalancer")
@@ -512,18 +512,6 @@ def test_server_network_overlaps_other_servertype(self):
512512
overlaps["ip_config"] = "10.0.1.0/30"
513513
self.assertIsNone(overlaps.commit(user=User.objects.first()))
514514

515-
def test_server_with_duplicate_inet_different_attrs(self):
516-
server = self._get_server("network")
517-
server["intern_ip"] = "10.0.0.0/30"
518-
server["ip_config"] = "10.0.1.0/30"
519-
server.commit(user=User.objects.first())
520-
521-
duplicate = self._get_server("network")
522-
duplicate["intern_ip"] = "10.0.2.0/30"
523-
duplicate["ip_config_new"] = "10.0.1.0/30"
524-
with self.assertRaises(ValidationError):
525-
duplicate.commit(user=User.objects.first())
526-
527515
def test_change_server_hostname(self):
528516
server = self._get_server("network")
529517
server["intern_ip"] = "10.0.0.0/30"

0 commit comments

Comments
 (0)