1
0
mirror of https://github.com/netbox-community/netbox.git synced 2024-05-10 07:54:54 +00:00

Merge branch 'develop' into feature

This commit is contained in:
Jeremy Stretch
2023-12-15 16:52:42 -05:00
40 changed files with 396 additions and 85 deletions

View File

@@ -1,6 +1,6 @@
from django.apps import apps
from django.db.models import F, Count, OuterRef, Subquery
from django.db.models.signals import post_delete, post_save
from django.db.models.signals import post_delete, post_save, pre_delete
from netbox.registry import registry
from .fields import CounterCacheField
@@ -62,6 +62,12 @@ def post_save_receiver(sender, instance, created, **kwargs):
update_counter(parent_model, new_pk, counter_name, 1)
def pre_delete_receiver(sender, instance, origin, **kwargs):
model = instance._meta.model
if not model.objects.filter(pk=instance.pk).exists():
instance._previously_removed = True
def post_delete_receiver(sender, instance, origin, **kwargs):
"""
Update counter fields on related objects when a TrackingModelMixin subclass is deleted.
@@ -71,10 +77,8 @@ def post_delete_receiver(sender, instance, origin, **kwargs):
parent_pk = getattr(instance, field_name, None)
# Decrement the parent's counter by one
if parent_pk is not None:
# MPTT sends two delete signals for child elements so guard against multiple decrements
if not origin or origin == instance:
update_counter(parent_model, parent_pk, counter_name, -1)
if parent_pk is not None and not hasattr(instance, "_previously_removed"):
update_counter(parent_model, parent_pk, counter_name, -1)
#
@@ -106,6 +110,12 @@ def connect_counters(*models):
weak=False,
dispatch_uid=f'{model._meta.label}.{field.name}'
)
pre_delete.connect(
pre_delete_receiver,
sender=to_model,
weak=False,
dispatch_uid=f'{model._meta.label}.{field.name}'
)
post_delete.connect(
post_delete_receiver,
sender=to_model,

View File

@@ -65,5 +65,5 @@ class ChoicesWidget(forms.Textarea):
if not value:
return None
if type(value) is list:
return '\n'.join([f'{k},{v}' for k, v in value])
return '\n'.join([f'{k}:{v}' for k, v in value])
return value

View File

@@ -1,4 +1,5 @@
from netaddr import IPAddress
from netaddr import AddrFormatError, IPAddress
from urllib.parse import urlparse
__all__ = (
'get_client_ip',
@@ -17,11 +18,18 @@ def get_client_ip(request, additional_headers=()):
)
for header in HTTP_HEADERS:
if header in request.META:
client_ip = request.META[header].split(',')[0].partition(':')[0]
ip = request.META[header].split(',')[0].strip()
try:
return IPAddress(client_ip)
except ValueError:
raise ValueError(f"Invalid IP address set for {header}: {client_ip}")
return IPAddress(ip)
except AddrFormatError:
# Parse the string with urlparse() to remove port number or any other cruft
ip = urlparse(f'//{ip}').hostname
try:
return IPAddress(ip)
except AddrFormatError:
# We did our best
raise ValueError(f"Invalid IP address set for {header}: {ip}")
# Could not determine the client IP address from request headers
return None

View File

@@ -1,3 +1,3 @@
<a class="btn btn-sm {{ color }} copy-content" data-clipboard-target="{{ target }}" title="Copy to clipboard">
<a class="btn btn-sm {{ color }} copy-content {{ classes }}" data-clipboard-target="{{ target }}" title="Copy to clipboard">
<i class="mdi mdi-content-copy"></i>
</a>

View File

@@ -87,13 +87,14 @@ def checkmark(value, show_false=True, true='Yes', false='No'):
@register.inclusion_tag('builtins/copy_content.html')
def copy_content(target, prefix=None, color='primary'):
def copy_content(target, prefix=None, color='primary', classes=None):
"""
Display a copy button to copy the content of a field.
"""
return {
'target': f'#{prefix or ""}{target}',
'color': f'btn-{color}'
'color': f'btn-{color}',
'classes': classes or '',
}

View File

@@ -0,0 +1,28 @@
from django.test import TestCase, RequestFactory
from netaddr import IPAddress
from utilities.request import get_client_ip
class GetClientIPTests(TestCase):
def setUp(self):
self.factory = RequestFactory()
def test_ipv4_address(self):
request = self.factory.get('/', HTTP_X_FORWARDED_FOR='192.168.1.1')
self.assertEqual(get_client_ip(request), IPAddress('192.168.1.1'))
request = self.factory.get('/', HTTP_X_FORWARDED_FOR='192.168.1.1:8080')
self.assertEqual(get_client_ip(request), IPAddress('192.168.1.1'))
def test_ipv6_address(self):
request = self.factory.get('/', HTTP_X_FORWARDED_FOR='2001:db8::8a2e:370:7334')
self.assertEqual(get_client_ip(request), IPAddress('2001:db8::8a2e:370:7334'))
request = self.factory.get('/', HTTP_X_FORWARDED_FOR='[2001:db8::8a2e:370:7334]')
self.assertEqual(get_client_ip(request), IPAddress('2001:db8::8a2e:370:7334'))
request = self.factory.get('/', HTTP_X_FORWARDED_FOR='[2001:db8::8a2e:370:7334]:8080')
self.assertEqual(get_client_ip(request), IPAddress('2001:db8::8a2e:370:7334'))
def test_invalid_ip_address(self):
request = self.factory.get('/', HTTP_X_FORWARDED_FOR='invalid_ip')
with self.assertRaises(ValueError):
get_client_ip(request)