diff --git a/netbox/netbox/tests/test_import.py b/netbox/netbox/tests/test_import.py index 73f775bd7..d5d8861f9 100644 --- a/netbox/netbox/tests/test_import.py +++ b/netbox/netbox/tests/test_import.py @@ -1,3 +1,8 @@ +import csv +import io +import json +import yaml + from django.contrib.contenttypes.models import ContentType from django.test import override_settings @@ -17,8 +22,16 @@ class CSVImportTestCase(ModelViewTestCase): def _get_csv_data(self, csv_data): return '\n'.join(csv_data) + def _get_yaml_data(self, csv_data): + data = [*csv.DictReader(io.StringIO(self._get_csv_data(csv_data)))] + return yaml.dump(data) + + def _get_json_data(self, csv_data): + data = [*csv.DictReader(io.StringIO(self._get_csv_data(csv_data)))] + return json.dumps(data) + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) - def test_valid_tags(self): + def test_valid_tags_csv(self): csv_data = ( 'name,slug,tags', 'Region 1,region-1,"alpha,bravo"', @@ -61,7 +74,93 @@ class CSVImportTestCase(ModelViewTestCase): self.assertEqual(regions[3].tags.count(), 0) @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) - def test_invalid_tags(self): + def test_valid_tags_yaml(self): + csv_data = ( + 'name,slug,tags', + 'Region 1,region-1,"alpha,bravo"', + 'Region 2,region-2,"charlie,delta"', + 'Region 3,region-3,echo', + 'Region 4,region-4,', + ) + + data = { + 'format': ImportFormatChoices.YAML, + 'data': self._get_yaml_data(csv_data), + } + + # Assign model-level permission + obj_perm = ObjectPermission(name='Test permission', actions=['add']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + # Try GET with model-level permission + self.assertHttpStatus(self.client.get(self._get_url('import')), 200) + + # Test POST with permission + self.assertHttpStatus(self.client.post(self._get_url('import'), data), 200) + regions = Region.objects.all() + self.assertEqual(regions.count(), 4) + region = Region.objects.get(slug="region-4") + self.assertEqual( + list(regions[0].tags.values_list('name', flat=True)), + ['Alpha', 'Bravo'] + ) + self.assertEqual( + list(regions[1].tags.values_list('name', flat=True)), + ['Charlie', 'Delta'] + ) + self.assertEqual( + list(regions[2].tags.values_list('name', flat=True)), + ['Echo'] + ) + self.assertEqual(regions[3].tags.count(), 0) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_valid_tags_json(self): + csv_data = ( + 'name,slug,tags', + 'Region 1,region-1,"alpha,bravo"', + 'Region 2,region-2,"charlie,delta"', + 'Region 3,region-3,echo', + 'Region 4,region-4,', + ) + + data = { + 'format': ImportFormatChoices.JSON, + 'data': self._get_json_data(csv_data), + } + + # Assign model-level permission + obj_perm = ObjectPermission(name='Test permission', actions=['add']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + # Try GET with model-level permission + self.assertHttpStatus(self.client.get(self._get_url('import')), 200) + + # Test POST with permission + self.assertHttpStatus(self.client.post(self._get_url('import'), data), 200) + regions = Region.objects.all() + self.assertEqual(regions.count(), 4) + region = Region.objects.get(slug="region-4") + self.assertEqual( + list(regions[0].tags.values_list('name', flat=True)), + ['Alpha', 'Bravo'] + ) + self.assertEqual( + list(regions[1].tags.values_list('name', flat=True)), + ['Charlie', 'Delta'] + ) + self.assertEqual( + list(regions[2].tags.values_list('name', flat=True)), + ['Echo'] + ) + self.assertEqual(regions[3].tags.count(), 0) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_invalid_tags_csv(self): csv_data = ( 'name,slug,tags', 'Region 1,region-1,"Alpha,Bravo"', # Valid @@ -85,3 +184,124 @@ class CSVImportTestCase(ModelViewTestCase): # Test POST with permission self.assertHttpStatus(self.client.post(self._get_url('import'), data), 200) self.assertEqual(Region.objects.count(), 0) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_invalid_tags_yaml(self): + csv_data = ( + 'name,slug,tags', + 'Region 1,region-1,"Alpha,Bravo"', # Valid + 'Region 2,region-2,"Alpha,Tango"', # Invalid + ) + + data = { + 'format': ImportFormatChoices.YAML, + 'data': self._get_yaml_data(csv_data), + } + + # Assign model-level permission + obj_perm = ObjectPermission(name='Test permission', actions=['add']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + # Try GET with model-level permission + self.assertHttpStatus(self.client.get(self._get_url('import')), 200) + + # Test POST with permission + self.assertHttpStatus(self.client.post(self._get_url('import'), data), 200) + self.assertEqual(Region.objects.count(), 0) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_invalid_tags_json(self): + csv_data = ( + 'name,slug,tags', + 'Region 1,region-1,"Alpha,Bravo"', # Valid + 'Region 2,region-2,"Alpha,Tango"', # Invalid + ) + + data = { + 'format': ImportFormatChoices.JSON, + 'data': self._get_json_data(csv_data), + } + + # Assign model-level permission + obj_perm = ObjectPermission(name='Test permission', actions=['add']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + # Try GET with model-level permission + self.assertHttpStatus(self.client.get(self._get_url('import')), 200) + + # Test POST with permission + self.assertHttpStatus(self.client.post(self._get_url('import'), data), 200) + self.assertEqual(Region.objects.count(), 0) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_invalid_header_csv(self): + csv_data = ( + 'name,slug,tags,xxx', + 'Region 1,region-1,"alpha,bravo",yyy', + ) + + data = { + 'format': ImportFormatChoices.CSV, + 'data': self._get_csv_data(csv_data), + } + + # Assign model-level permission + obj_perm = ObjectPermission(name='Test permission', actions=['add']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + # Test POST with permission + ret = self.client.post(self._get_url('import'), data) + self.assertHttpStatus(self.client.post(self._get_url('import'), data), 200) + self.assertEqual(Region.objects.count(), 0) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_invalid_header_yaml(self): + csv_data = ( + 'name,slug,tags,xxx', + 'Region 1,region-1,"alpha,bravo",yyy', + ) + + data = { + 'format': ImportFormatChoices.YAML, + 'data': self._get_yaml_data(csv_data), + } + + # Assign model-level permission + obj_perm = ObjectPermission(name='Test permission', actions=['add']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + # Test POST with permission + ret = self.client.post(self._get_url('import'), data) + self.assertHttpStatus(self.client.post(self._get_url('import'), data), 200) + self.assertEqual(Region.objects.count(), 0) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_invalid_header_json(self): + csv_data = ( + 'name,slug,tags,xxx', + 'Region 1,region-1,"alpha,bravo",yyy', + ) + + data = { + 'format': ImportFormatChoices.JSON, + 'data': self._get_json_data(csv_data), + } + + # Assign model-level permission + obj_perm = ObjectPermission(name='Test permission', actions=['add']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + # Test POST with permission + ret = self.client.post(self._get_url('import'), data) + self.assertHttpStatus(self.client.post(self._get_url('import'), data), 200) + self.assertEqual(Region.objects.count(), 0) diff --git a/netbox/netbox/views/generic/bulk_views.py b/netbox/netbox/views/generic/bulk_views.py index 0e7bfdcd7..1334d2c19 100644 --- a/netbox/netbox/views/generic/bulk_views.py +++ b/netbox/netbox/views/generic/bulk_views.py @@ -23,8 +23,7 @@ from utilities.exceptions import AbortRequest, AbortTransaction, PermissionsViol from utilities.forms import BulkRenameForm, ConfirmationForm, restrict_form_fields from utilities.forms.bulk_import import BulkImportForm from utilities.forms.utils import headers_to_dict, validate_import_headers -from utilities.htmx import is_embedded -from utilities.htmx import is_htmx +from utilities.htmx import is_embedded, is_htmx from utilities.permissions import get_permission_for_model from utilities.utils import get_viewname from utilities.views import GetReturnURLMixin