Address PR feedback, ensure that parent GFK underlying fields are not nullable

This changes an existing (within the branch) migration, so if you're
testing this you'll need to back up migrations and re-run.
This commit is contained in:
Jason Novinger 2025-04-10 14:24:38 -05:00
parent fbcbe273b8
commit 06aa7eade4
4 changed files with 44 additions and 15 deletions

View File

@ -571,6 +571,10 @@ class ServiceImportForm(NetBoxModelImportForm):
to_field_name='name', to_field_name='name',
help_text=_('Parent object name') help_text=_('Parent object name')
) )
parent_object_id = forms.IntegerField(
required=False,
help_text=_('Parent object ID'),
)
protocol = CSVChoiceField( protocol = CSVChoiceField(
label=_('Protocol'), label=_('Protocol'),
choices=ServiceProtocolChoices, choices=ServiceProtocolChoices,
@ -586,8 +590,7 @@ class ServiceImportForm(NetBoxModelImportForm):
class Meta: class Meta:
model = Service model = Service
fields = ( fields = (
'parent_object_type', 'ipaddresses', 'name', 'protocol', 'ports', 'description', 'comments', 'ipaddresses', 'name', 'protocol', 'ports', 'description', 'comments', 'tags',
'tags', 'parent_object_id',
) )
def __init__(self, data=None, *args, **kwargs): def __init__(self, data=None, *args, **kwargs):
@ -619,16 +622,20 @@ class ServiceImportForm(NetBoxModelImportForm):
parent = parent_ct.model_class().objects.filter(id=parent_id).first() parent = parent_ct.model_class().objects.filter(id=parent_id).first()
self.cleaned_data['parent'] = parent self.cleaned_data['parent'] = parent
else: else:
# If a parent object type is passed and we've made it to here, then raise a validation # If a parent object type is passed and we've made it here, then raise a validation
# error since an associated parent object or parent object id has nto be passed # error since an associated parent object or parent object id has not been passed
raise forms.ValidationError( raise forms.ValidationError(
_("One of parent or parent_object_id needs to be included with parent_object_type") _("One of parent or parent_object_id must be included with parent_object_type")
) )
# making sure parent is defined. In cases where an import is resulting in an update, the
# import data might not include the parent object and so the above logic might not be
# triggered
parent = self.cleaned_data.get('parent')
for ip_address in self.cleaned_data.get('ipaddresses', []): for ip_address in self.cleaned_data.get('ipaddresses', []):
if not ip_address.assigned_object or getattr(ip_address.assigned_object, 'parent_object') != parent: if not ip_address.assigned_object or getattr(ip_address.assigned_object, 'parent_object') != parent:
raise forms.ValidationError( raise forms.ValidationError(
_("{ip} is not assigned to this device/VM.").format(ip=ip_address) _("{ip} is not assigned to this parent.").format(ip=ip_address)
) )
return self.cleaned_data return self.cleaned_data

View File

@ -18,6 +18,18 @@ class Migration(migrations.Migration):
model_name='service', model_name='service',
name='virtual_machine', name='virtual_machine',
), ),
migrations.AlterField(
model_name='service',
name='parent_object_id',
field=models.PositiveBigIntegerField(),
),
migrations.AlterField(
model_name='service',
name='parent_object_type',
field=models.ForeignKey(
on_delete=models.deletion.PROTECT, related_name='+', to='contenttypes.contenttype'
),
),
migrations.AddIndex( migrations.AddIndex(
model_name='service', model_name='service',
index=models.Index( index=models.Index(

View File

@ -68,13 +68,8 @@ class Service(ContactsMixin, ServiceBase, PrimaryModel):
to='contenttypes.ContentType', to='contenttypes.ContentType',
on_delete=models.PROTECT, on_delete=models.PROTECT,
related_name='+', related_name='+',
blank=True,
null=True
)
parent_object_id = models.PositiveBigIntegerField(
blank=True,
null=True
) )
parent_object_id = models.PositiveBigIntegerField()
parent = GenericForeignKey( parent = GenericForeignKey(
ct_field='parent_object_type', ct_field='parent_object_type',
fk_field='parent_object_id' fk_field='parent_object_id'

View File

@ -1258,9 +1258,24 @@ class IPAddressTestCase(TestCase, ChangeLoggedFilterSetTests):
IPAddress.objects.bulk_create(ipaddresses) IPAddress.objects.bulk_create(ipaddresses)
services = ( services = (
Service(name='Service 1', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[1]), Service(
Service(name='Service 2', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[1]), parent=devices[0],
Service(name='Service 3', protocol=ServiceProtocolChoices.PROTOCOL_TCP, ports=[1]), name='Service 1',
protocol=ServiceProtocolChoices.PROTOCOL_TCP,
ports=[1],
),
Service(
parent=devices[1],
name='Service 2',
protocol=ServiceProtocolChoices.PROTOCOL_TCP,
ports=[1],
),
Service(
parent=devices[2],
name='Service 3',
protocol=ServiceProtocolChoices.PROTOCOL_TCP,
ports=[1],
),
) )
Service.objects.bulk_create(services) Service.objects.bulk_create(services)
services[0].ipaddresses.add(ipaddresses[0]) services[0].ipaddresses.add(ipaddresses[0])