From f4950c6388141c18c5a1aa26b84cfc9b3ab519b9 Mon Sep 17 00:00:00 2001 From: Jonas Braathen Date: Thu, 19 Mar 2026 02:25:30 +0100 Subject: [PATCH] Reformat all files with ruff --- fabfile.py | 26 +++-- manage.py | 1 + mdb/__init__.py | 2 +- mdb/admin.py | 142 ++++++++++++++++-------- mdb/api/serializers.py | 34 +++--- mdb/api/views.py | 2 +- mdb/apps.py | 4 +- mdb/forms.py | 12 +-- mdb/management/commands/dhcp_sync.py | 3 +- mdb/models.py | 153 +++++++++++++++----------- mdb/pxe.py | 27 ++--- mdb/signals.py | 25 +++-- mdb/utils.py | 16 +-- mdb/validators.py | 14 ++- mdbsite/settings.py | 155 +++++++++++++-------------- mdbsite/urls.py | 10 +- pyproject.toml | 15 +++ tests/conftest.py | 9 +- tests/test_api.py | 12 ++- tests/test_dhcp_sync.py | 12 ++- tests/test_dhcp_sync_command.py | 4 + tests/test_discord.py | 13 ++- 22 files changed, 420 insertions(+), 271 deletions(-) diff --git a/fabfile.py b/fabfile.py index 08f9867..7cdd66f 100644 --- a/fabfile.py +++ b/fabfile.py @@ -7,17 +7,23 @@ from invoke import task @task() def deploy(c): """Make sure proxy_user is set to your neuf username.""" - project_path = '/opt/mdb' - proxy_user = os.getenv('DEPLOY_USER', os.getenv('USER')) + project_path = "/opt/mdb" + proxy_user = os.getenv("DEPLOY_USER", os.getenv("USER")) - c = Connection(host='gitdeploy@sega.neuf.no', gateway=Connection('login.neuf.no', user=proxy_user)) + c = Connection( + host="gitdeploy@sega.neuf.no", + gateway=Connection("login.neuf.no", user=proxy_user), + ) - with c.cd(project_path), c.prefix('source {}/.venv/bin/activate'.format(project_path)): - c.run('git pull') # Get source - c.run('pip install -U uv') - c.run('uv sync') # install deps in virtualenv - c.run('uv run manage.py collectstatic --noinput') # Collect static - c.run('uv run manage.py migrate') # Run DB migrations + with ( + c.cd(project_path), + c.prefix("source {}/.venv/bin/activate".format(project_path)), + ): + c.run("git pull") # Get source + c.run("pip install -U uv") + c.run("uv sync") # install deps in virtualenv + c.run("uv run manage.py collectstatic --noinput") # Collect static + c.run("uv run manage.py migrate") # Run DB migrations # Reload gunicorn - c.sudo('/usr/bin/supervisorctl pid mdb.neuf.no | xargs kill -HUP', shell=False) + c.sudo("/usr/bin/supervisorctl pid mdb.neuf.no | xargs kill -HUP", shell=False) diff --git a/manage.py b/manage.py index f6ab478..2c9e30b 100755 --- a/manage.py +++ b/manage.py @@ -5,4 +5,5 @@ import sys if __name__ == "__main__": os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mdbsite.settings") from django.core.management import execute_from_command_line + execute_from_command_line(sys.argv) diff --git a/mdb/__init__.py b/mdb/__init__.py index d600276..4bbde1c 100644 --- a/mdb/__init__.py +++ b/mdb/__init__.py @@ -1 +1 @@ -default_app_config = 'mdb.apps.MdbAppConfig' +default_app_config = "mdb.apps.MdbAppConfig" diff --git a/mdb/admin.py b/mdb/admin.py index 94a8796..2a3a1fa 100644 --- a/mdb/admin.py +++ b/mdb/admin.py @@ -48,45 +48,70 @@ class InterfaceInline(admin.TabularInline): class HostAdmin(admin.ModelAdmin): - ordering = ('hostname',) + ordering = ("hostname",) inlines = [InterfaceInline] - list_display = ['hostname', 'owner', 'host_type', 'location', 'mac_addresses', 'ip_addresses', 'in_domain', - 'pxe_installable', 'ipv6_enabled'] - list_filter = ['host_type', 'owner', 'location', 'pxe_installable'] - readonly_fields = ['kerberos_principal_name', 'kerberos_principal_created_date', - 'kerberos_principal_created'] - search_fields = ['hostname', 'location', 'interface__macaddr', 'interface__ip4address__address'] + list_display = [ + "hostname", + "owner", + "host_type", + "location", + "mac_addresses", + "ip_addresses", + "in_domain", + "pxe_installable", + "ipv6_enabled", + ] + list_filter = ["host_type", "owner", "location", "pxe_installable"] + readonly_fields = [ + "kerberos_principal_name", + "kerberos_principal_created_date", + "kerberos_principal_created", + ] + search_fields = [ + "hostname", + "location", + "interface__macaddr", + "interface__ip4address__address", + ] fieldsets = ( - ('Owner Information', { - 'fields': ('owner', 'location', 'description') - }), - ('Hardware and Software Information', { - 'fields': ( - ('hostname', 'host_type'), - ('pxe_key', 'pxe_installable'), - ('brand', 'model'), - 'serial_number', - ('operating_system', 'virtual')) - }), + ("Owner Information", {"fields": ("owner", "location", "description")}), + ( + "Hardware and Software Information", + { + "fields": ( + ("hostname", "host_type"), + ("pxe_key", "pxe_installable"), + ("brand", "model"), + "serial_number", + ("operating_system", "virtual"), + ) + }, + ), ) - actions = ['set_installable'] + actions = ["set_installable"] def mac_addresses(self, host): - addresses = host.interface_set.filter(macaddr__isnull=False).values_list('macaddr', flat=True) + addresses = host.interface_set.filter(macaddr__isnull=False).values_list( + "macaddr", flat=True + ) return ", ".join(addresses) def ip_addresses(self, host): - addresses = host.interface_set.filter(ip4address__isnull=False).values_list('ip4address__address', flat=True) + addresses = host.interface_set.filter(ip4address__isnull=False).values_list( + "ip4address__address", flat=True + ) return ", ".join(addresses) def in_domain(self, host): - domains = host.interface_set.values_list('domain__domain_name', flat=True) + domains = host.interface_set.values_list("domain__domain_name", flat=True) return ",".join(domains) in_domain.short_description = "in domains" def ipv6_enabled(self, host): - ipv6_ifs = host.interface_set.annotate(num_ipv6=Count('ip6address')).filter(num_ipv6__gt=0) + ipv6_ifs = host.interface_set.annotate(num_ipv6=Count("ip6address")).filter( + num_ipv6__gt=0 + ) return ipv6_ifs.exists() ipv6_enabled.boolean = True @@ -94,8 +119,10 @@ class HostAdmin(admin.ModelAdmin): @staticmethod def _get_host_warning_message(host, ifs): - ifs_str = ', '.join([_if.name for _if in ifs]) - return 'Host {} might not be installable since interface(s) {} has no IP set.'.format(host, ifs_str) + ifs_str = ", ".join([_if.name for _if in ifs]) + return "Host {} might not be installable since interface(s) {} has no IP set.".format( + host, ifs_str + ) def set_installable(self, request, queryset): invalid_ifs_per_host = [] @@ -103,12 +130,16 @@ class HostAdmin(admin.ModelAdmin): for obj in queryset: invalid_ifs = obj.interface_set.filter(ip4address__isnull=True) if invalid_ifs.exists(): - invalid_ifs_per_host.append(self._get_host_warning_message(obj, invalid_ifs)) + invalid_ifs_per_host.append( + self._get_host_warning_message(obj, invalid_ifs) + ) obj.pxe_installable = True obj.save() # triggers post save signal if invalid_ifs_per_host: - self.message_user(request, message='\n'.join(invalid_ifs_per_host), level=messages.WARNING) + self.message_user( + request, message="\n".join(invalid_ifs_per_host), level=messages.WARNING + ) set_installable.short_description = "Mark selected hosts as PXE installable" @@ -119,7 +150,7 @@ class HostAdmin(admin.ModelAdmin): class Ip4AddressInline(admin.TabularInline): model = Ip4Address extra = 0 - readonly_fields = ['address', 'assigned_to_host'] + readonly_fields = ["address", "assigned_to_host"] def assigned_to_host(self, addr): return addr.interface.host @@ -138,7 +169,15 @@ class DhcpCustomFieldInline(admin.TabularInline): class SubnetAdmin(admin.ModelAdmin): - list_display = ['name', 'network', 'netmask', 'num_addresses', 'broadcast_address', 'first_address', 'last_address'] + list_display = [ + "name", + "network", + "netmask", + "num_addresses", + "broadcast_address", + "first_address", + "last_address", + ] inlines = [DhcpOptionInline, DhcpCustomFieldInline, Ip4AddressInline] def num_addresses(self, ip4subnet): @@ -157,10 +196,10 @@ class SubnetAdmin(admin.ModelAdmin): subnet = ipaddress.IPv4Network(ip4subnet.network + "/" + ip4subnet.netmask) return list(subnet.hosts())[-1] - broadcast_address.short_description = 'broadcast' - num_addresses.short_description = '#addresses' - first_address.short_description = 'first address' - last_address.short_description = 'last address' + broadcast_address.short_description = "broadcast" + num_addresses.short_description = "#addresses" + first_address.short_description = "first address" + last_address.short_description = "last address" def get_queryset(self, request): # TODO: optimize @@ -193,12 +232,24 @@ class DomainAAAARecordInline(admin.TabularInline): class DomainAdmin(admin.ModelAdmin): - inlines = [DomainSrvRecordInline, DomainTxtRecordInline, DomainARecordInline, DomainAAAARecordInline, - DomainCnameRecordInline] - list_display = ['domain_name', 'domain_soa', 'domain_admin', 'num_records', 'domain_ipaddr', 'domain_ip6addr'] - search_fields = ['domain_name'] + inlines = [ + DomainSrvRecordInline, + DomainTxtRecordInline, + DomainARecordInline, + DomainAAAARecordInline, + DomainCnameRecordInline, + ] + list_display = [ + "domain_name", + "domain_soa", + "domain_admin", + "num_records", + "domain_ipaddr", + "domain_ip6addr", + ] + search_fields = ["domain_name"] formfield_overrides = { - models.IntegerField: {'widget': NumberInput(attrs={'style': 'width: 10em'})}, + models.IntegerField: {"widget": NumberInput(attrs={"style": "width: 10em"})}, } def num_records(self, domain): @@ -207,19 +258,26 @@ class DomainAdmin(admin.ModelAdmin): "CNAME": domain.domaincnamerecord_set.count(), "SRV": domain.domainsrvrecord_set.count(), "TXT": domain.domaintxtrecord_set.count(), - "A": domain.domainarecord_set.count() + host_a_records + "A": domain.domainarecord_set.count() + host_a_records, } - return ', '.join(['{}: {}'.format(k, v) for k, v in size.items()]) + return ", ".join(["{}: {}".format(k, v) for k, v in size.items()]) num_records.short_description = "Num Records" class HostTypeAdmin(admin.ModelAdmin): - list_display = ['host_type', 'description', 'num_members'] + list_display = ["host_type", "description", "num_members"] class OperatingSystemAdmin(admin.ModelAdmin): - list_display = ['name', 'version', 'arch', 'pxe_kernel', 'pxe_initrd', 'pxe_preseed_config_url'] + list_display = [ + "name", + "version", + "arch", + "pxe_kernel", + "pxe_initrd", + "pxe_preseed_config_url", + ] admin.site.register(Domain, DomainAdmin) diff --git a/mdb/api/serializers.py b/mdb/api/serializers.py index 0685b79..4b18a99 100644 --- a/mdb/api/serializers.py +++ b/mdb/api/serializers.py @@ -10,13 +10,15 @@ class HostPXEValidateSerializer(serializers.Serializer): host = None def _get_host_by_fqdn(self, attrs): - fqdn = attrs['certname'] - host_name = fqdn.split('.')[0] - domain = '.'.join(fqdn.split('.')[1:]) + fqdn = attrs["certname"] + host_name = fqdn.split(".")[0] + domain = ".".join(fqdn.split(".")[1:]) try: - return Host.objects.get(hostname=host_name, interface__domain__domain_name=domain) - except (Host.DoesNotExist, Host.MultipleObjectsReturned): + return Host.objects.get( + hostname=host_name, interface__domain__domain_name=domain + ) + except Host.DoesNotExist, Host.MultipleObjectsReturned: return None def validate(self, attrs): @@ -24,18 +26,24 @@ class HostPXEValidateSerializer(serializers.Serializer): host = self._get_host_by_fqdn(attrs) if host is None: - raise ValidationError('Host \'{}\' not found.'.format(attrs['certname'])) + raise ValidationError("Host '{}' not found.".format(attrs["certname"])) if not host.pxe_installable: - raise ValidationError('Host \'{}\' is marked as not installable via PXE.'.format(attrs['certname'])) + raise ValidationError( + "Host '{}' is marked as not installable via PXE.".format( + attrs["certname"] + ) + ) if not host.pxe_key: - raise ValidationError('Host \'{}\' has no pxe_key.'.format(attrs['certname'])) + raise ValidationError("Host '{}' has no pxe_key.".format(attrs["certname"])) - if host.pxe_key != attrs['pxe_key']: - raise ValidationError('Supplied pxe_key \'{}\' does not match host \'{}\'.'.format( - attrs['pxe_key'], - attrs['certname'])) + if host.pxe_key != attrs["pxe_key"]: + raise ValidationError( + "Supplied pxe_key '{}' does not match host '{}'.".format( + attrs["pxe_key"], attrs["certname"] + ) + ) self.host = host @@ -44,7 +52,7 @@ class HostPXEValidateSerializer(serializers.Serializer): def create(self, validated_data): # Make requests kind of idempotent by only allowing 1 request / installation self.host.pxe_installable = False - self.host.pxe_key = '' + self.host.pxe_key = "" self.host.save() return self.host diff --git a/mdb/api/views.py b/mdb/api/views.py index 20e8966..643865c 100644 --- a/mdb/api/views.py +++ b/mdb/api/views.py @@ -17,4 +17,4 @@ class HostPXEValidate(APIView): serializer.save() # sets Host.pxe_installable=False - return Response({'valid_pxe_key': True}) + return Response({"valid_pxe_key": True}) diff --git a/mdb/apps.py b/mdb/apps.py index 6a0f310..a116792 100644 --- a/mdb/apps.py +++ b/mdb/apps.py @@ -2,8 +2,8 @@ from django.apps import AppConfig class MdbAppConfig(AppConfig): - name = 'mdb' - verbose_name = 'Machine Database' + name = "mdb" + verbose_name = "Machine Database" def ready(self): # all models are loaded, now attach signals diff --git a/mdb/forms.py b/mdb/forms.py index de454c4..23843b6 100644 --- a/mdb/forms.py +++ b/mdb/forms.py @@ -7,7 +7,7 @@ from mdb.models import Interface, Ip4Address class Ip4ModelChoiceField(ModelChoiceField): def label_from_instance(self, obj): - if not hasattr(obj, 'interface'): + if not hasattr(obj, "interface"): return obj.address return "{} ({})".format(obj.address, obj.interface.host.hostname) @@ -17,18 +17,18 @@ class InterfaceForm(forms.ModelForm): model = Interface def clean(self): - macaddr = self.cleaned_data['macaddr'] - ip4address = self.cleaned_data['ip4address'] + macaddr = self.cleaned_data["macaddr"] + ip4address = self.cleaned_data["ip4address"] for interface in Interface.objects.filter(macaddr=macaddr): subnet = interface.ip4address.subnet if subnet == ip4address.subnet and interface != self.instance: - raise ValidationError('Macaddr already in this subnet') + raise ValidationError("Macaddr already in this subnet") def __init__(self, *args, **kwargs): super(InterfaceForm, self).__init__(*args, **kwargs) # Join Ip4Address with tables interface and host, # before printing the select labels that walk those two relations. - queryset = Ip4Address.objects.select_related('interface', 'interface__host') - self.fields['ip4address'] = Ip4ModelChoiceField(queryset=queryset) + queryset = Ip4Address.objects.select_related("interface", "interface__host") + self.fields["ip4address"] = Ip4ModelChoiceField(queryset=queryset) diff --git a/mdb/management/commands/dhcp_sync.py b/mdb/management/commands/dhcp_sync.py index 517c1b3..b333fb4 100644 --- a/mdb/management/commands/dhcp_sync.py +++ b/mdb/management/commands/dhcp_sync.py @@ -64,7 +64,8 @@ class Command(BaseCommand): self.stdout.write("fail\n") message = ( "Failed to restart isc-dhcp-server, please inspect...\n\n" - "%s@%s$ %s\n%s" % (os.getlogin(), os.uname()[1], service.restart_command, output) + "%s@%s$ %s\n%s" + % (os.getlogin(), os.uname()[1], service.restart_command, output) ) discord.notify( subject="⚠️ FAIL: dhcpd %s -> %s" % (old_serial, config.active_serial), diff --git a/mdb/models.py b/mdb/models.py index c7a7569..b46e86d 100644 --- a/mdb/models.py +++ b/mdb/models.py @@ -34,8 +34,10 @@ class Domain(models.Model): domain_expire = models.IntegerField(default=604800) domain_minimum_ttl = models.IntegerField(default=86400) domain_admin = models.EmailField() - domain_ipaddr = models.GenericIPAddressField(protocol='IPv4') - domain_ip6addr = models.GenericIPAddressField(protocol='IPv6', blank=True, null=True) + domain_ipaddr = models.GenericIPAddressField(protocol="IPv4") + domain_ip6addr = models.GenericIPAddressField( + protocol="IPv6", blank=True, null=True + ) domain_filename = models.CharField(max_length=256) created_date = models.DateTimeField(auto_now_add=True) @@ -46,7 +48,7 @@ class Domain(models.Model): return self.domain_name def __eq__(self, other): - if not other or not hasattr(other, 'domain_name'): + if not other or not hasattr(other, "domain_name"): return False return self.domain_name == other.domain_name @@ -59,7 +61,8 @@ class Domain(models.Model): content += "$TTL %s\n" % self.domain_ttl content += "@ IN SOA %s. %s. (\n" % ( self.domain_soa, - self.domain_admin.replace("@", ".")) + self.domain_admin.replace("@", "."), + ) content += "\t%d\t; serial\n" % self.domain_serial content += "\t%d\t; refresh\n" % self.domain_refresh content += "\t%d\t; retry\n" % self.domain_retry @@ -111,11 +114,13 @@ class Domain(models.Model): if interface.ip4address: content += "%-20s\tIN\tA\t%s\n" % ( host.hostname, - interface.ip4address.address) + interface.ip4address.address, + ) for ipv6addr in interface.ip6address_set.all(): content += "%-20s\tIN\tAAAA\t%s\n" % ( host.hostname, - ipv6addr.full_address()) + ipv6addr.full_address(), + ) return content @@ -132,15 +137,14 @@ class DomainSrvRecord(models.Model): created_date = models.DateTimeField(auto_now_add=True) def as_record(self): - return '{}.{}.{}. IN SRV {} {} {} {}.'.format( + return "{}.{}.{}. IN SRV {} {} {} {}.".format( self.srvce, self.prot, self.name, - self.priority, self.weight, self.port, - self.target + self.target, ) def __str__(self): @@ -157,7 +161,7 @@ class DomainTxtRecord(models.Model): created_date = models.DateTimeField(auto_now_add=True) def as_record(self): - return '{} TXT {}'.format(self.name, self.target) + return "{} TXT {}".format(self.name, self.target) def __str__(self): return self.as_record() @@ -170,7 +174,7 @@ class DomainCnameRecord(models.Model): created_date = models.DateTimeField(auto_now_add=True) def as_record(self): - return '{} IN CNAME {}'.format(self.name, self.target) + return "{} IN CNAME {}".format(self.name, self.target) def __str__(self): return self.as_record() @@ -178,12 +182,12 @@ class DomainCnameRecord(models.Model): class DomainARecord(models.Model): name = models.CharField(max_length=256) - target = models.GenericIPAddressField(protocol='IPv4') + target = models.GenericIPAddressField(protocol="IPv4") domain = models.ForeignKey(Domain, models.CASCADE) created_date = models.DateTimeField(auto_now_add=True) def as_record(self): - return '{} IN A {}'.format(self.name, self.target) + return "{} IN A {}".format(self.name, self.target) def __str__(self): return self.as_record() @@ -191,12 +195,12 @@ class DomainARecord(models.Model): class DomainAAAARecord(models.Model): name = models.CharField(max_length=256) - target = models.GenericIPAddressField(protocol='IPv6') + target = models.GenericIPAddressField(protocol="IPv6") domain = models.ForeignKey(Domain, models.CASCADE) created_date = models.DateTimeField(auto_now_add=True) def as_record(self): - return '{} IN AAAA {}'.format(self.name, self.target) + return "{} IN AAAA {}".format(self.name, self.target) def __str__(self): return self.as_record() @@ -224,9 +228,7 @@ class DhcpConfig(models.Model): # time to write the subnet definitions for subnet in self.ip4subnet_set.all(): content += "\n# %s\n" % subnet.name - content += "subnet %s netmask %s {\n" % ( - subnet.network, - subnet.netmask) + content += "subnet %s netmask %s {\n" % (subnet.network, subnet.netmask) for option in subnet.dhcpoption_set.all(): content += "\toption %s %s;\n" % (option.key, option.value) @@ -235,23 +237,30 @@ class DhcpConfig(models.Model): content += "\t%s;\n" % option.value if subnet.dhcp_dynamic: - content += "\trange %s %s;\n" \ - % (subnet.dhcp_dynamic_start, subnet.dhcp_dynamic_end) + content += "\trange %s %s;\n" % ( + subnet.dhcp_dynamic_start, + subnet.dhcp_dynamic_end, + ) content += "}\n" # time to write host definitions for subnet in self.ip4subnet_set.all(): for ip4address in subnet.ip4address_set.all(): - if not hasattr(ip4address, 'interface') or not ip4address.interface.dhcp_client: + if ( + not hasattr(ip4address, "interface") + or not ip4address.interface.dhcp_client + ): continue _if = ip4address.interface content += "\nhost %s {\n" % _if.host.hostname content += "\thardware ethernet %s;\n" % _if.macaddr - content += "\tfixed-address %s.%s;\n" % \ - (_if.host.hostname, _if.domain.domain_name) + content += "\tfixed-address %s.%s;\n" % ( + _if.host.hostname, + _if.domain.domain_name, + ) if len(_if.pxe_filename) > 0: - content += "\tfilename \"%s\";\n" % _if.pxe_filename + content += '\tfilename "%s";\n' % _if.pxe_filename content += "}\n" return content @@ -260,7 +269,7 @@ class DhcpConfig(models.Model): return self.name class Meta: - verbose_name = 'DHCP config' + verbose_name = "DHCP config" class Ip6Subnet(models.Model): @@ -292,7 +301,8 @@ class Ip6Subnet(models.Model): content += "$TTL %s\n" % self.domain_ttl content += "@ IN SOA %s. %s. (\n" % ( self.domain_soa, - self.domain_admin.replace("@", ".")) + self.domain_admin.replace("@", "."), + ) content += "\t%d\t; serial\n" % self.domain_serial content += "\t%d\t; refresh\n" % self.domain_refresh content += "\t%d\t; retry\n" % self.domain_retry @@ -317,7 +327,8 @@ class Ip6Subnet(models.Model): continue hostname = "%s.%s" % ( addr.interface.host.hostname, - addr.interface.domain.domain_name) + addr.interface.domain.domain_name, + ) ip = ipaddress.IPv6Address(self.network + addr.address) ip = ".".join(ip.exploded.replace(":", "")[16:])[::-1] @@ -329,13 +340,13 @@ class Ip6Subnet(models.Model): return content class Meta: - verbose_name = 'IPv6 subnet' + verbose_name = "IPv6 subnet" class Ip4Subnet(models.Model): name = models.CharField(max_length=256) - netmask = models.GenericIPAddressField(protocol='IPv4') - network = models.GenericIPAddressField(protocol='IPv4') + netmask = models.GenericIPAddressField(protocol="IPv4") + network = models.GenericIPAddressField(protocol="IPv4") created_date = models.DateTimeField(auto_now_add=True) domain_name = models.CharField(max_length=255, editable=False) domain_nameservers = models.ManyToManyField(Nameserver) @@ -351,8 +362,12 @@ class Ip4Subnet(models.Model): domain_filename = models.CharField(max_length=256) dhcp_dynamic = models.BooleanField(default=False) - dhcp_dynamic_start = models.GenericIPAddressField(protocol='IPv4', null=True, blank=True) - dhcp_dynamic_end = models.GenericIPAddressField(protocol='IPv4', null=True, blank=True) + dhcp_dynamic_start = models.GenericIPAddressField( + protocol="IPv4", null=True, blank=True + ) + dhcp_dynamic_end = models.GenericIPAddressField( + protocol="IPv4", null=True, blank=True + ) dhcp_config = models.ForeignKey(DhcpConfig, models.CASCADE) def __str__(self): @@ -366,7 +381,8 @@ class Ip4Subnet(models.Model): content += "$TTL %s\n" % self.domain_ttl content += "@ IN SOA %s. %s. (\n" % ( self.domain_soa, - self.domain_admin.replace("@", ".")) + self.domain_admin.replace("@", "."), + ) content += "\t%d\t; serial\n" % self.domain_serial content += "\t%d\t; refresh\n" % self.domain_refresh content += "\t%d\t; retry\n" % self.domain_retry @@ -382,22 +398,26 @@ class Ip4Subnet(models.Model): for addr in self.ip4address_set.all(): """ Get PTR records for each v4 address in subnet """ - if hasattr(addr, 'interface') and addr.interface.domain: + if hasattr(addr, "interface") and addr.interface.domain: hostname = "%s.%s" % ( addr.interface.host.hostname, - addr.interface.domain.domain_name) - content += "%-20s\tIN\tPTR\t%s.\n" % \ - (addr.address.split(".")[3], hostname) - elif not hasattr(addr, 'interface') and generate_unassigned: + addr.interface.domain.domain_name, + ) + content += "%-20s\tIN\tPTR\t%s.\n" % ( + addr.address.split(".")[3], + hostname, + ) + elif not hasattr(addr, "interface") and generate_unassigned: content += "%s\tIN\tPTR\t%s.%s\n" % ( addr.address, addr.address.split(".")[3], - "dhcp.neuf.no.") + "dhcp.neuf.no.", + ) return content class Meta: - verbose_name = 'IPv4 subnet' + verbose_name = "IPv4 subnet" class DhcpOption(models.Model): @@ -409,7 +429,7 @@ class DhcpOption(models.Model): return self.key + " " + self.value class Meta: - verbose_name = 'DHCP option' + verbose_name = "DHCP option" class DhcpCustomField(models.Model): @@ -417,12 +437,12 @@ class DhcpCustomField(models.Model): ip4subnet = models.ForeignKey(Ip4Subnet, models.CASCADE) class Meta: - verbose_name = 'DHCP custom field' + verbose_name = "DHCP custom field" class Ip4Address(models.Model): subnet = models.ForeignKey(Ip4Subnet, models.CASCADE) - address = models.GenericIPAddressField(protocol='IPv4') + address = models.GenericIPAddressField(protocol="IPv4") last_contact = models.DateTimeField(null=True, blank=True) ping_avg_rtt = models.FloatField(null=True, blank=True) @@ -430,8 +450,8 @@ class Ip4Address(models.Model): return self.address class Meta: - verbose_name = 'IPv4 address' - verbose_name_plural = 'IPv4 addresses' + verbose_name = "IPv4 address" + verbose_name_plural = "IPv4 addresses" class HostType(models.Model): @@ -450,19 +470,21 @@ class HostType(models.Model): class OperatingSystem(models.Model): OS_ARCHITECTURES = ( - ('mips', _('MIPS')), - ('arm', _('ARM')), - ('broadcom', _('Broadcom')), - ('rc32300', _('RC32300')), - ('powerpc', _('PowerPC')), - ('powerpc403ga', _('PowerPC403GA')), - ('unknown', _('Unknown')), - ('x86_64', _('x86-64')), - ('i386', _('x86')), + ("mips", _("MIPS")), + ("arm", _("ARM")), + ("broadcom", _("Broadcom")), + ("rc32300", _("RC32300")), + ("powerpc", _("PowerPC")), + ("powerpc403ga", _("PowerPC403GA")), + ("unknown", _("Unknown")), + ("x86_64", _("x86-64")), + ("i386", _("x86")), ) name = models.CharField(max_length=256) version = models.CharField(max_length=64) - arch = models.CharField(max_length=255, choices=OS_ARCHITECTURES, blank=True, null=True) + arch = models.CharField( + max_length=255, choices=OS_ARCHITECTURES, blank=True, null=True + ) pxe_kernel = models.CharField(max_length=255, blank=True, null=True) pxe_initrd = models.CharField(max_length=255, blank=True, null=True) @@ -486,12 +508,16 @@ class Host(models.Model): created_date = models.DateTimeField(auto_now_add=True) host_type = models.ForeignKey(HostType, models.CASCADE) # FIXME: SET_NULL? virtual = models.BooleanField(default=False) - operating_system = models.ForeignKey(OperatingSystem, models.CASCADE) # FIXME: SET_NULL? + operating_system = models.ForeignKey( + OperatingSystem, models.CASCADE + ) # FIXME: SET_NULL? request_kerberos_principal = models.BooleanField(default=False) kerberos_principal_created = models.BooleanField(default=False, editable=False) kerberos_principal_name = models.CharField(max_length=256, editable=False) - kerberos_principal_created_date = models.DateTimeField(null=True, blank=True, editable=False) + kerberos_principal_created_date = models.DateTimeField( + null=True, blank=True, editable=False + ) pxe_key = models.CharField(max_length=254, blank=True) pxe_installable = models.BooleanField(default=False) @@ -509,7 +535,9 @@ class Interface(models.Model): pxe_filename = models.CharField(max_length=64, blank=True) dhcp_client = models.BooleanField(default=False) host = models.ForeignKey(Host, models.CASCADE) - ip4address = models.OneToOneField(Ip4Address, blank=True, null=True, on_delete=models.SET_NULL) + ip4address = models.OneToOneField( + Ip4Address, blank=True, null=True, on_delete=models.SET_NULL + ) created_date = models.DateTimeField(auto_now_add=True) domain = models.ForeignKey(Domain, models.CASCADE) @@ -522,7 +550,7 @@ class Interface(models.Model): class Ip6Address(models.Model): subnet = models.ForeignKey(Ip6Subnet, models.CASCADE) - address = models.GenericIPAddressField(protocol='IPv6') + address = models.GenericIPAddressField(protocol="IPv6") interface = models.ForeignKey(Interface, models.CASCADE) def full_address(self): @@ -532,8 +560,9 @@ class Ip6Address(models.Model): return "%s (%s on %s)" % ( self.full_address(), self.interface.name, - self.interface.host.hostname) + self.interface.host.hostname, + ) class Meta: - verbose_name = 'IPv6 address' - verbose_name_plural = 'IPv6 addresses' + verbose_name = "IPv6 address" + verbose_name_plural = "IPv6 addresses" diff --git a/mdb/pxe.py b/mdb/pxe.py index 6ac6fe9..1268f5b 100644 --- a/mdb/pxe.py +++ b/mdb/pxe.py @@ -14,39 +14,42 @@ def render_pxelinux_cfg(_if, host): # Per host config (via OS model) _os = host.operating_system pxe_attrs = [_os.pxe_initrd, _os.pxe_kernel, _os.pxe_preseed_config_url] - if None not in pxe_attrs and '' not in pxe_attrs: + if None not in pxe_attrs and "" not in pxe_attrs: kernel = _os.pxe_kernel initrd = _os.pxe_initrd preseed_config_url = _os.pxe_preseed_config_url context = { - 'kernel': kernel, - 'initrd': initrd, - 'preseed_config_url': preseed_config_url, - 'host': host, - 'interface': _if + "kernel": kernel, + "initrd": initrd, + "preseed_config_url": preseed_config_url, + "host": host, + "interface": _if, } - return render_to_string('pxelinux.cfg', context=context) + return render_to_string("pxelinux.cfg", context=context) -def get_pxe_filename(interface, filename_format='mac_addr'): +def get_pxe_filename(interface, filename_format="mac_addr"): """ Mac or IP address based PXE filename Ref: http://www.syslinux.org/wiki/index.php/PXELINUX#Examples """ - ip_addr_in_hex = '{:02X}'.format(int(ipaddress.IPv4Address(interface.ip4address.address))) + ip_addr_in_hex = "{:02X}".format( + int(ipaddress.IPv4Address(interface.ip4address.address)) + ) formats = { - 'ip_addr': ip_addr_in_hex, - 'mac_addr': '01-{}'.format(interface.macaddr.lower().replace(':', '-')) + "ip_addr": ip_addr_in_hex, + "mac_addr": "01-{}".format(interface.macaddr.lower().replace(":", "-")), } - return formats.get(filename_format, formats['mac_addr']) + return formats.get(filename_format, formats["mac_addr"]) def host_as_pxe_files(host): """Returns a list of filename,content for each host interface with an IPv4 address.""" from mdb.models import Host + assert isinstance(host, Host) pxe_files = [] diff --git a/mdb/signals.py b/mdb/signals.py index 3a2018f..d959cb2 100644 --- a/mdb/signals.py +++ b/mdb/signals.py @@ -46,7 +46,9 @@ def set_domain_name_for_subnet(sender, instance, **kwargs): # when the subnet is changed if instance.dhcp_config: # instance.dhcp_config.serial = instance.dhcp_config.serial + 1 - instance.dhcp_config.serial = format_domain_serial_and_add_one(instance.dhcp_config.serial) + instance.dhcp_config.serial = format_domain_serial_and_add_one( + instance.dhcp_config.serial + ) instance.dhcp_config.save() @@ -56,7 +58,9 @@ def set_domain_name_for_ipv6_subnet(sender, instance, **kwargs): return network = ipaddress.IPv6Address("%s::" % instance.network) - instance.domain_name = ".".join(network.exploded.replace(":", "")[:16])[::-1] + ".ip6.arpa" + instance.domain_name = ( + ".".join(network.exploded.replace(":", "")[:16])[::-1] + ".ip6.arpa" + ) @receiver(post_save, sender=Interface) @@ -77,16 +81,22 @@ def update_domain_serial_when_change_to_host(sender, instance, created, **kwargs for interface in instance.interface_set.all(): if interface.domain is not None: domain = interface.domain - domain.domain_serial = format_domain_serial_and_add_one(domain.domain_serial) + domain.domain_serial = format_domain_serial_and_add_one( + domain.domain_serial + ) domain.save() if interface.ip4address is not None: subnet = interface.ip4address.subnet - subnet.domain_serial = format_domain_serial_and_add_one(subnet.domain_serial) + subnet.domain_serial = format_domain_serial_and_add_one( + subnet.domain_serial + ) subnet.save() @receiver(post_save, sender=Host) -def create_pxe_key_and_write_pxe_files_when_host_changes(sender, instance, created, **kwargs): +def create_pxe_key_and_write_pxe_files_when_host_changes( + sender, instance, created, **kwargs +): host = instance if not host.pxe_key: @@ -99,7 +109,7 @@ def create_pxe_key_and_write_pxe_files_when_host_changes(sender, instance, creat path = os.path.join(settings.MDB_PXE_TFTP_ROOT, pxe_file_name) if host.pxe_installable: # Write pxeconfig file to disk - with open(path, 'w+') as f: + with open(path, "w+") as f: f.write(pxe_file) logger.info("Created or updated {}".format(path)) os.chmod(path, 0o644) # Not writeable for all @@ -110,7 +120,7 @@ def create_pxe_key_and_write_pxe_files_when_host_changes(sender, instance, creat for _if in interfaces: changed = False if not _if.pxe_filename: - _if.pxe_filename = 'pxelinux.0' # don't overwrite + _if.pxe_filename = "pxelinux.0" # don't overwrite changed = True if not _if.dhcp_client: _if.dhcp_client = True @@ -138,6 +148,7 @@ def update_domain_serial_when_interface_deleted(sender, instance, **kwargs): subnet.domain_serial += 1 subnet.save() + # @receiver(pre_save, sender=Domain) # def update_domain_serial_when_domain_is_saved(sender, instance, **kwargs): # instance.domain_serial = format_domain_serial_and_add_one(instance.domain_serial) diff --git a/mdb/utils.py b/mdb/utils.py index de83d38..7429066 100644 --- a/mdb/utils.py +++ b/mdb/utils.py @@ -4,15 +4,17 @@ import re def format_domain_serial_and_add_one(serial): today = datetime.datetime.now() - res = re.findall(r"^%4d%02d%02d(\d\d)$" % ( - today.year, today.month, today.day), str(serial), re.DOTALL) + res = re.findall( + r"^%4d%02d%02d(\d\d)$" % (today.year, today.month, today.day), + str(serial), + re.DOTALL, + ) if len(res) == 0: """ This probably means that the serial is malformed or the date is wrong. We assume that if the date is wrong, it is in the past. Just create a new serial starting from 1.""" - return "%4d%02d%02d%02d" % \ - (today.year, today.month, today.day, 1) + return "%4d%02d%02d%02d" % (today.year, today.month, today.day, 1) elif len(res) == 1: """ The serial contains todays date, just update it. """ try: @@ -26,9 +28,7 @@ def format_domain_serial_and_add_one(serial): else: number += 1 - return "%4d%02d%02d%02d" % ( - today.year, today.month, today.day, number) + return "%4d%02d%02d%02d" % (today.year, today.month, today.day, number) else: """ Just return the first serial for today. """ - return "%4d%02d%02d%02d" % \ - (today.year, today.month, today.day, 1) + return "%4d%02d%02d%02d" % (today.year, today.month, today.day, 1) diff --git a/mdb/validators.py b/mdb/validators.py index e2b2a0a..eaff67f 100644 --- a/mdb/validators.py +++ b/mdb/validators.py @@ -2,14 +2,12 @@ from django.core.validators import RegexValidator import re # http://en.wikipedia.org/wiki/Hostname#Restrictions_on_valid_host_names -hostname_re = re.compile(r'^(?!-)[-a-z0-9]+(?=0.15.6,<1.0.0", ] +[tool.ruff] +exclude = ["mdb/migrations/**"] + +[tool.ruff.lint] +select = [ + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # pyflakes + "I", # isort + "UP", # pyupgrade +] +ignore = [ + "E501", # line-too-long +] + [tool.pytest.ini_options] DJANGO_SETTINGS_MODULE = "mdbsite.settings" python_files = ["tests/test_*.py"] diff --git a/tests/conftest.py b/tests/conftest.py index a1b65db..3a135b2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -48,16 +48,15 @@ def db_domain(db, tmp_path): DomainARecord.objects.create(domain=domain, name="mail", target="9.10.11.12") DomainARecord.objects.create(domain=domain, name="www", target="1.2.3.4") DomainAAAARecord.objects.create(domain=domain, name="www", target="2001:db8::1") - DomainCnameRecord.objects.create(domain=domain, name="ftp", target="www.example.org.") + DomainCnameRecord.objects.create( + domain=domain, name="ftp", target="www.example.org." + ) DomainTxtRecord.objects.create(domain=domain, name="@", target='"v=spf1 mx ~all"') return domain def load_command_class(): - path = ( - pathlib.Path(__file__).parent.parent - / "mdb/management/commands/zone-sync.py" - ) + path = pathlib.Path(__file__).parent.parent / "mdb/management/commands/zone-sync.py" spec = importlib.util.spec_from_file_location("zone_sync_command", path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) diff --git a/tests/test_api.py b/tests/test_api.py index 3455241..0b60679 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -20,7 +20,7 @@ def api_client(): @pytest.fixture def user_with_token(db_with_fixtures): - user = User.objects.create(username='admin') + user = User.objects.create(username="admin") token = Token.objects.create(user=user) return user, token @@ -34,10 +34,12 @@ def host(db_with_fixtures): def test_validate_puppet_host_by_secret(api_client, user_with_token, host): _, token = user_with_token data = { - 'pxe_key': host.pxe_key, - 'certname': '{}.{}'.format(host.hostname, host.interface_set.first().domain.domain_name), + "pxe_key": host.pxe_key, + "certname": "{}.{}".format( + host.hostname, host.interface_set.first().domain.domain_name + ), } - api_client.credentials(HTTP_AUTHORIZATION='Token {}'.format(token.key)) - res = api_client.post(reverse('api-validate-host-secret'), data, format='json') + api_client.credentials(HTTP_AUTHORIZATION="Token {}".format(token.key)) + res = api_client.post(reverse("api-validate-host-secret"), data, format="json") assert res.status_code == 200, res.content assert not Host.objects.get(pk=host.pk).pxe_installable diff --git a/tests/test_dhcp_sync.py b/tests/test_dhcp_sync.py index 4b8c382..2989e32 100644 --- a/tests/test_dhcp_sync.py +++ b/tests/test_dhcp_sync.py @@ -9,7 +9,9 @@ def make_config(serial=2, active_serial=1): config = MagicMock() config.serial = serial config.active_serial = active_serial - config.dhcpd_configuration.return_value = "# dhcpd config\nsubnet 10.0.0.0 netmask 255.255.255.0 {}\n" + config.dhcpd_configuration.return_value = ( + "# dhcpd config\nsubnet 10.0.0.0 netmask 255.255.255.0 {}\n" + ) return config @@ -20,6 +22,7 @@ def service(): # --- needs_update --- + def test_needs_update_true_when_serials_differ(service): config = make_config(serial=2, active_serial=1) assert service.needs_update(config) is True @@ -38,6 +41,7 @@ def test_needs_update_true_when_force_even_if_serials_match(): # --- calculate_diff --- + def test_calculate_diff_identical_returns_empty(service): config = make_config() assert service.calculate_diff(config, "line\n", "line\n") == "" @@ -54,6 +58,7 @@ def test_calculate_diff_format(service): # --- write_config --- + def test_write_config_creates_file_with_config_contents(service, tmp_path): config = make_config() output_path = str(tmp_path / "dhcpd.conf") @@ -99,6 +104,7 @@ def test_write_config_diff_reflects_changes(service, tmp_path): # --- restart_service --- + def test_restart_service_returns_true_on_success(service): with patch("mdb.services.dhcp_sync.getstatusoutput", return_value=(0, "")): success, output = service.restart_service() @@ -107,7 +113,9 @@ def test_restart_service_returns_true_on_success(service): def test_restart_service_returns_false_on_failure(service): - with patch("mdb.services.dhcp_sync.getstatusoutput", return_value=(1, "Job failed")): + with patch( + "mdb.services.dhcp_sync.getstatusoutput", return_value=(1, "Job failed") + ): success, output = service.restart_service() assert success is False assert output == "Job failed" diff --git a/tests/test_dhcp_sync_command.py b/tests/test_dhcp_sync_command.py index 89c5443..2e50da1 100644 --- a/tests/test_dhcp_sync_command.py +++ b/tests/test_dhcp_sync_command.py @@ -31,6 +31,7 @@ def make_service(serial=2, active_serial=1, restart_success=True, restart_output # --- handle: nothing to do --- + def test_handle_prints_nothing_to_do_when_no_update_needed(cmd): svc, config = make_service(serial=1, active_serial=1) svc.needs_update.return_value = False @@ -41,6 +42,7 @@ def test_handle_prints_nothing_to_do_when_no_update_needed(cmd): # --- handle: stdout mode --- + def test_handle_prints_config_to_stdout_when_no_output_path(cmd): svc, config = make_service() with patch("mdb.management.commands.dhcp_sync.DhcpSyncService", return_value=svc): @@ -58,6 +60,7 @@ def test_handle_does_not_write_file_when_no_output_path(cmd): # --- handle: file mode --- + def test_handle_calls_write_config_when_output_given(cmd, tmp_path): svc, config = make_service() output_path = str(tmp_path / "dhcpd.conf") @@ -86,6 +89,7 @@ def test_handle_calls_restart_and_notify_when_restart_true(cmd, tmp_path): # --- restart_and_notify --- + def test_restart_and_notify_success_writes_ok(cmd): svc, config = make_service(restart_success=True) with patch("mdb.management.commands.dhcp_sync.discord"): diff --git a/tests/test_discord.py b/tests/test_discord.py index 1e1ada1..ecc6277 100644 --- a/tests/test_discord.py +++ b/tests/test_discord.py @@ -59,16 +59,23 @@ def test_send_json_posts_with_mdb_username(): def test_send_as_file_posts_multipart_with_attachment(): mock_resp = MagicMock() with patch("requests.post", return_value=mock_resp) as mock_post: - discord._send_as_file("https://discord.example/webhook", "Subject", "full content") - assert mock_post.call_args.kwargs.get("files") or "files" in str(mock_post.call_args) + discord._send_as_file( + "https://discord.example/webhook", "Subject", "full content" + ) + assert mock_post.call_args.kwargs.get("files") or "files" in str( + mock_post.call_args + ) mock_resp.raise_for_status.assert_called_once() def test_send_as_file_content_mentions_attachment(): import json + mock_resp = MagicMock() with patch("requests.post", return_value=mock_resp) as mock_post: - discord._send_as_file("https://discord.example/webhook", "MySubject", "full content") + discord._send_as_file( + "https://discord.example/webhook", "MySubject", "full content" + ) data = mock_post.call_args.kwargs["data"] payload = json.loads(data["payload_json"]) assert "MySubject" in payload["content"]