edb/mdb
9
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reformat all files with ruff

Browse Source
This commit is contained in:
ponas committed 2026-03-19 02:25:30 +01:00
1 parent 94c2b03138
commit f4950c6388
22 files changed
+420 -271
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

fabfile.py
Vendored
+16 -10
View File
@@ -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)
manage.py
+1
View File
@@ -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)
mdb/__init__.py
+1 -1
View File
@@ -1 +1 @@
default_app_config = 'mdb.apps.MdbAppConfig'
default_app_config = "mdb.apps.MdbAppConfig"
mdb/admin.py
+100 -42
View File
@@ -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)
mdb/api/serializers.py
+21 -13
View File
@@ -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
mdb/api/views.py
+1 -1
View File
@@ -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})
mdb/apps.py
+2 -2
View File
@@ -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
mdb/forms.py
+6 -6
View File
@@ -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)
mdb/management/commands/dhcp_sync.py
+2 -1
View File
@@ -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),
mdb/models.py
+91 -62
View File
@@ -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"
mdb/pxe.py
+15 -12
View File
@@ -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 = []
mdb/signals.py
+18 -7
View File
@@ -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)
mdb/utils.py
+8 -8
View File
@@ -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)
mdb/validators.py
+6 -8
View File
@@ -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]+(?<!-)$', re.IGNORECASE)
hostname_re = re.compile(r"^(?!-)[-a-z0-9]+(?<!-)$", re.IGNORECASE)
validate_hostname = RegexValidator(
regex=hostname_re,
message='Enter a valid hostname',
code='invalid')
regex=hostname_re, message="Enter a valid hostname", code="invalid"
)
macaddr_re = re.compile(r'^([0-9A-F]{2}[:]){5}([0-9A-F]{2})$', re.IGNORECASE)
macaddr_re = re.compile(r"^([0-9A-F]{2}[:]){5}([0-9A-F]{2})$", re.IGNORECASE)
validate_macaddr = RegexValidator(
regex=macaddr_re,
message='Enter a valid MAC address',
code='invalid')
regex=macaddr_re, message="Enter a valid MAC address", code="invalid"
)
mdbsite/settings.py
+75 -80
View File
@@ -2,140 +2,135 @@ import os
BASE_DIR = os.path.dirname(os.path.dirname(__file__))
DEBUG = os.getenv('DJANGO_DEBUG', True)
DEBUG = os.getenv("DJANGO_DEBUG", True)
ADMINS = (
('System Administrator', 'sysadmin@example.com'),
)
ADMINS = (("System Administrator", "sysadmin@example.com"),)
MANAGERS = ADMINS
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': 'db.sqlite3',
"default": {
"ENGINE": "django.db.backends.sqlite3",
"NAME": "db.sqlite3",
}
}
TIME_ZONE = 'Europe/Oslo'
LANGUAGE_CODE = 'en-us'
TIME_ZONE = "Europe/Oslo"
LANGUAGE_CODE = "en-us"
SITE_ID = 1
USE_I18N = True
DEFAULT_AUTO_FIELD = 'django.db.models.AutoField'
DEFAULT_AUTO_FIELD = "django.db.models.AutoField"
MEDIA_ROOT = os.path.join(BASE_DIR, 'media')
MEDIA_URL = '/media/'
ADMIN_MEDIA_PREFIX = '/media/'
MEDIA_ROOT = os.path.join(BASE_DIR, "media")
MEDIA_URL = "/media/"
ADMIN_MEDIA_PREFIX = "/media/"
# Static
STATIC_ROOT = os.path.join(BASE_DIR, 'staticfiles')
STATIC_URL = '/static/'
STATIC_ROOT = os.path.join(BASE_DIR, "staticfiles")
STATIC_URL = "/static/"
STATICFILES_FINDERS = (
'django.contrib.staticfiles.finders.FileSystemFinder',
'django.contrib.staticfiles.finders.AppDirectoriesFinder',
"django.contrib.staticfiles.finders.FileSystemFinder",
"django.contrib.staticfiles.finders.AppDirectoriesFinder",
)
# Make this unique, and don't share it with anybody.
SECRET_KEY = '@4ps=ebofcpv8sb#y7gm896=l&l*u0u-pgo7bcdj%+(k+2lqnl'
SECRET_KEY = "@4ps=ebofcpv8sb#y7gm896=l&l*u0u-pgo7bcdj%+(k+2lqnl"
MIDDLEWARE = (
'django.middleware.common.CommonMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
"django.middleware.common.CommonMiddleware",
"django.contrib.sessions.middleware.SessionMiddleware",
"django.middleware.csrf.CsrfViewMiddleware",
"django.contrib.auth.middleware.AuthenticationMiddleware",
"django.contrib.messages.middleware.MessageMiddleware",
)
ROOT_URLCONF = 'mdbsite.urls'
ROOT_URLCONF = "mdbsite.urls"
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.contrib.auth.context_processors.auth',
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.template.context_processors.i18n',
'django.template.context_processors.media',
'django.template.context_processors.static',
'django.template.context_processors.tz',
'django.contrib.messages.context_processors.messages',
"BACKEND": "django.template.backends.django.DjangoTemplates",
"DIRS": [],
"APP_DIRS": True,
"OPTIONS": {
"context_processors": [
"django.contrib.auth.context_processors.auth",
"django.template.context_processors.debug",
"django.template.context_processors.request",
"django.template.context_processors.i18n",
"django.template.context_processors.media",
"django.template.context_processors.static",
"django.template.context_processors.tz",
"django.contrib.messages.context_processors.messages",
],
},
},
]
INSTALLED_APPS = (
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.sites',
'django.contrib.messages',
'django.contrib.staticfiles',
'django.contrib.admin',
'rest_framework',
'rest_framework.authtoken',
'django_extensions',
'mdb',
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sessions",
"django.contrib.sites",
"django.contrib.messages",
"django.contrib.staticfiles",
"django.contrib.admin",
"rest_framework",
"rest_framework.authtoken",
"django_extensions",
"mdb",
)
EMAIL_SUBJECT_PREFIX = '[MDB] '
EMAIL_SUBJECT_PREFIX = "[MDB] "
# Logging
LOGGING = {
'version': 1,
'disable_existing_loggers': True,
'formatters': {
'verbose': {
'format': '%(levelname)s %(asctime)s %(module)s '
'%(process)d %(thread)d %(message)s'
"version": 1,
"disable_existing_loggers": True,
"formatters": {
"verbose": {
"format": "%(levelname)s %(asctime)s %(module)s "
"%(process)d %(thread)d %(message)s"
},
},
'handlers': {
'console': {
'level': 'DEBUG',
'class': 'logging.StreamHandler',
'formatter': 'verbose'
"handlers": {
"console": {
"level": "DEBUG",
"class": "logging.StreamHandler",
"formatter": "verbose",
}
},
'loggers': {
'root': {
'level': 'WARNING',
'handlers': ['console'],
"loggers": {
"root": {
"level": "WARNING",
"handlers": ["console"],
},
'django.db.backends': {
'level': 'ERROR',
'handlers': ['console'],
'propagate': False,
"django.db.backends": {
"level": "ERROR",
"handlers": ["console"],
"propagate": False,
},
},
}
# REST Framework
REST_FRAMEWORK = {
'DEFAULT_PERMISSION_CLASSES': ('rest_framework.permissions.IsAuthenticated',),
'PAGINATE_BY': 10
"DEFAULT_PERMISSION_CLASSES": ("rest_framework.permissions.IsAuthenticated",),
"PAGINATE_BY": 10,
}
# PXE
MDB_PXE_TFTP_ROOT = '/var/lib/tftpboot/pxelinux/pxelinux.cfg/'
MDB_PXE_KERNEL = 'ubuntu/14_04/amd64/alternate/linux'
MDB_PXE_INITRD = 'ubuntu/14_04/amd64/alternate/initrd.gz'
MDB_PXE_PRESEED_URL = 'http://158.36.190.194/ubuntu/preseed_1404.cfg'
MDB_PXE_TFTP_ROOT = "/var/lib/tftpboot/pxelinux/pxelinux.cfg/"
MDB_PXE_KERNEL = "ubuntu/14_04/amd64/alternate/linux"
MDB_PXE_INITRD = "ubuntu/14_04/amd64/alternate/initrd.gz"
MDB_PXE_PRESEED_URL = "http://158.36.190.194/ubuntu/preseed_1404.cfg"
# Sentry
SENTRY_DSN = os.getenv('SENTRY_DSN')
SENTRY_ENVIRONMENT = os.getenv('SENTRY_ENVIRONENT', 'dev')
SENTRY_DSN = os.getenv("SENTRY_DSN")
SENTRY_ENVIRONMENT = os.getenv("SENTRY_ENVIRONENT", "dev")
ADMIN_SITE_NAME = 'MDB'
ADMIN_SITE_NAME = "MDB"
try:
from .local_settings import *
mdbsite/urls.py
+7 -3
View File
@@ -7,9 +7,13 @@ from mdb.api.views import HostPXEValidate
admin.autodiscover()
urlpatterns = [
re_path(r'^$', RedirectView.as_view(url='admin/', permanent=True)),
path('admin/', admin.site.urls),
path('api/validate-host-secret/', HostPXEValidate.as_view(), name='api-validate-host-secret')
re_path(r"^$", RedirectView.as_view(url="admin/", permanent=True)),
path("admin/", admin.site.urls),
path(
"api/validate-host-secret/",
HostPXEValidate.as_view(),
name="api-validate-host-secret",
),
]
urlpatterns += staticfiles_urlpatterns()
pyproject.toml
+15
View File
@@ -28,6 +28,21 @@ dev = [
"ruff>=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"]
tests/conftest.py
+4 -5
View File
@@ -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)
tests/test_api.py
+7 -5
View File
@@ -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
tests/test_dhcp_sync.py
+10 -2
View File
@@ -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"
tests/test_dhcp_sync_command.py
+4
View File
@@ -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"):
tests/test_discord.py
+10 -3
View File
@@ -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"]