import datetime import json from collections import defaultdict from decimal import Decimal from unittest.mock import patch import django_filters from django.core.exceptions import ValidationError from django.db import connection from django.db.models import QuerySet from django.test import tag from django.test.utils import CaptureQueriesContext from django.urls import reverse from rest_framework import status from core.models import ObjectChange, ObjectType from dcim.filtersets import SiteFilterSet from dcim.forms import SiteImportForm from dcim.models import Manufacturer, Rack, Site from dcim.tables import SiteTable from extras.choices import * from extras.filters import MissingKeyAwareFilterMixin, missing_key_aware_filter_factory from extras.models import CustomField, CustomFieldChoiceSet from ipam.models import VLAN from netbox.choices import CSVDelimiterChoices, ImportFormatChoices from netbox.context import query_cache from utilities.filters import MultiValueCharFilter, MultiValueMACAddressFilter from utilities.testing import APITestCase, TestCase from virtualization.models import VirtualMachine class CustomFieldTestCase(TestCase): @classmethod def setUpTestData(cls): Site.objects.bulk_create([ Site(name='Site A', slug='site-a'), Site(name='Site B', slug='site-b'), Site(name='Site C', slug='site-c'), ]) cls.object_type = ObjectType.objects.get_for_model(Site) def test_invalid_name(self): """ Try creating a CustomField with an invalid name. """ with self.assertRaises(ValidationError): # Invalid character CustomField(name='?', type=CustomFieldTypeChoices.TYPE_TEXT).full_clean() with self.assertRaises(ValidationError): # Double underscores not permitted CustomField(name='foo__bar', type=CustomFieldTypeChoices.TYPE_TEXT).full_clean() def test_text_field(self): value = 'Foobar!' # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='text_field', type=CustomFieldTypeChoices.TYPE_TEXT, required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) # Assign a value and check that it is saved instance.custom_field_data[cf.name] = value instance.save() instance.refresh_from_db() self.assertEqual(instance.custom_field_data[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_longtext_field(self): value = 'A' * 256 # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='longtext_field', type=CustomFieldTypeChoices.TYPE_LONGTEXT, required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) # Assign a value and check that it is saved instance.custom_field_data[cf.name] = value instance.save() instance.refresh_from_db() self.assertEqual(instance.custom_field_data[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_integer_field(self): # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='integer_field', type=CustomFieldTypeChoices.TYPE_INTEGER, required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) for value in (123456, 0, -123456): # Assign a value and check that it is saved instance.custom_field_data[cf.name] = value instance.save() instance.refresh_from_db() self.assertEqual(instance.custom_field_data[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_decimal_field(self): # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='decimal_field', type=CustomFieldTypeChoices.TYPE_DECIMAL, required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) for value in (123456.54, 0, -123456.78): # Assign a value and check that it is saved instance.custom_field_data[cf.name] = value instance.save() instance.refresh_from_db() self.assertEqual(instance.custom_field_data[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_boolean_field(self): # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='boolean_field', type=CustomFieldTypeChoices.TYPE_INTEGER, required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) for value in (True, False): # Assign a value and check that it is saved instance.custom_field_data[cf.name] = value instance.save() instance.refresh_from_db() self.assertEqual(instance.custom_field_data[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_date_field(self): value = datetime.date(2016, 6, 23) # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='date_field', type=CustomFieldTypeChoices.TYPE_DATE, required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) # Assign a value and check that it is saved instance.custom_field_data[cf.name] = cf.serialize(value) instance.save() instance.refresh_from_db() self.assertEqual(instance.cf[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_datetime_field(self): value = datetime.datetime(2016, 6, 23, 9, 45, 0) # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='date_field', type=CustomFieldTypeChoices.TYPE_DATETIME, required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) # Assign a value and check that it is saved instance.custom_field_data[cf.name] = cf.serialize(value) instance.save() instance.refresh_from_db() self.assertEqual(instance.cf[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_url_field(self): value = 'http://example.com/' # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='url_field', type=CustomFieldTypeChoices.TYPE_URL, required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) # Assign a value and check that it is saved instance.custom_field_data[cf.name] = value instance.save() instance.refresh_from_db() self.assertEqual(instance.custom_field_data[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_json_field(self): value = '{"foo": 1, "bar": 2}' # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='json_field', type=CustomFieldTypeChoices.TYPE_JSON, required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) # Assign a value and check that it is saved instance.custom_field_data[cf.name] = value instance.save() instance.refresh_from_db() self.assertEqual(instance.custom_field_data[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) @tag('regression') def test_json_field_falsy_defaults(self): """Test that falsy JSON default values are properly handled""" falsy_test_cases = [ ({}, 'empty_dict'), ([], 'empty_array'), (0, 'zero'), (False, 'false_bool'), ("", 'empty_string'), ] for default, suffix in falsy_test_cases: with self.subTest(default=default, suffix=suffix): cf = CustomField.objects.create( name=f'json_falsy_{suffix}', type=CustomFieldTypeChoices.TYPE_JSON, default=default, required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.create(name=f'Test Site {suffix}', slug=f'test-site-{suffix}') self.assertIsNotNone(instance.custom_field_data) self.assertIn(cf.name, instance.custom_field_data) instance.refresh_from_db() stored = instance.custom_field_data[cf.name] self.assertEqual(stored, default) @tag('regression') def test_json_field_falsy_to_form_field(self): """Test form field generation preserves falsy defaults""" falsy_test_cases = ( ({}, json.dumps({}), 'empty_dict'), ([], json.dumps([]), 'empty_array'), (0, json.dumps(0), 'zero'), (False, json.dumps(False), 'false_bool'), ("", '""', 'empty_string'), ) for default, expected, suffix in falsy_test_cases: with self.subTest(default=default, expected=expected, suffix=suffix): cf = CustomField.objects.create( name=f'json_falsy_{suffix}', type=CustomFieldTypeChoices.TYPE_JSON, default=default, required=False ) cf.object_types.set([self.object_type]) form_field = cf.to_form_field(set_initial=True) self.assertEqual(form_field.initial, expected) def test_select_field(self): CHOICES = ( ('a', 'Option A'), ('b', 'Option B'), ('c', 'Option C'), ) value = 'a' # Create a set of custom field choices choice_set = CustomFieldChoiceSet.objects.create( name='Custom Field Choice Set 1', extra_choices=CHOICES ) # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='select_field', type=CustomFieldTypeChoices.TYPE_SELECT, required=False, choice_set=choice_set ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) # Assign a value and check that it is saved instance.custom_field_data[cf.name] = value instance.save() instance.refresh_from_db() self.assertEqual(instance.custom_field_data[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_multiselect_field(self): CHOICES = ( ('a', 'Option A'), ('b', 'Option B'), ('c', 'Option C'), ) value = ['a', 'b'] # Create a set of custom field choices choice_set = CustomFieldChoiceSet.objects.create( name='Custom Field Choice Set 1', extra_choices=CHOICES ) # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='multiselect_field', type=CustomFieldTypeChoices.TYPE_MULTISELECT, required=False, choice_set=choice_set ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) # Assign a value and check that it is saved instance.custom_field_data[cf.name] = value instance.save() instance.refresh_from_db() self.assertEqual(instance.custom_field_data[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_choice_set_colors(self): choice_set = CustomFieldChoiceSet( name='Test Choice Set', extra_choices=( ('a', 'Option A'), ('b', 'Option B'), ), choice_colors={ 'a': CustomFieldChoiceColorChoices.RED, 'b': CustomFieldChoiceColorChoices.GREEN, }, ) choice_set.full_clean() self.assertEqual( choice_set.colors, { 'a': CustomFieldChoiceColorChoices.RED, 'b': CustomFieldChoiceColorChoices.GREEN, }, ) def test_choice_set_invalid_color_mapping_value(self): choice_set = CustomFieldChoiceSet( name='Test Choice Set', extra_choices=( ('a', 'Option A'), ('b', 'Option B'), ), choice_colors={'c': CustomFieldChoiceColorChoices.RED}, ) with self.assertRaises(ValidationError) as cm: choice_set.full_clean() self.assertIn('choice_colors', cm.exception.message_dict) def test_choice_set_invalid_color_value(self): choice_set = CustomFieldChoiceSet( name='Test Choice Set', extra_choices=( ('a', 'Option A'), ('b', 'Option B'), ), choice_colors={'a': 'magenta'}, ) with self.assertRaises(ValidationError) as cm: choice_set.full_clean() self.assertIn('choice_colors', cm.exception.message_dict) def test_choice_set_invalid_color_mapping_structure(self): choice_set = CustomFieldChoiceSet( name='Test Choice Set', extra_choices=( ('a', 'Option A'), ('b', 'Option B'), ), choice_colors=['a:red'], ) with self.assertRaises(ValidationError) as cm: choice_set.full_clean() self.assertIn('choice_colors', cm.exception.message_dict) @tag('regression') def test_choice_set_with_base_choices_validates_without_error(self): """Regression test for #22325: base-only choice sets must validate.""" for base in ('IATA', 'ISO_3166', 'UN_LOCODE'): with self.subTest(base=base): choice_set = CustomFieldChoiceSet(name=f'Test {base}', base_choices=base, order_alphabetically=True) choice_set.full_clean() # must not raise choice_set.save() # must not raise (extra_choices is None) def test_remove_selected_choice(self): """ Removing a ChoiceSet choice that is referenced by an object should raise a ValidationError exception. """ CHOICES = ( ('a', 'Option A'), ('b', 'Option B'), ('c', 'Option C'), ('d', 'Option D'), ) # Create a set of custom field choices choice_set = CustomFieldChoiceSet.objects.create( name='Custom Field Choice Set 1', extra_choices=CHOICES ) # Create a select custom field cf = CustomField.objects.create( name='select_field', type=CustomFieldTypeChoices.TYPE_SELECT, required=False, choice_set=choice_set ) cf.object_types.set([self.object_type]) # Create a multi-select custom field cf_multiselect = CustomField.objects.create( name='multiselect_field', type=CustomFieldTypeChoices.TYPE_MULTISELECT, required=False, choice_set=choice_set ) cf_multiselect.object_types.set([self.object_type]) # Assign a choice for both custom fields on an object instance = Site.objects.first() instance.custom_field_data[cf.name] = 'a' instance.custom_field_data[cf_multiselect.name] = ['b', 'c'] instance.save() # Attempting to delete a selected choice should fail with self.assertRaises(ValidationError): choice_set.extra_choices = ( ('b', 'Option B'), ('c', 'Option C'), ('d', 'Option D'), ) choice_set.full_clean() # Attempting to delete either of the multi-select choices should fail with self.assertRaises(ValidationError): choice_set.extra_choices = ( ('a', 'Option A'), ('b', 'Option B'), ('d', 'Option D'), ) choice_set.full_clean() # Removing a non-selected choice should succeed choice_set.extra_choices = ( ('a', 'Option A'), ('b', 'Option B'), ('c', 'Option C'), ) choice_set.full_clean() def test_object_field(self): value = VLAN.objects.create(name='VLAN 1', vid=1).pk # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='object_field', type=CustomFieldTypeChoices.TYPE_OBJECT, related_object_type=ObjectType.objects.get_for_model(VLAN), required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) # Assign a value and check that it is saved instance.custom_field_data[cf.name] = value instance.save() instance.refresh_from_db() self.assertEqual(instance.custom_field_data[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_multiobject_field(self): vlans = ( VLAN(name='VLAN 1', vid=1), VLAN(name='VLAN 2', vid=2), VLAN(name='VLAN 3', vid=3), ) VLAN.objects.bulk_create(vlans) value = [vlan.pk for vlan in vlans] # Create a custom field & check that no initial data is written cf = CustomField.objects.create( name='object_field', type=CustomFieldTypeChoices.TYPE_MULTIOBJECT, related_object_type=ObjectType.objects.get_for_model(VLAN), required=False ) cf.object_types.set([self.object_type]) instance = Site.objects.first() self.assertNotIn(cf.name, instance.custom_field_data) # Assign a value and check that it is saved instance.custom_field_data[cf.name] = value instance.save() instance.refresh_from_db() self.assertEqual(instance.custom_field_data[cf.name], value) # Delete the stored value and check that it is now null instance.custom_field_data.pop(cf.name) instance.save() instance.refresh_from_db() self.assertIsNone(instance.custom_field_data.get(cf.name)) def test_rename_customfield(self): obj_type = ObjectType.objects.get_for_model(Site) FIELD_DATA = 'abc' # Create a custom field cf = CustomField(type=CustomFieldTypeChoices.TYPE_TEXT, name='field1') cf.save() cf.object_types.set([obj_type]) # Assign custom field data to an object site = Site.objects.create( name='Site 1', slug='site-1', custom_field_data={'field1': FIELD_DATA} ) site.refresh_from_db() self.assertEqual(site.custom_field_data['field1'], FIELD_DATA) # Rename the custom field cf.name = 'field2' cf.save() # Check that custom field data on the object has been updated site.refresh_from_db() self.assertNotIn('field1', site.custom_field_data) self.assertEqual(site.custom_field_data['field2'], FIELD_DATA) @patch('extras.models.customfields.CUSTOMFIELD_DATA_BATCH_SIZE', 2) def test_batched_object_data_updates(self): """ Provisioning, renaming, and removing custom field data is applied in batches. Use a small batch size to ensure the data on every object is updated across multiple batches. """ # The existing sites (created in setUpTestData) span multiple batches of size 2 site_count = Site.objects.count() self.assertGreater(site_count, 2) # Provisioning: a default value is populated onto every existing object cf = CustomField.objects.create( name='batched_field', type=CustomFieldTypeChoices.TYPE_TEXT, default='foo' ) cf.object_types.set([self.object_type]) self.assertEqual( Site.objects.filter(custom_field_data__batched_field='foo').count(), site_count ) # Renaming: the key is renamed on every existing object, preserving its value cf.name = 'renamed_field' cf.save() self.assertEqual( Site.objects.filter(custom_field_data__renamed_field='foo').count(), site_count ) self.assertEqual( Site.objects.filter(custom_field_data__has_key='batched_field').count(), 0 ) # Removal: deleting the field strips the key from every existing object cf.delete() self.assertEqual( Site.objects.filter(custom_field_data__has_key='renamed_field').count(), 0 ) def test_provisioning_writes_nothing_without_a_default(self): """ A field with no default has no value to record, so creating one must not touch any object. """ cf = CustomField.objects.create( name='unset_field', type=CustomFieldTypeChoices.TYPE_TEXT ) with CaptureQueriesContext(connection) as queries: cf.object_types.set([self.object_type]) # No object data is written at all -- the cost of adding a field no longer scales with the # number of objects it applies to self.assertFalse([ q['sql'] for q in queries.captured_queries if q['sql'].lstrip().upper().startswith('UPDATE "DCIM_SITE"'.upper()) ]) self.assertEqual(Site.objects.filter(custom_field_data__has_key='unset_field').count(), 0) for site in Site.objects.all(): self.assertEqual(site.custom_field_data, {}) self.assertIsNone(site.cf['unset_field']) def test_provisioning_applies_a_default_immediately(self): """ A default value, by contrast, must be recorded on every existing object as soon as the field is created -- it has to be filterable straight away, so it cannot be deferred. """ cf = CustomField.objects.create( name='defaulted_field', type=CustomFieldTypeChoices.TYPE_TEXT, default='bar' ) cf.object_types.set([self.object_type]) self.assertEqual( Site.objects.filter(custom_field_data__defaulted_field='bar').count(), Site.objects.count() ) def test_rename_touches_only_objects_holding_a_value(self): """ Renaming rewrites the key only where a value is actually stored. This is what keeps a rename cheap now that objects are no longer provisioned with a placeholder each. """ cf = CustomField.objects.create( name='sparse_field', type=CustomFieldTypeChoices.TYPE_TEXT ) cf.object_types.set([self.object_type]) site = Site.objects.first() site.custom_field_data['sparse_field'] = 'value' site.save() cf.name = 'sparse_renamed' cf.save() self.assertEqual( list( Site.objects.filter(custom_field_data__has_key='sparse_renamed') .values_list('pk', flat=True) ), [site.pk] ) self.assertEqual(Site.objects.filter(custom_field_data__has_key='sparse_field').count(), 0) site.refresh_from_db() self.assertEqual(site.custom_field_data['sparse_renamed'], 'value') def test_removal_from_object_type_purges_data(self): """ Unassigning a field from an object type removes its data from those objects. """ cf = CustomField.objects.create( name='unassigned_field', type=CustomFieldTypeChoices.TYPE_TEXT, default='baz' ) cf.object_types.set([self.object_type]) self.assertEqual( Site.objects.filter(custom_field_data__has_key='unassigned_field').count(), Site.objects.count() ) cf.object_types.remove(self.object_type) self.assertEqual( Site.objects.filter(custom_field_data__has_key='unassigned_field').count(), 0 ) def test_clearing_object_types_purges_data(self): """ clear() unassigns every object type at once and reports no pk_set, so it must be handled before the fact. Its data is removed just as remove()'s is. """ cf = CustomField.objects.create( name='cleared_field', type=CustomFieldTypeChoices.TYPE_TEXT, default='baz' ) cf.object_types.set([self.object_type]) self.assertEqual( Site.objects.filter(custom_field_data__has_key='cleared_field').count(), Site.objects.count() ) cf.object_types.clear() self.assertEqual( Site.objects.filter(custom_field_data__has_key='cleared_field').count(), 0 ) def test_batch_update_excludes_rows_which_no_longer_match(self): """ A caller's filters must constrain the UPDATE as well as the selection of each batch. rename_object_data() builds a jsonb_set() expression which evaluates to NULL for a row not holding the key being renamed, so a row which loses it between the two statements would otherwise have its entire custom_field_data column nulled out. """ cf = CustomField.objects.create( name='drifting_field', type=CustomFieldTypeChoices.TYPE_TEXT ) cf.object_types.set([self.object_type]) sites = list(Site.objects.order_by('pk')) holder, bystander = sites[0], sites[-1] Site.objects.filter(pk=holder.pk).update(custom_field_data={'drifting_field': 'value'}) Site.objects.filter(pk=bystander.pk).update(custom_field_data={'other': 'untouched'}) # Simulate a concurrent write: the batch selection yields a pk which no longer satisfies # the has_key filter by the time the UPDATE is issued. select_pks = QuerySet.values_list injected = [] def inject_stale_pk(self, *args, **kwargs): result = select_pks(self, *args, **kwargs) if self.model is Site and args == ('pk',) and kwargs.get('flat') and not injected: injected.append(bystander.pk) return [*result, bystander.pk] return result with patch.object(QuerySet, 'values_list', inject_stale_pk): cf.name = 'drifted_field' cf.save() self.assertEqual(injected, [bystander.pk], "the stale pk was never injected") # The renamed value landed, and the bystander was left entirely alone holder.refresh_from_db() self.assertEqual(holder.custom_field_data, {'drifted_field': 'value'}) bystander.refresh_from_db() self.assertEqual(bystander.custom_field_data, {'other': 'untouched'}) @staticmethod def order_sites_by(*aliases): """ Order a SiteTable by the given column aliases and return the underlying QuerySet. """ table = SiteTable(Site.objects.all()) table.order_by = aliases return table.data.data def test_table_ordering_groups_objects_with_no_value(self): """ Objects holding no value sort together regardless of whether they store a JSON null or carry no key at all, and numeric fields still sort numerically rather than lexically. """ cf = CustomField.objects.create( name='sort_field', type=CustomFieldTypeChoices.TYPE_INTEGER ) cf.object_types.set([self.object_type]) sites = list(Site.objects.order_by('name')) # Site A holds a value, Site B an explicit null, Site C no key whatsoever Site.objects.filter(pk=sites[0].pk).update(custom_field_data={'sort_field': 20}) Site.objects.filter(pk=sites[1].pk).update(custom_field_data={'sort_field': None}) Site.objects.filter(pk=sites[2].pk).update(custom_field_data={}) extra = Site.objects.create( name='Site D', slug='site-d', custom_field_data={'sort_field': 100} ) ordered = self.order_sites_by('cf_sort_field') self.assertEqual( [s.pk for s in ordered][:2], [sites[0].pk, extra.pk], "20 must sort before 100 (numerically, not lexically) ahead of the empty rows" ) self.assertEqual( {s.pk for s in ordered[2:]}, {sites[1].pk, sites[2].pk}, "the JSON-null and missing-key rows must group together at the end" ) # Reversing the ordering carries the empty rows to the front, as it would SQL nulls ordered = self.order_sites_by('-cf_sort_field') self.assertEqual( {s.pk for s in ordered[:2]}, {sites[1].pk, sites[2].pk}, "the JSON-null and missing-key rows must still group together" ) self.assertEqual([s.pk for s in ordered[2:]], [extra.pk, sites[0].pk]) def test_table_ordering_breaks_ties_by_primary_key(self): """ Rows tying on the sort value -- every object holding no value ties on both sort keys -- must still be totally ordered, or paginated results may skip or repeat rows between page requests. """ cf = CustomField.objects.create( name='sort_field', type=CustomFieldTypeChoices.TYPE_INTEGER ) cf.object_types.set([self.object_type]) # None of these hold a value for the field, so all of them tie Site.objects.bulk_create([ Site(name=f'Tied Site {i}', slug=f'tied-site-{i}') for i in range(1, 11) ]) for alias in ('cf_sort_field', '-cf_sort_field'): ordered = self.order_sites_by(alias) self.assertEqual( ordered.query.order_by[-1], 'pk', "the primary key must be applied as the final sort key" ) # Paging through the results must yield each object exactly once expected = [site.pk for site in ordered] paginated = [] for offset in range(0, len(expected), 4): paginated.extend(site.pk for site in ordered[offset:offset + 4]) self.assertEqual(paginated, expected) def test_table_ordering_composes_with_other_columns(self): """ A custom field column must contribute its sort keys to a multi-column ordering rather than replace it. (The sort parameter is read with getlist(), and a saved TableConfig records an ordering of arbitrary length.) """ cf = CustomField.objects.create( name='sort_field', type=CustomFieldTypeChoices.TYPE_INTEGER ) cf.object_types.set([self.object_type]) sites = list(Site.objects.order_by('name')) # Ordering by the custom field alone would reverse the first two sites Site.objects.filter(pk=sites[0].pk).update(custom_field_data={'sort_field': 2}) Site.objects.filter(pk=sites[1].pk).update(custom_field_data={'sort_field': 1}) Site.objects.filter(pk=sites[2].pk).update(custom_field_data={'sort_field': 3}) ordered = self.order_sites_by('name', 'cf_sort_field') self.assertEqual( [site.pk for site in ordered], [site.pk for site in sites], "the preceding sort key must survive the addition of a custom field column" ) # A column named after the custom field column must likewise still apply Site.objects.update(custom_field_data={'sort_field': 1}) ordered = self.order_sites_by('cf_sort_field', '-name') self.assertEqual( [site.pk for site in ordered], [site.pk for site in reversed(sites)], "the trailing sort key must be applied before the primary key tie breaker" ) def test_table_ordering_tolerates_a_repeated_sort_alias(self): """ The sort parameter is read with getlist(), so the same custom field column can appear in the ordering more than once, applying the same annotation to the queryset twice. """ cf = CustomField.objects.create( name='sort_field', type=CustomFieldTypeChoices.TYPE_INTEGER ) cf.object_types.set([self.object_type]) table = SiteTable(Site.objects.all()) table.order_by = ['cf_sort_field', '-cf_sort_field'] self.assertEqual(len(list(table.rows)), Site.objects.count()) def test_default_value_validation(self): choiceset = CustomFieldChoiceSet.objects.create( name="Test Choice Set", extra_choices=( ('choice1', 'Choice 1'), ('choice2', 'Choice 2'), ) ) site = Site.objects.create(name='Site 1', slug='site-1') object_type = ObjectType.objects.get_for_model(Site) # Text CustomField(name='test', type='text', required=True, default="Default text").full_clean() # Integer CustomField(name='test', type='integer', required=True, default=1).full_clean() with self.assertRaises(ValidationError): CustomField(name='test', type='integer', required=True, default='xxx').full_clean() # Boolean CustomField(name='test', type='boolean', required=True, default=True).full_clean() with self.assertRaises(ValidationError): CustomField(name='test', type='boolean', required=True, default='xxx').full_clean() # Date CustomField(name='test', type='date', required=True, default="2023-02-25").full_clean() with self.assertRaises(ValidationError): CustomField(name='test', type='date', required=True, default='xxx').full_clean() # Datetime CustomField(name='test', type='datetime', required=True, default="2023-02-25 02:02:02").full_clean() with self.assertRaises(ValidationError): CustomField(name='test', type='datetime', required=True, default='xxx').full_clean() # URL CustomField(name='test', type='url', required=True, default="https://www.netbox.dev").full_clean() # JSON CustomField(name='test', type='json', required=True, default='{"test": "object"}').full_clean() # Selection CustomField(name='test', type='select', required=True, choice_set=choiceset, default='choice1').full_clean() with self.assertRaises(ValidationError): CustomField(name='test', type='select', required=True, choice_set=choiceset, default='xxx').full_clean() # Multi-select CustomField( name='test', type='multiselect', required=True, choice_set=choiceset, default=['choice1'] # Single default choice ).full_clean() CustomField( name='test', type='multiselect', required=True, choice_set=choiceset, default=['choice1', 'choice2'] # Multiple default choices ).full_clean() with self.assertRaises(ValidationError): CustomField( name='test', type='multiselect', required=True, choice_set=choiceset, default=['xxx'] ).full_clean() # Object CustomField( name='test', type='object', required=True, related_object_type=object_type, default=site.pk ).full_clean() with (self.assertRaises(ValidationError)): CustomField( name='test', type='object', required=True, related_object_type=object_type, default="xxx" ).full_clean() # Multi-object CustomField( name='test', type='multiobject', required=True, related_object_type=object_type, default=[site.pk] ).full_clean() with self.assertRaises(ValidationError): CustomField( name='test', type='multiobject', required=True, related_object_type=object_type, default=["xxx"] ).full_clean() def test_validation_schema_only_for_json_type(self): schema = { 'type': 'object', 'properties': { 'name': {'type': 'string'}, }, } # Valid: schema on a JSON field CustomField(name='test', type=CustomFieldTypeChoices.TYPE_JSON, validation_schema=schema).full_clean() # Invalid: schema on a non-JSON field with self.assertRaises(ValidationError): CustomField(name='test', type=CustomFieldTypeChoices.TYPE_TEXT, validation_schema=schema).full_clean() with self.assertRaises(ValidationError): CustomField(name='test', type=CustomFieldTypeChoices.TYPE_INTEGER, validation_schema=schema).full_clean() def test_json_schema_default_validation(self): schema = { 'type': 'object', 'properties': { 'name': {'type': 'string'}, }, 'required': ['name'], } # Valid default CustomField( name='test', type=CustomFieldTypeChoices.TYPE_JSON, validation_schema=schema, default={'name': 'test'} ).full_clean() # Invalid default (missing required 'name') with self.assertRaises(ValidationError): CustomField( name='test', type=CustomFieldTypeChoices.TYPE_JSON, validation_schema=schema, default={'age': 25} ).full_clean() class CustomFieldManagerTestCase(TestCase): @classmethod def setUpTestData(cls): object_type = ObjectType.objects.get_for_model(Site) custom_field = CustomField(type=CustomFieldTypeChoices.TYPE_TEXT, name='text_field', default='foo') custom_field.save() custom_field.object_types.set([object_type]) def test_get_for_model(self): self.assertEqual(CustomField.objects.get_for_model(Site).count(), 1) self.assertEqual(CustomField.objects.get_for_model(VirtualMachine).count(), 0) def test_get_for_model_caches_models_with_no_custom_fields(self): """ A model with no custom fields assigned must be served from the request cache like any other. An empty QuerySet is falsy, so testing the cached value for truthiness would treat it as a miss and re-query on every call. """ token = query_cache.set(defaultdict(dict)) self.addCleanup(query_cache.reset, token) # Site has one custom field assigned, VirtualMachine none for model in (Site, VirtualMachine): # Prime the cache, iterating so that the QuerySet's own result cache is populated too list(CustomField.objects.get_for_model(model)) with self.assertNumQueries(0): list(CustomField.objects.get_for_model(model)) class CustomFieldAPITestCase(APITestCase): @classmethod def setUpTestData(cls): object_type = ObjectType.objects.get_for_model(Site) # Create some VLANs vlans = ( VLAN(name='VLAN 1', vid=1), VLAN(name='VLAN 2', vid=2), VLAN(name='VLAN 3', vid=3), VLAN(name='VLAN 4', vid=4), VLAN(name='VLAN 5', vid=5), ) VLAN.objects.bulk_create(vlans) # Create a set of custom field choices choice_set = CustomFieldChoiceSet.objects.create( name='Custom Field Choice Set 1', extra_choices=(('foo', 'Foo'), ('bar', 'Bar'), ('baz', 'Baz')) ) custom_fields = ( CustomField( type=CustomFieldTypeChoices.TYPE_TEXT, name='text_field', default='foo' ), CustomField( type=CustomFieldTypeChoices.TYPE_LONGTEXT, name='longtext_field', default='ABC' ), CustomField( type=CustomFieldTypeChoices.TYPE_INTEGER, name='integer_field', default=123 ), CustomField( type=CustomFieldTypeChoices.TYPE_DECIMAL, name='decimal_field', default=123.45 ), CustomField( type=CustomFieldTypeChoices.TYPE_BOOLEAN, name='boolean_field', default=False ), CustomField( type=CustomFieldTypeChoices.TYPE_DATE, name='date_field', default='2020-01-01' ), CustomField( type=CustomFieldTypeChoices.TYPE_DATETIME, name='datetime_field', default='2020-01-01T01:23:45' ), CustomField( type=CustomFieldTypeChoices.TYPE_URL, name='url_field', default='http://example.com/1' ), CustomField( type=CustomFieldTypeChoices.TYPE_JSON, name='json_field', default='{"x": "y"}' ), CustomField( type=CustomFieldTypeChoices.TYPE_SELECT, name='select_field', default='foo', choice_set=choice_set ), CustomField( type=CustomFieldTypeChoices.TYPE_MULTISELECT, name='multiselect_field', default=['foo'], choice_set=choice_set, ), CustomField( type=CustomFieldTypeChoices.TYPE_OBJECT, name='object_field', related_object_type=ObjectType.objects.get_for_model(VLAN), default=vlans[0].pk, ), CustomField( type=CustomFieldTypeChoices.TYPE_MULTIOBJECT, name='multiobject_field', related_object_type=ObjectType.objects.get_for_model(VLAN), default=[vlans[0].pk, vlans[1].pk], ), ) for cf in custom_fields: cf.save() cf.object_types.set([object_type]) # Create some sites *after* creating the custom fields. This ensures that # default values are not set for the assigned objects. sites = ( Site(name='Site 1', slug='site-1'), Site(name='Site 2', slug='site-2'), ) Site.objects.bulk_create(sites) # Assign custom field values for site 2 sites[1].custom_field_data = { custom_fields[0].name: 'bar', custom_fields[1].name: 'DEF', custom_fields[2].name: 456, custom_fields[3].name: Decimal('456.78'), custom_fields[4].name: True, custom_fields[5].name: '2020-01-02', custom_fields[6].name: '2020-01-02 12:00:00', custom_fields[7].name: 'http://example.com/2', custom_fields[8].name: '{"foo": 1, "bar": 2}', custom_fields[9].name: 'bar', custom_fields[10].name: ['bar', 'baz'], custom_fields[11].name: vlans[1].pk, custom_fields[12].name: [vlans[2].pk, vlans[3].pk], } sites[1].save() def test_get_custom_fields(self): TYPES = { CustomFieldTypeChoices.TYPE_TEXT: 'string', CustomFieldTypeChoices.TYPE_LONGTEXT: 'string', CustomFieldTypeChoices.TYPE_INTEGER: 'integer', CustomFieldTypeChoices.TYPE_DECIMAL: 'decimal', CustomFieldTypeChoices.TYPE_BOOLEAN: 'boolean', CustomFieldTypeChoices.TYPE_DATE: 'string', CustomFieldTypeChoices.TYPE_DATETIME: 'string', CustomFieldTypeChoices.TYPE_URL: 'string', CustomFieldTypeChoices.TYPE_JSON: 'object', CustomFieldTypeChoices.TYPE_SELECT: 'string', CustomFieldTypeChoices.TYPE_MULTISELECT: 'array', CustomFieldTypeChoices.TYPE_OBJECT: 'object', CustomFieldTypeChoices.TYPE_MULTIOBJECT: 'array', } self.add_permissions('extras.view_customfield') url = reverse('extras-api:customfield-list') response = self.client.get(url, **self.header) self.assertEqual(response.data['count'], len(TYPES)) # Validate data types for customfield in response.data['results']: cf_type = customfield['type']['value'] self.assertEqual(customfield['data_type'], TYPES[cf_type]) def test_get_single_object_without_custom_field_data(self): """ Validate that custom fields are present on an object even if it has no values defined. """ site1 = Site.objects.get(name='Site 1') url = reverse('dcim-api:site-detail', kwargs={'pk': site1.pk}) self.add_permissions('dcim.view_site') response = self.client.get(url, **self.header) self.assertEqual(response.data['name'], site1.name) self.assertEqual(response.data['custom_fields'], { 'text_field': None, 'longtext_field': None, 'integer_field': None, 'decimal_field': None, 'boolean_field': None, 'date_field': None, 'datetime_field': None, 'url_field': None, 'json_field': None, 'select_field': None, 'multiselect_field': None, 'object_field': None, 'multiobject_field': None, }) def test_get_single_object_with_custom_field_data(self): """ Validate that custom fields are present and correctly set for an object with values defined. """ site2 = Site.objects.get(name='Site 2') site2_cfvs = site2.cf url = reverse('dcim-api:site-detail', kwargs={'pk': site2.pk}) self.add_permissions('dcim.view_site') response = self.client.get(url, **self.header) self.assertEqual(response.data['name'], site2.name) self.assertEqual(response.data['custom_fields']['text_field'], site2_cfvs['text_field']) self.assertEqual(response.data['custom_fields']['longtext_field'], site2_cfvs['longtext_field']) self.assertEqual(response.data['custom_fields']['integer_field'], site2_cfvs['integer_field']) self.assertEqual(response.data['custom_fields']['decimal_field'], site2_cfvs['decimal_field']) self.assertEqual(response.data['custom_fields']['boolean_field'], site2_cfvs['boolean_field']) self.assertEqual(response.data['custom_fields']['date_field'], site2_cfvs['date_field']) self.assertEqual(response.data['custom_fields']['datetime_field'], site2_cfvs['datetime_field']) self.assertEqual(response.data['custom_fields']['url_field'], site2_cfvs['url_field']) self.assertEqual(response.data['custom_fields']['json_field'], site2_cfvs['json_field']) self.assertEqual(response.data['custom_fields']['select_field'], site2_cfvs['select_field']) self.assertEqual(response.data['custom_fields']['multiselect_field'], site2_cfvs['multiselect_field']) self.assertEqual(response.data['custom_fields']['object_field']['id'], site2_cfvs['object_field'].pk) self.assertEqual( [obj['id'] for obj in response.data['custom_fields']['multiobject_field']], [obj.pk for obj in site2_cfvs['multiobject_field']] ) def test_create_single_object_with_defaults(self): """ Create a new site with no specified custom field values and check that it received the default values. """ cf_defaults = { cf.name: cf.default for cf in CustomField.objects.all() } data = { 'name': 'Site 3', 'slug': 'site-3', } url = reverse('dcim-api:site-list') self.add_permissions('dcim.add_site') response = self.client.post(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_201_CREATED) # Validate response data response_cf = response.data['custom_fields'] self.assertEqual(response_cf['text_field'], cf_defaults['text_field']) self.assertEqual(response_cf['longtext_field'], cf_defaults['longtext_field']) self.assertEqual(response_cf['integer_field'], cf_defaults['integer_field']) self.assertEqual(response_cf['decimal_field'], cf_defaults['decimal_field']) self.assertEqual(response_cf['boolean_field'], cf_defaults['boolean_field']) self.assertEqual(response_cf['date_field'].isoformat(), cf_defaults['date_field']) self.assertEqual(response_cf['datetime_field'].isoformat(), cf_defaults['datetime_field']) self.assertEqual(response_cf['url_field'], cf_defaults['url_field']) self.assertEqual(response_cf['json_field'], cf_defaults['json_field']) self.assertEqual(response_cf['select_field'], cf_defaults['select_field']) self.assertEqual(response_cf['multiselect_field'], cf_defaults['multiselect_field']) self.assertEqual(response_cf['object_field']['id'], cf_defaults['object_field']) self.assertEqual( [obj['id'] for obj in response.data['custom_fields']['multiobject_field']], cf_defaults['multiobject_field'] ) # Validate database data site = Site.objects.get(pk=response.data['id']) self.assertEqual(site.custom_field_data['text_field'], cf_defaults['text_field']) self.assertEqual(site.custom_field_data['longtext_field'], cf_defaults['longtext_field']) self.assertEqual(site.custom_field_data['integer_field'], cf_defaults['integer_field']) self.assertEqual(site.custom_field_data['decimal_field'], cf_defaults['decimal_field']) self.assertEqual(site.custom_field_data['boolean_field'], cf_defaults['boolean_field']) self.assertEqual(site.custom_field_data['date_field'], cf_defaults['date_field']) self.assertEqual(site.custom_field_data['datetime_field'], cf_defaults['datetime_field']) self.assertEqual(site.custom_field_data['url_field'], cf_defaults['url_field']) self.assertEqual(site.custom_field_data['json_field'], cf_defaults['json_field']) self.assertEqual(site.custom_field_data['select_field'], cf_defaults['select_field']) self.assertEqual(site.custom_field_data['multiselect_field'], cf_defaults['multiselect_field']) self.assertEqual(site.custom_field_data['object_field'], cf_defaults['object_field']) self.assertEqual(site.custom_field_data['multiobject_field'], cf_defaults['multiobject_field']) def test_create_single_object_with_values(self): """ Create a single new site with a value for each type of custom field. """ data = { 'name': 'Site 3', 'slug': 'site-3', 'custom_fields': { 'text_field': 'bar', 'longtext_field': 'blah blah blah', 'integer_field': 456, 'decimal_field': 456.78, 'boolean_field': True, 'date_field': datetime.date(2020, 1, 2), 'datetime_field': datetime.datetime(2020, 1, 2, 12, 0, 0), 'url_field': 'http://example.com/2', 'json_field': '{"foo": 1, "bar": 2}', 'select_field': 'bar', 'multiselect_field': ['bar', 'baz'], 'object_field': VLAN.objects.get(vid=2).pk, 'multiobject_field': list(VLAN.objects.filter(vid__in=[3, 4]).values_list('pk', flat=True)), }, } url = reverse('dcim-api:site-list') self.add_permissions('dcim.add_site') response = self.client.post(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_201_CREATED) # Validate response data response_cf = response.data['custom_fields'] data_cf = data['custom_fields'] self.assertEqual(response_cf['text_field'], data_cf['text_field']) self.assertEqual(response_cf['longtext_field'], data_cf['longtext_field']) self.assertEqual(response_cf['integer_field'], data_cf['integer_field']) self.assertEqual(response_cf['decimal_field'], data_cf['decimal_field']) self.assertEqual(response_cf['boolean_field'], data_cf['boolean_field']) self.assertEqual(response_cf['date_field'], data_cf['date_field']) self.assertEqual(response_cf['datetime_field'], data_cf['datetime_field']) self.assertEqual(response_cf['url_field'], data_cf['url_field']) self.assertEqual(response_cf['json_field'], data_cf['json_field']) self.assertEqual(response_cf['select_field'], data_cf['select_field']) self.assertEqual(response_cf['multiselect_field'], data_cf['multiselect_field']) self.assertEqual(response_cf['object_field']['id'], data_cf['object_field']) self.assertEqual( [obj['id'] for obj in response_cf['multiobject_field']], data_cf['multiobject_field'] ) # Validate database data site = Site.objects.get(pk=response.data['id']) self.assertEqual(site.custom_field_data['text_field'], data_cf['text_field']) self.assertEqual(site.custom_field_data['longtext_field'], data_cf['longtext_field']) self.assertEqual(site.custom_field_data['integer_field'], data_cf['integer_field']) self.assertEqual(site.custom_field_data['decimal_field'], data_cf['decimal_field']) self.assertEqual(site.custom_field_data['boolean_field'], data_cf['boolean_field']) self.assertEqual(site.cf['date_field'], data_cf['date_field']) self.assertEqual(site.cf['datetime_field'], data_cf['datetime_field']) self.assertEqual(site.custom_field_data['url_field'], data_cf['url_field']) self.assertEqual(site.custom_field_data['json_field'], data_cf['json_field']) self.assertEqual(site.custom_field_data['select_field'], data_cf['select_field']) self.assertEqual(site.custom_field_data['multiselect_field'], data_cf['multiselect_field']) self.assertEqual(site.custom_field_data['object_field'], data_cf['object_field']) self.assertEqual(site.custom_field_data['multiobject_field'], data_cf['multiobject_field']) def test_create_multiple_objects_with_defaults(self): """ Create three new sites with no specified custom field values and check that each received the default custom field values. """ cf_defaults = { cf.name: cf.default for cf in CustomField.objects.all() } data = ( { 'name': 'Site 3', 'slug': 'site-3', }, { 'name': 'Site 4', 'slug': 'site-4', }, { 'name': 'Site 5', 'slug': 'site-5', }, ) url = reverse('dcim-api:site-list') self.add_permissions('dcim.add_site') response = self.client.post(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_201_CREATED) self.assertEqual(len(response.data), len(data)) for i, obj in enumerate(data): # Validate response data response_cf = response.data[i]['custom_fields'] self.assertEqual(response_cf['text_field'], cf_defaults['text_field']) self.assertEqual(response_cf['longtext_field'], cf_defaults['longtext_field']) self.assertEqual(response_cf['integer_field'], cf_defaults['integer_field']) self.assertEqual(response_cf['decimal_field'], cf_defaults['decimal_field']) self.assertEqual(response_cf['boolean_field'], cf_defaults['boolean_field']) self.assertEqual(response_cf['date_field'].isoformat(), cf_defaults['date_field']) self.assertEqual(response_cf['datetime_field'].isoformat(), cf_defaults['datetime_field']) self.assertEqual(response_cf['url_field'], cf_defaults['url_field']) self.assertEqual(response_cf['json_field'], cf_defaults['json_field']) self.assertEqual(response_cf['select_field'], cf_defaults['select_field']) self.assertEqual(response_cf['multiselect_field'], cf_defaults['multiselect_field']) self.assertEqual(response_cf['object_field']['id'], cf_defaults['object_field']) self.assertEqual( [obj['id'] for obj in response_cf['multiobject_field']], cf_defaults['multiobject_field'] ) # Validate database data site = Site.objects.get(pk=response.data[i]['id']) self.assertEqual(site.custom_field_data['text_field'], cf_defaults['text_field']) self.assertEqual(site.custom_field_data['longtext_field'], cf_defaults['longtext_field']) self.assertEqual(site.custom_field_data['integer_field'], cf_defaults['integer_field']) self.assertEqual(site.custom_field_data['decimal_field'], cf_defaults['decimal_field']) self.assertEqual(site.custom_field_data['boolean_field'], cf_defaults['boolean_field']) self.assertEqual(site.custom_field_data['date_field'], cf_defaults['date_field']) self.assertEqual(site.custom_field_data['datetime_field'], cf_defaults['datetime_field']) self.assertEqual(site.custom_field_data['url_field'], cf_defaults['url_field']) self.assertEqual(site.custom_field_data['json_field'], cf_defaults['json_field']) self.assertEqual(site.custom_field_data['select_field'], cf_defaults['select_field']) self.assertEqual(site.custom_field_data['multiselect_field'], cf_defaults['multiselect_field']) self.assertEqual(site.custom_field_data['object_field'], cf_defaults['object_field']) self.assertEqual(site.custom_field_data['multiobject_field'], cf_defaults['multiobject_field']) def test_create_multiple_objects_with_values(self): """ Create a three new sites, each with custom fields defined. """ custom_field_data = { 'text_field': 'bar', 'longtext_field': 'abcdefghij', 'integer_field': 456, 'decimal_field': 456.78, 'boolean_field': True, 'date_field': datetime.date(2020, 1, 2), 'datetime_field': datetime.datetime(2020, 1, 2, 12, 0, 0), 'url_field': 'http://example.com/2', 'json_field': '{"foo": 1, "bar": 2}', 'select_field': 'bar', 'multiselect_field': ['bar', 'baz'], 'object_field': VLAN.objects.get(vid=2).pk, 'multiobject_field': list(VLAN.objects.filter(vid__in=[3, 4]).values_list('pk', flat=True)), } data = ( { 'name': 'Site 3', 'slug': 'site-3', 'custom_fields': custom_field_data, }, { 'name': 'Site 4', 'slug': 'site-4', 'custom_fields': custom_field_data, }, { 'name': 'Site 5', 'slug': 'site-5', 'custom_fields': custom_field_data, }, ) url = reverse('dcim-api:site-list') self.add_permissions('dcim.add_site') response = self.client.post(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_201_CREATED) self.assertEqual(len(response.data), len(data)) for i, obj in enumerate(data): # Validate response data response_cf = response.data[i]['custom_fields'] self.assertEqual(response_cf['text_field'], custom_field_data['text_field']) self.assertEqual(response_cf['longtext_field'], custom_field_data['longtext_field']) self.assertEqual(response_cf['integer_field'], custom_field_data['integer_field']) self.assertEqual(response_cf['decimal_field'], custom_field_data['decimal_field']) self.assertEqual(response_cf['boolean_field'], custom_field_data['boolean_field']) self.assertEqual(response_cf['date_field'], custom_field_data['date_field']) self.assertEqual(response_cf['datetime_field'], custom_field_data['datetime_field']) self.assertEqual(response_cf['url_field'], custom_field_data['url_field']) self.assertEqual(response_cf['json_field'], custom_field_data['json_field']) self.assertEqual(response_cf['select_field'], custom_field_data['select_field']) self.assertEqual(response_cf['multiselect_field'], custom_field_data['multiselect_field']) self.assertEqual(response_cf['object_field']['id'], custom_field_data['object_field']) self.assertEqual( [obj['id'] for obj in response_cf['multiobject_field']], custom_field_data['multiobject_field'] ) # Validate database data site = Site.objects.get(pk=response.data[i]['id']) self.assertEqual(site.custom_field_data['text_field'], custom_field_data['text_field']) self.assertEqual(site.custom_field_data['longtext_field'], custom_field_data['longtext_field']) self.assertEqual(site.custom_field_data['integer_field'], custom_field_data['integer_field']) self.assertEqual(site.custom_field_data['decimal_field'], custom_field_data['decimal_field']) self.assertEqual(site.custom_field_data['boolean_field'], custom_field_data['boolean_field']) self.assertEqual(site.cf['date_field'], custom_field_data['date_field']) self.assertEqual(site.cf['datetime_field'], custom_field_data['datetime_field']) self.assertEqual(site.custom_field_data['url_field'], custom_field_data['url_field']) self.assertEqual(site.custom_field_data['json_field'], custom_field_data['json_field']) self.assertEqual(site.custom_field_data['select_field'], custom_field_data['select_field']) self.assertEqual(site.custom_field_data['multiselect_field'], custom_field_data['multiselect_field']) self.assertEqual(site.custom_field_data['object_field'], custom_field_data['object_field']) self.assertEqual(site.custom_field_data['multiobject_field'], custom_field_data['multiobject_field']) def test_update_single_object_with_values(self): """ Update an object with existing custom field values. Ensure that only the updated custom field values are modified. """ site2 = Site.objects.get(name='Site 2') original_cfvs = {**site2.cf} data = { 'custom_fields': { 'text_field': 'ABCD', 'integer_field': 1234, }, } url = reverse('dcim-api:site-detail', kwargs={'pk': site2.pk}) self.add_permissions('dcim.change_site') response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) # Validate response data response_cf = response.data['custom_fields'] self.assertEqual(response_cf['text_field'], data['custom_fields']['text_field']) self.assertEqual(response_cf['longtext_field'], original_cfvs['longtext_field']) self.assertEqual(response_cf['integer_field'], data['custom_fields']['integer_field']) self.assertEqual(response_cf['decimal_field'], original_cfvs['decimal_field']) self.assertEqual(response_cf['boolean_field'], original_cfvs['boolean_field']) self.assertEqual(response_cf['date_field'], original_cfvs['date_field']) self.assertEqual(response_cf['datetime_field'], original_cfvs['datetime_field']) self.assertEqual(response_cf['url_field'], original_cfvs['url_field']) self.assertEqual(response_cf['json_field'], original_cfvs['json_field']) self.assertEqual(response_cf['select_field'], original_cfvs['select_field']) self.assertEqual(response_cf['multiselect_field'], original_cfvs['multiselect_field']) self.assertEqual(response_cf['object_field']['id'], original_cfvs['object_field'].pk) self.assertListEqual( [obj['id'] for obj in response_cf['multiobject_field']], [obj.pk for obj in original_cfvs['multiobject_field']] ) # Validate database data site2 = Site.objects.get(pk=site2.pk) self.assertEqual(site2.cf['text_field'], data['custom_fields']['text_field']) self.assertEqual(site2.cf['longtext_field'], original_cfvs['longtext_field']) self.assertEqual(site2.cf['integer_field'], data['custom_fields']['integer_field']) self.assertEqual(site2.cf['decimal_field'], original_cfvs['decimal_field']) self.assertEqual(site2.cf['boolean_field'], original_cfvs['boolean_field']) self.assertEqual(site2.cf['date_field'], original_cfvs['date_field']) self.assertEqual(site2.cf['datetime_field'], original_cfvs['datetime_field']) self.assertEqual(site2.cf['url_field'], original_cfvs['url_field']) self.assertEqual(site2.cf['json_field'], original_cfvs['json_field']) self.assertEqual(site2.cf['select_field'], original_cfvs['select_field']) self.assertEqual(site2.cf['multiselect_field'], original_cfvs['multiselect_field']) self.assertEqual(site2.cf['object_field'], original_cfvs['object_field']) self.assertListEqual( list(site2.cf['multiobject_field']), list(original_cfvs['multiobject_field']) ) @tag('regression') def test_update_single_object_rejects_unknown_custom_fields(self): site2 = Site.objects.get(name='Site 2') original_cf_data = {**site2.custom_field_data} url = reverse('dcim-api:site-detail', kwargs={'pk': site2.pk}) self.add_permissions('dcim.change_site') data = { 'custom_fields': { 'text_field': 'valid', 'thisfieldshouldntexist': 'random text here', }, } response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) self.assertIn('custom_fields', response.data) self.assertIn('thisfieldshouldntexist', response.data['custom_fields']) # Ensure the object was not modified site2.refresh_from_db() self.assertEqual(site2.custom_field_data, original_cf_data) @tag('regression') def test_update_single_object_prunes_stale_custom_field_data_from_database_and_postchange_data(self): stale_key = 'thisfieldshouldntexist' stale_value = 'random text here' updated_text_value = 'ABCD' site2 = Site.objects.get(name='Site 2') original_text_value = site2.custom_field_data['text_field'] object_type = ObjectType.objects.get_for_model(Site) # Seed stale custom field data directly in the database to mimic a polluted row. Site.objects.filter(pk=site2.pk).update( custom_field_data={ **site2.custom_field_data, stale_key: stale_value, } ) site2.refresh_from_db() self.assertIn(stale_key, site2.custom_field_data) existing_change_ids = set( ObjectChange.objects.filter( changed_object_type=object_type, changed_object_id=site2.pk, ).values_list('pk', flat=True) ) url = reverse('dcim-api:site-detail', kwargs={'pk': site2.pk}) self.add_permissions('dcim.change_site') data = { 'custom_fields': { 'text_field': updated_text_value, }, } response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) site2.refresh_from_db() self.assertEqual(site2.cf['text_field'], updated_text_value) self.assertNotIn(stale_key, site2.custom_field_data) object_changes = ObjectChange.objects.filter( changed_object_type=object_type, changed_object_id=site2.pk, ).exclude(pk__in=existing_change_ids) self.assertEqual(object_changes.count(), 1) object_change = object_changes.get() self.assertEqual(object_change.prechange_data['custom_fields']['text_field'], original_text_value) self.assertEqual(object_change.postchange_data['custom_fields']['text_field'], updated_text_value) self.assertNotIn(stale_key, object_change.postchange_data['custom_fields']) def test_specify_related_object_by_attr(self): site1 = Site.objects.get(name='Site 1') vlans = VLAN.objects.all()[:3] url = reverse('dcim-api:site-detail', kwargs={'pk': site1.pk}) self.add_permissions('dcim.change_site', 'ipam.view_vlan') # Set related objects by PK data = { 'custom_fields': { 'object_field': vlans[0].pk, 'multiobject_field': [vlans[1].pk, vlans[2].pk], }, } response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) self.assertEqual( response.data['custom_fields']['object_field']['id'], vlans[0].pk ) self.assertListEqual( [obj['id'] for obj in response.data['custom_fields']['multiobject_field']], [vlans[1].pk, vlans[2].pk] ) # Set related objects by name data = { 'custom_fields': { 'object_field': { 'name': vlans[0].name, }, 'multiobject_field': [ { 'name': vlans[1].name }, { 'name': vlans[2].name }, ], }, } response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) self.assertEqual( response.data['custom_fields']['object_field']['id'], vlans[0].pk ) self.assertListEqual( [obj['id'] for obj in response.data['custom_fields']['multiobject_field']], [vlans[1].pk, vlans[2].pk] ) # Clear related objects data = { 'custom_fields': { 'object_field': None, 'multiobject_field': [], }, } response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) self.assertIsNone(response.data['custom_fields']['object_field']) self.assertListEqual(response.data['custom_fields']['multiobject_field'], []) def test_minimum_maximum_values_validation(self): site2 = Site.objects.get(name='Site 2') url = reverse('dcim-api:site-detail', kwargs={'pk': site2.pk}) self.add_permissions('dcim.change_site') cf_integer = CustomField.objects.get(name='integer_field') cf_integer.validation_minimum = 10 cf_integer.validation_maximum = 20 cf_integer.save() data = {'custom_fields': {'integer_field': 9}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) data = {'custom_fields': {'integer_field': 21}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) data = {'custom_fields': {'integer_field': 15}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) def test_regex_validation(self): site2 = Site.objects.get(name='Site 2') url = reverse('dcim-api:site-detail', kwargs={'pk': site2.pk}) self.add_permissions('dcim.change_site') cf_text = CustomField.objects.get(name='text_field') cf_text.validation_regex = r'^[A-Z]{3}$' # Three uppercase letters cf_text.save() data = {'custom_fields': {'text_field': 'ABC123'}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) data = {'custom_fields': {'text_field': 'abc'}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) data = {'custom_fields': {'text_field': 'ABC'}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) def test_url_regex_validation(self): """ Test that validation_regex is applied to URL custom fields (fixes #20498). """ site2 = Site.objects.get(name='Site 2') url = reverse('dcim-api:site-detail', kwargs={'pk': site2.pk}) self.add_permissions('dcim.change_site') cf_url = CustomField.objects.get(name='url_field') cf_url.validation_regex = r'^https://' # Require HTTPS cf_url.save() # Test invalid URL (http instead of https) data = {'custom_fields': {'url_field': 'http://example.com'}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) # Test valid URL (https) data = {'custom_fields': {'url_field': 'https://example.com'}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) def test_json_schema_validation(self): site2 = Site.objects.get(name='Site 2') url = reverse('dcim-api:site-detail', kwargs={'pk': site2.pk}) self.add_permissions('dcim.change_site') cf_json = CustomField.objects.get(name='json_field') cf_json.validation_schema = { 'type': 'object', 'properties': { 'name': {'type': 'string'}, 'age': {'type': 'integer'}, }, 'required': ['name'], } cf_json.save() # Invalid: missing required 'name' property data = {'custom_fields': {'json_field': {'age': 25}}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) # Invalid: 'age' is not an integer data = {'custom_fields': {'json_field': {'name': 'test', 'age': 'not_an_int'}}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) # Valid: conforms to schema data = {'custom_fields': {'json_field': {'name': 'test', 'age': 25}}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) # Valid: null value (schema not enforced on empty) data = {'custom_fields': {'json_field': None}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) def test_uniqueness_validation(self): # Create a unique custom field cf_text = CustomField.objects.get(name='text_field') cf_text.unique = True cf_text.save() # Set a value on site 1 site1 = Site.objects.get(name='Site 1') site1.custom_field_data['text_field'] = 'ABC123' site1.save() site2 = Site.objects.get(name='Site 2') url = reverse('dcim-api:site-detail', kwargs={'pk': site2.pk}) self.add_permissions('dcim.change_site') data = {'custom_fields': {'text_field': 'ABC123'}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) data = {'custom_fields': {'text_field': 'DEF456'}} response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) class CustomFieldImportTestCase(TestCase): user_permissions = ( 'dcim.view_site', 'dcim.add_site', ) @classmethod def setUpTestData(cls): # Create a set of custom field choices choice_set = CustomFieldChoiceSet.objects.create( name='Custom Field Choice Set 1', extra_choices=( ('a', 'Option A'), ('b', 'Option B'), ('c', 'Option C'), ) ) custom_fields = ( CustomField(name='text', type=CustomFieldTypeChoices.TYPE_TEXT), CustomField(name='longtext', type=CustomFieldTypeChoices.TYPE_LONGTEXT), CustomField(name='integer', type=CustomFieldTypeChoices.TYPE_INTEGER), CustomField(name='decimal', type=CustomFieldTypeChoices.TYPE_DECIMAL), CustomField(name='boolean', type=CustomFieldTypeChoices.TYPE_BOOLEAN), CustomField(name='date', type=CustomFieldTypeChoices.TYPE_DATE), CustomField(name='datetime', type=CustomFieldTypeChoices.TYPE_DATETIME), CustomField(name='url', type=CustomFieldTypeChoices.TYPE_URL), CustomField(name='json', type=CustomFieldTypeChoices.TYPE_JSON), CustomField(name='select', type=CustomFieldTypeChoices.TYPE_SELECT, choice_set=choice_set), CustomField(name='multiselect', type=CustomFieldTypeChoices.TYPE_MULTISELECT, choice_set=choice_set), ) for cf in custom_fields: cf.save() cf.object_types.set([ObjectType.objects.get_for_model(Site)]) def test_import(self): """ Import a Site in CSV format, including a value for each CustomField. """ data = ( ( 'name', 'slug', 'status', 'cf_text', 'cf_longtext', 'cf_integer', 'cf_decimal', 'cf_boolean', 'cf_date', 'cf_datetime', 'cf_url', 'cf_json', 'cf_select', 'cf_multiselect', ), ( 'Site 1', 'site-1', 'active', 'ABC', 'Foo', '123', '123.45', 'True', '2020-01-01', '2020-01-01 12:00:00', 'http://example.com/1', '{"foo": 123}', 'a', '"a,b"', ), ( 'Site 2', 'site-2', 'active', 'DEF', 'Bar', '456', '456.78', 'False', '2020-01-02', '2020-01-02 12:00:00', 'http://example.com/2', '{"bar": 456}', 'b', '"b,c"', ), ('Site 3', 'site-3', 'active', '', '', '', '', '', '', '', '', '', '', ''), ) csv_data = '\n'.join(','.join(row) for row in data) response = self.client.post(reverse('dcim:site_bulk_import'), { 'data': csv_data, 'format': ImportFormatChoices.CSV, 'csv_delimiter': CSVDelimiterChoices.AUTO, }) self.assertEqual(response.status_code, 302) self.assertEqual(Site.objects.count(), 3) # Validate data for site 1 site1 = Site.objects.get(name='Site 1') self.assertEqual(len(site1.custom_field_data), 11) self.assertEqual(site1.custom_field_data['text'], 'ABC') self.assertEqual(site1.custom_field_data['longtext'], 'Foo') self.assertEqual(site1.custom_field_data['integer'], 123) self.assertEqual(site1.custom_field_data['decimal'], 123.45) self.assertEqual(site1.custom_field_data['boolean'], True) self.assertEqual(site1.cf['date'].isoformat(), '2020-01-01') self.assertEqual(site1.cf['datetime'].isoformat(), '2020-01-01T12:00:00+00:00') self.assertEqual(site1.custom_field_data['url'], 'http://example.com/1') self.assertEqual(site1.custom_field_data['json'], {"foo": 123}) self.assertEqual(site1.custom_field_data['select'], 'a') self.assertEqual(site1.custom_field_data['multiselect'], ['a', 'b']) # Validate data for site 2 site2 = Site.objects.get(name='Site 2') self.assertEqual(len(site2.custom_field_data), 11) self.assertEqual(site2.custom_field_data['text'], 'DEF') self.assertEqual(site2.custom_field_data['longtext'], 'Bar') self.assertEqual(site2.custom_field_data['integer'], 456) self.assertEqual(site2.custom_field_data['decimal'], 456.78) self.assertEqual(site2.custom_field_data['boolean'], False) self.assertEqual(site2.cf['date'].isoformat(), '2020-01-02') self.assertEqual(site2.cf['datetime'].isoformat(), '2020-01-02T12:00:00+00:00') self.assertEqual(site2.custom_field_data['url'], 'http://example.com/2') self.assertEqual(site2.custom_field_data['json'], {"bar": 456}) self.assertEqual(site2.custom_field_data['select'], 'b') self.assertEqual(site2.custom_field_data['multiselect'], ['b', 'c']) # No custom field data should be set for site 3 site3 = Site.objects.get(name='Site 3') self.assertFalse(any(site3.custom_field_data.values())) def test_import_missing_required(self): """ Attempt to import an object missing a required custom field. """ # Set one of our CustomFields to required CustomField.objects.filter(name='text').update(required=True) form_data = { 'name': 'Site 1', 'slug': 'site-1', } form = SiteImportForm(data=form_data) self.assertFalse(form.is_valid()) self.assertIn('cf_text', form.errors) def test_import_invalid_choice(self): """ Attempt to import an object with an invalid choice selection. """ form_data = { 'name': 'Site 1', 'slug': 'site-1', 'cf_select': 'Choice X' } form = SiteImportForm(data=form_data) self.assertFalse(form.is_valid()) self.assertIn('cf_select', form.errors) class CustomFieldModelTestCase(TestCase): @classmethod def setUpTestData(cls): cf1 = CustomField(type=CustomFieldTypeChoices.TYPE_TEXT, name='foo') cf1.save() cf1.object_types.set([ObjectType.objects.get_for_model(Site)]) cf2 = CustomField(type=CustomFieldTypeChoices.TYPE_TEXT, name='bar') cf2.save() cf2.object_types.set([ObjectType.objects.get_for_model(Rack)]) def test_cf_data(self): """ Check that custom field data is present on the instance immediately after being set and after being fetched from the database. """ site = Site(name='Test Site', slug='test-site') # Check custom field data on new instance site.custom_field_data['foo'] = 'abc' self.assertEqual(site.cf['foo'], 'abc') # Check custom field data from database site.save() site = Site.objects.get(name='Test Site') self.assertEqual(site.cf['foo'], 'abc') def test_invalid_data(self): """ Any invalid or stale custom field data should be removed from the instance. """ site = Site(name='Test Site', slug='test-site') # Set custom field data site.custom_field_data['foo'] = 'abc' site.custom_field_data['bar'] = 'def' site.clean() self.assertIn('foo', site.custom_field_data) self.assertNotIn('bar', site.custom_field_data) def test_missing_required_field(self): """ Check that a ValidationError is raised if any required custom fields are not present. """ cf3 = CustomField(type=CustomFieldTypeChoices.TYPE_TEXT, name='baz', required=True) cf3.save() cf3.object_types.set([ObjectType.objects.get_for_model(Site)]) site = Site(name='Test Site', slug='test-site') # Set custom field data with a required field omitted site.custom_field_data['foo'] = 'abc' with self.assertRaises(ValidationError): site.clean() site.custom_field_data['baz'] = 'def' site.clean() def test_required_field_enforced_on_existing_objects(self): """ Adding a required custom field invalidates the objects which already exist, whether they carry no key for it -- the normal state now that empty values are not provisioned -- or an explicit null. Both are rejected, as they were before: every object then held a materialized null, which CustomField.validate() rejects for a required field. """ site = Site.objects.create(name='Test Site', slug='test-site') cf = CustomField(type=CustomFieldTypeChoices.TYPE_TEXT, name='req', required=True) cf.save() cf.object_types.set([ObjectType.objects.get_for_model(Site)]) # No value was provisioned onto the existing object site.refresh_from_db() self.assertNotIn('req', site.custom_field_data) with self.assertRaises(ValidationError): site.clean() # An explicit null is rejected identically site.custom_field_data['req'] = None with self.assertRaises(ValidationError): site.clean() site.custom_field_data['req'] = 'value' site.clean() class MissingKeyAwareFilterTestCase(TestCase): """ MissingKeyAwareFilterMixin reimplements MultipleChoiceFilter.filter() for the negated case, so it may only be mixed into a class which inherits that method unmodified and which does not filter conjoined. Both constraints are enforced, as violating either would yield a wrong result set rather than an error. """ def test_factory_rejects_a_class_which_defines_filter(self): # MultiValueMACAddressFilter overrides filter() to swallow ValidationError with self.assertRaises(TypeError): missing_key_aware_filter_factory(MultiValueMACAddressFilter) # BooleanFilter does not inherit MultipleChoiceFilter.filter() at all with self.assertRaises(TypeError): missing_key_aware_filter_factory(django_filters.BooleanFilter) def test_factory_accepts_a_class_which_inherits_filter(self): filter_class = missing_key_aware_filter_factory(MultiValueCharFilter) self.assertTrue(issubclass(filter_class, MissingKeyAwareFilterMixin)) self.assertTrue(issubclass(filter_class, MultiValueCharFilter)) # The factory is cached, so a class yields a single stable subclass self.assertIs(filter_class, missing_key_aware_filter_factory(MultiValueCharFilter)) def test_conjoined_filtering_is_rejected(self): filter_class = missing_key_aware_filter_factory(MultiValueCharFilter) filter_class(field_name='custom_field_data__foo') filter_class(field_name='custom_field_data__foo', conjoined=False) with self.assertRaises(TypeError): filter_class(field_name='custom_field_data__foo', conjoined=True) def test_every_supported_custom_field_type_satisfies_the_constraints(self): """ The filter classes CustomField.to_filter() selects must all remain admissible. """ for cf_type in CustomFieldTypeChoices.values(): with self.subTest(cf_type): cf = CustomField(name='test', type=cf_type) # Raises TypeError if the selected filter class violates a constraint cf.to_filter() cf.to_filter(lookup_expr='empty') class CustomFieldModelFilterTestCase(TestCase): queryset = Site.objects.all() filterset = SiteFilterSet @classmethod def setUpTestData(cls): object_type = ObjectType.objects.get_for_model(Site) manufacturers = Manufacturer.objects.bulk_create(( Manufacturer(name='Manufacturer 1', slug='manufacturer-1'), Manufacturer(name='Manufacturer 2', slug='manufacturer-2'), Manufacturer(name='Manufacturer 3', slug='manufacturer-3'), Manufacturer(name='Manufacturer 4', slug='manufacturer-4'), )) choice_set = CustomFieldChoiceSet.objects.create( name='Custom Field Choice Set 1', extra_choices=(('a', 'A'), ('b', 'B'), ('c', 'C')) ) # Integer filtering cf = CustomField(name='cf1', type=CustomFieldTypeChoices.TYPE_INTEGER) cf.save() cf.object_types.set([object_type]) # Decimal filtering cf = CustomField(name='cf2', type=CustomFieldTypeChoices.TYPE_DECIMAL) cf.save() cf.object_types.set([object_type]) # Boolean filtering cf = CustomField(name='cf3', type=CustomFieldTypeChoices.TYPE_BOOLEAN) cf.save() cf.object_types.set([object_type]) # Exact text filtering cf = CustomField( name='cf4', type=CustomFieldTypeChoices.TYPE_TEXT, filter_logic=CustomFieldFilterLogicChoices.FILTER_EXACT ) cf.save() cf.object_types.set([object_type]) # Loose text filtering cf = CustomField( name='cf5', type=CustomFieldTypeChoices.TYPE_TEXT, filter_logic=CustomFieldFilterLogicChoices.FILTER_LOOSE ) cf.save() cf.object_types.set([object_type]) # Date filtering cf = CustomField(name='cf6', type=CustomFieldTypeChoices.TYPE_DATE) cf.save() cf.object_types.set([object_type]) # Exact URL filtering cf = CustomField( name='cf7', type=CustomFieldTypeChoices.TYPE_URL, filter_logic=CustomFieldFilterLogicChoices.FILTER_EXACT ) cf.save() cf.object_types.set([object_type]) # Loose URL filtering cf = CustomField( name='cf8', type=CustomFieldTypeChoices.TYPE_URL, filter_logic=CustomFieldFilterLogicChoices.FILTER_LOOSE ) cf.save() cf.object_types.set([object_type]) # Selection filtering cf = CustomField( name='cf9', type=CustomFieldTypeChoices.TYPE_SELECT, choice_set=choice_set ) cf.save() cf.object_types.set([object_type]) # Multiselect filtering cf = CustomField( name='cf10', type=CustomFieldTypeChoices.TYPE_MULTISELECT, choice_set=choice_set ) cf.save() cf.object_types.set([object_type]) # Object filtering cf = CustomField( name='cf11', type=CustomFieldTypeChoices.TYPE_OBJECT, related_object_type=ObjectType.objects.get_for_model(Manufacturer) ) cf.save() cf.object_types.set([object_type]) # Multi-object filtering cf = CustomField( name='cf12', type=CustomFieldTypeChoices.TYPE_MULTIOBJECT, related_object_type=ObjectType.objects.get_for_model(Manufacturer) ) cf.save() cf.object_types.set([object_type]) Site.objects.bulk_create([ Site(name='Site 1', slug='site-1', custom_field_data={ 'cf1': 100, 'cf2': 100.1, 'cf3': True, 'cf4': 'foo', 'cf5': 'foo', 'cf6': '2016-06-26', 'cf7': 'http://a.example.com', 'cf8': 'http://a.example.com', 'cf9': 'A', 'cf10': ['A', 'B'], 'cf11': manufacturers[0].pk, 'cf12': [manufacturers[0].pk, manufacturers[3].pk], }), Site(name='Site 2', slug='site-2', custom_field_data={ 'cf1': 200, 'cf2': 200.2, 'cf3': True, 'cf4': 'foobar', 'cf5': 'foobar', 'cf6': '2016-06-27', 'cf7': 'http://b.example.com', 'cf8': 'http://b.example.com', 'cf9': 'B', 'cf10': ['B', 'C'], 'cf11': manufacturers[1].pk, 'cf12': [manufacturers[1].pk, manufacturers[3].pk], }), Site(name='Site 3', slug='site-3', custom_field_data={ 'cf1': 300, 'cf2': 300.3, 'cf3': False, 'cf4': 'bar', 'cf5': 'bar', 'cf6': '2016-06-28', 'cf7': 'http://c.example.com', 'cf8': 'http://c.example.com', 'cf9': 'C', 'cf10': None, 'cf11': manufacturers[2].pk, 'cf12': [manufacturers[2].pk, manufacturers[3].pk], }), # Carries no custom field data at all. Negated lookups ("is not x") match it, as they # do an object holding an explicit null; see MissingKeyAwareFilterMixin. Site(name='Site 4', slug='site-4'), ]) def test_filter_integer(self): self.assertEqual(self.filterset({'cf_cf1': [100, 200]}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf1__n': [200]}, self.queryset).qs.count(), 3) self.assertEqual(self.filterset({'cf_cf1__gt': [200]}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf1__gte': [200]}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf1__lt': [200]}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf1__lte': [200]}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf1__empty': True}, self.queryset).qs.count(), 1) def test_filter_decimal(self): self.assertEqual(self.filterset({'cf_cf2': [100.1, 200.2]}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf2__n': [200.2]}, self.queryset).qs.count(), 3) self.assertEqual(self.filterset({'cf_cf2__gt': [200.2]}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf2__gte': [200.2]}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf2__lt': [200.2]}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf2__lte': [200.2]}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf2__empty': True}, self.queryset).qs.count(), 1) def test_filter_boolean(self): self.assertEqual(self.filterset({'cf_cf3': True}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf3': False}, self.queryset).qs.count(), 1) def test_filter_text_strict(self): self.assertEqual(self.filterset({'cf_cf4': ['foo']}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf4__n': ['foo']}, self.queryset).qs.count(), 3) self.assertEqual(self.filterset({'cf_cf4__ic': ['foo']}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf4__nic': ['foo']}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf4__isw': ['foo']}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf4__nisw': ['foo']}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf4__iew': ['bar']}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf4__niew': ['bar']}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf4__ie': ['FOO']}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf4__nie': ['FOO']}, self.queryset).qs.count(), 3) self.assertEqual(self.filterset({'cf_cf4__empty': True}, self.queryset).qs.count(), 1) def test_filter_text_loose(self): self.assertEqual(self.filterset({'cf_cf5': ['foo']}, self.queryset).qs.count(), 2) def test_filter_date(self): self.assertEqual(self.filterset({'cf_cf6': ['2016-06-26', '2016-06-27']}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf6__n': ['2016-06-27']}, self.queryset).qs.count(), 3) self.assertEqual(self.filterset({'cf_cf6__gt': ['2016-06-27']}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf6__gte': ['2016-06-27']}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf6__lt': ['2016-06-27']}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf6__lte': ['2016-06-27']}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf6__empty': True}, self.queryset).qs.count(), 1) def test_filter_url_strict(self): self.assertEqual( self.filterset({'cf_cf7': ['http://a.example.com', 'http://b.example.com']}, self.queryset).qs.count(), 2 ) self.assertEqual(self.filterset({'cf_cf7__n': ['http://b.example.com']}, self.queryset).qs.count(), 3) self.assertEqual(self.filterset({'cf_cf7__ic': ['b']}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf7__nic': ['b']}, self.queryset).qs.count(), 3) self.assertEqual(self.filterset({'cf_cf7__isw': ['http://']}, self.queryset).qs.count(), 3) self.assertEqual(self.filterset({'cf_cf7__nisw': ['http://']}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf7__iew': ['.com']}, self.queryset).qs.count(), 3) self.assertEqual(self.filterset({'cf_cf7__niew': ['.com']}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf7__ie': ['HTTP://A.EXAMPLE.COM']}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf7__nie': ['HTTP://A.EXAMPLE.COM']}, self.queryset).qs.count(), 3) self.assertEqual(self.filterset({'cf_cf7__empty': True}, self.queryset).qs.count(), 1) def test_filter_url_loose(self): self.assertEqual(self.filterset({'cf_cf8': ['example.com']}, self.queryset).qs.count(), 3) def test_filter_negation_matches_unset_values(self): """ A negated lookup must match an object which holds no value for the field, whether that is recorded as an explicit null or by the absence of the key; see MissingKeyAwareFilterMixin. """ no_key = Site.objects.get(slug='site-4') explicit_null = Site.objects.create(name='Site 5', slug='site-5', custom_field_data={ 'cf1': None, 'cf4': None, 'cf6': None, 'cf7': None, }) for filter_name, value in ( ('cf_cf1__n', 100), ('cf_cf4__n', 'foo'), ('cf_cf4__nic', 'foo'), ('cf_cf4__nisw', 'foo'), ('cf_cf4__niew', 'bar'), ('cf_cf4__nie', 'FOO'), ('cf_cf6__n', '2016-06-26'), ('cf_cf7__n', 'http://a.example.com'), ('cf_cf7__nic', 'a'), ('cf_cf7__nisw', 'http://'), ('cf_cf7__niew', '.com'), ): with self.subTest(filter_name): pks = set( self.filterset({filter_name: [value]}, self.queryset).qs.values_list('pk', flat=True) ) self.assertIn(no_key.pk, pks, "an object carrying no key must match") self.assertIn(explicit_null.pk, pks, "an object holding a null must match") def test_filter_null_sentinel_matches_unset_values(self): """ The null sentinel (FILTERS_NULL_CHOICE_VALUE) asks for the objects holding no value, which must include those carrying no key as well as those holding an explicit null. Negating it must therefore return exactly the objects which do hold a value -- and in particular must not return the ones it is being asked to exclude. Only string-backed field types are exercised: a numeric or date field rejects 'null' during form validation ("Enter a whole number"), so the sentinel never reaches the filter at all. That is a property of multivalue_field_factory() and is unaffected by this behavior. """ no_key = Site.objects.get(slug='site-4') explicit_null = Site.objects.create(name='Site 5', slug='site-5', custom_field_data={ 'cf4': None, 'cf7': None, 'cf9': None, }) has_value = set( Site.objects.filter(slug__in=('site-1', 'site-2', 'site-3')).values_list('pk', flat=True) ) for filter_name in ('cf_cf4', 'cf_cf7', 'cf_cf9'): with self.subTest(filter_name): pks = set( self.filterset({filter_name: ['null']}, self.queryset).qs.values_list('pk', flat=True) ) self.assertEqual(pks, {no_key.pk, explicit_null.pk}) pks = set( self.filterset({f'{filter_name}__n': ['null']}, self.queryset) .qs.values_list('pk', flat=True) ) self.assertEqual(pks, has_value) def test_filter_null_sentinel_combined_with_a_value(self): """ The sentinel may be passed alongside real values, in which case it widens the match rather than replacing it. Under negation the valueless objects are then excluded, as they are among the values being negated. """ no_key = Site.objects.get(slug='site-4') site_1 = Site.objects.get(slug='site-1') pks = set( self.filterset({'cf_cf4': ['foo', 'null']}, self.queryset).qs.values_list('pk', flat=True) ) self.assertIn(site_1.pk, pks, "an object holding the value must match") self.assertIn(no_key.pk, pks, "an object holding no value must match") pks = set( self.filterset({'cf_cf4__n': ['foo', 'null']}, self.queryset).qs.values_list('pk', flat=True) ) self.assertNotIn(site_1.pk, pks, "an object holding the value must be excluded") self.assertNotIn(no_key.pk, pks, "an object holding no value must be excluded") def test_filter_select(self): self.assertEqual(self.filterset({'cf_cf9': ['A', 'B']}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf9__empty': True}, self.queryset).qs.count(), 1) def test_filter_multiselect(self): self.assertEqual(self.filterset({'cf_cf10': ['A']}, self.queryset).qs.count(), 1) self.assertEqual(self.filterset({'cf_cf10': ['A', 'C']}, self.queryset).qs.count(), 2) # Matches both the object holding a literal null and the one carrying no key, as `empty` does self.assertEqual(self.filterset({'cf_cf10': ['null']}, self.queryset).qs.count(), 2) self.assertEqual(self.filterset({'cf_cf10__empty': True}, self.queryset).qs.count(), 2) def test_filter_object(self): manufacturer_ids = Manufacturer.objects.values_list('id', flat=True) self.assertEqual( self.filterset({'cf_cf11': [manufacturer_ids[0], manufacturer_ids[1]]}, self.queryset).qs.count(), 2 ) self.assertEqual(self.filterset({'cf_cf11__empty': True}, self.queryset).qs.count(), 1) def test_filter_multiobject(self): manufacturer_ids = Manufacturer.objects.values_list('id', flat=True) self.assertEqual( self.filterset({'cf_cf12': [manufacturer_ids[0], manufacturer_ids[1]]}, self.queryset).qs.count(), 2 ) self.assertEqual( self.filterset({'cf_cf12': [manufacturer_ids[3]]}, self.queryset).qs.count(), 3 ) self.assertEqual(self.filterset({'cf_cf12__empty': True}, self.queryset).qs.count(), 1)