nautobot.apps.testing

Utilities for apps to implement test automation.

nautobot.apps.testing.APITestCase

Bases: views.ModelTestCase

Base test case for API requests.

api_version: Specific API version to test. Leave unset to test the default behavior. Override with set_api_version()

Source code in nautobot/core/testing/api.py

@tag("api")
class APITestCase(views.ModelTestCase):
    """
    Base test case for API requests.

    api_version: Specific API version to test. Leave unset to test the default behavior. Override with set_api_version()
    """

    api_version = None

    def setUp(self):
        """
        Create a token for API calls.
        """
        super().setUp()
        self.client.logout()
        self.token = users_models.Token.objects.create(user=self.user)
        self.header = {"HTTP_AUTHORIZATION": f"Token {self.token.key}"}
        if self.api_version:
            self.set_api_version(self.api_version)

    def set_api_version(self, api_version):
        """Set or unset a specific API version for requests in this test case."""
        if api_version is None:
            self.header["HTTP_ACCEPT"] = "application/json"
        else:
            self.header["HTTP_ACCEPT"] = f"application/json; version={api_version}"

    def _get_detail_url(self, instance):
        viewname = lookup.get_route_for_model(instance, "detail", api=True)
        return reverse(viewname, kwargs={"pk": instance.pk})

    def _get_list_url(self):
        viewname = lookup.get_route_for_model(self.model, "list", api=True)
        return reverse(viewname)

setUp()

Create a token for API calls.

Source code in nautobot/core/testing/api.py

def setUp(self):
    """
    Create a token for API calls.
    """
    super().setUp()
    self.client.logout()
    self.token = users_models.Token.objects.create(user=self.user)
    self.header = {"HTTP_AUTHORIZATION": f"Token {self.token.key}"}
    if self.api_version:
        self.set_api_version(self.api_version)

set_api_version(api_version)

Set or unset a specific API version for requests in this test case.

Source code in nautobot/core/testing/api.py

def set_api_version(self, api_version):
    """Set or unset a specific API version for requests in this test case."""
    if api_version is None:
        self.header["HTTP_ACCEPT"] = "application/json"
    else:
        self.header["HTTP_ACCEPT"] = f"application/json; version={api_version}"

nautobot.apps.testing.APIViewTestCases

Source code in nautobot/core/testing/api.py

@tag("unit")
class APIViewTestCases:
    class GetObjectViewTestCase(APITestCase):
        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_get_object_anonymous(self):
            """
            GET a single object as an unauthenticated user.
            """
            url = self._get_detail_url(self._get_queryset().first())
            if (
                self.model._meta.app_label,
                self.model._meta.model_name,
            ) in settings.EXEMPT_EXCLUDE_MODELS:
                # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users
                with testing.disable_warnings("django.request"):
                    self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)
            else:
                response = self.client.get(url, **self.header)
                self.assertHttpStatus(response, status.HTTP_200_OK)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_get_object_without_permission(self):
            """
            GET a single object as an authenticated user without the required permission.
            """
            url = self._get_detail_url(self._get_queryset().first())

            # Try GET without permission
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_get_object(self):
            """
            GET a single object as an authenticated user with permission to view the object.
            """
            self.assertGreaterEqual(
                self._get_queryset().count(),
                2,
                f"Test requires the creation of at least two {self.model} instances",
            )
            instance1, instance2 = self._get_queryset()[:2]

            # Add object-level permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"pk": instance1.pk},
                actions=["view"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try GET to non-permitted object
            url = self._get_detail_url(instance2)
            self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_404_NOT_FOUND)

            # Try GET to permitted object
            url = self._get_detail_url(instance1)
            response = self.client.get(url, **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertIsInstance(response.data, dict)
            # Fields that should be present in *ALL* model serializers:
            self.assertIn("id", response.data)
            self.assertEqual(str(response.data["id"]), str(instance1.pk))  # coerce to str to handle both int and uuid
            self.assertIn("url", response.data)
            self.assertIn("display", response.data)
            self.assertIsInstance(response.data["display"], str)
            # Fields that should be present in appropriate model serializers:
            if issubclass(self.model, extras_models.ChangeLoggedModel):
                self.assertIn("created", response.data)
                self.assertIn("last_updated", response.data)
            # Fields that should be absent by default (opt-in fields):
            self.assertNotIn("computed_fields", response.data)
            self.assertNotIn("relationships", response.data)

            # If opt-in fields are supported on this model, make sure they can be opted into

            custom_fields_registry = registry.registry["model_features"]["custom_fields"]
            # computed fields and custom fields use the same registry
            cf_supported = self.model._meta.model_name in custom_fields_registry.get(self.model._meta.app_label, {})
            if cf_supported:  # custom_fields is not an opt-in field, it should always be present if supported
                self.assertIn("custom_fields", response.data)
                self.assertIsInstance(response.data["custom_fields"], dict)

            relationships_registry = registry.registry["model_features"]["relationships"]
            rel_supported = self.model._meta.model_name in relationships_registry.get(self.model._meta.app_label, {})
            if cf_supported or rel_supported:
                query_params = []
                if cf_supported:
                    query_params.append("include=computed_fields")
                if rel_supported:
                    query_params.append("include=relationships")
                query_string = "&".join(query_params)
                url = f"{url}?{query_string}"

                response = self.client.get(url, **self.header)
                self.assertHttpStatus(response, status.HTTP_200_OK)
                self.assertIsInstance(response.data, dict)
                if cf_supported:
                    self.assertIn("computed_fields", response.data)
                    self.assertIsInstance(response.data["computed_fields"], dict)
                else:
                    self.assertNotIn("computed_fields", response.data)
                if rel_supported:
                    self.assertIn("relationships", response.data)
                    self.assertIsInstance(response.data["relationships"], dict)
                else:
                    self.assertNotIn("relationships", response.data)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_options_object(self):
            """
            Make an OPTIONS request for a single object.
            """
            url = self._get_detail_url(self._get_queryset().first())
            response = self.client.options(url, **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)

            with self.subTest("Assert Detail View Config is generated well"):
                # Namings Help
                # 1. detail_view_config: This is the detail view config set in the serializer.Meta.detail_view_config
                # 2. detail_view_schema: This is the retrieve schema generated from an OPTIONS request.
                # 3. advanced_view_config: This is the advanced view config set in the serializer.Meta.advanced_view_config
                # 4. advanced_view_schema: This is the advanced tab schema generated from an OPTIONS request.
                serializer = get_serializer_for_model(self._get_queryset().model)
                advanced_view_schema = response.data["view_options"]["advanced"]

                if getattr(serializer.Meta, "advanced_view_config", None):
                    # Get all custom advanced tab fields
                    advanced_view_config = serializer.Meta.advanced_view_config
                    self.assertGreaterEqual(len(advanced_view_schema), 1)
                    advanced_tab_fields = []
                    advanced_view_layout = advanced_view_config["layout"]
                    for section in advanced_view_layout:
                        for value in section.values():
                            advanced_tab_fields += value["fields"]
                else:
                    # Get default advanced tab fields
                    self.assertEqual(len(advanced_view_schema), 1)
                    self.assertIn("Object Details", advanced_view_schema[0])
                    advanced_tab_fields = advanced_view_schema[0].get("Object Details")["fields"]

                if detail_view_config := getattr(serializer.Meta, "detail_view_config", None):
                    detail_view_schema = response.data["view_options"]["retrieve"]
                    self.assertHttpStatus(response, status.HTTP_200_OK)

                    # According to convention, fields in the advanced tab fields should not exist in
                    # the `detail_view_schema`. Assert this is True.
                    with self.subTest("Assert advanced tab fields should not exist in the detail_view_schema."):
                        if detail_view_config.get("include_others"):
                            # Handle `Other Fields` specially as `Other Field` is dynamically added by nautobot and not part of the serializer view config
                            other_fields = detail_view_schema[0]["Other Fields"]["fields"]
                            for field in advanced_tab_fields:
                                self.assertNotIn(field, other_fields)

                        for col_idx, col in enumerate(detail_view_schema):
                            for group_idx, (group_title, group) in enumerate(col.items()):
                                if group_title == "Other Fields":
                                    continue
                                group_fields = group["fields"]
                                # Config on the serializer
                                fields = detail_view_config["layout"][col_idx][group_title]["fields"]
                                # According to convention, advanced_tab_fields should only exist at the end of the first col group and should be
                                # deleted from any other places it may appear in the layout. Assert this is True.
                                if group_idx == 0 == col_idx:
                                    self.assertFalse(
                                        any(
                                            field in advanced_tab_fields
                                            for field in group_fields[: -len(advanced_tab_fields)]
                                        )
                                    )
                                else:
                                    for field in group_fields:
                                        self.assertNotIn(field, advanced_tab_fields)
                                # Assert response from options correspond to the view config set on the serializer
                                for field in fields:
                                    if field not in advanced_tab_fields:
                                        self.assertIn(field, group_fields)

    class ListObjectsViewTestCase(APITestCase):
        choices_fields = None
        filterset = None

        def get_filterset(self):
            return self.filterset or lookup.get_filterset_for_model(self.model)

        def get_depth_fields(self):
            """Get a list of model fields that could be tested with the ?depth query parameter"""
            depth_fields = []
            for field in self.model._meta.fields:
                if not field.name.startswith("_"):
                    if isinstance(field, (ForeignKey, GenericForeignKey, ManyToManyField, core_fields.TagsField)) and (
                        # we represent content-types as "app_label.modelname" rather than as FKs
                        field.related_model != ContentType
                        # user is a model field on Token but not a field on TokenSerializer
                        and not (field.name == "user" and self.model == users_models.Token)
                    ):
                        depth_fields.append(field.name)
            return depth_fields

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_list_objects_anonymous(self):
            """
            GET a list of objects as an unauthenticated user.
            """
            url = self._get_list_url()
            if (
                self.model._meta.app_label,
                self.model._meta.model_name,
            ) in settings.EXEMPT_EXCLUDE_MODELS:
                # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users
                with testing.disable_warnings("django.request"):
                    self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)
            else:
                # TODO(Glenn): if we're passing **self.header, we are *by definition* **NOT** anonymous!!
                response = self.client.get(url, **self.header)
                self.assertHttpStatus(response, status.HTTP_200_OK)
                self.assertIsInstance(response.data, dict)
                self.assertIn("results", response.data)
                self.assertEqual(len(response.data["results"]), self._get_queryset().count())

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects_depth_0(self):
            """
            GET a list of objects using the "?depth=0" parameter.
            """
            depth_fields = self.get_depth_fields()
            self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
            url = f"{self._get_list_url()}?depth=0"
            response = self.client.get(url, **self.header)

            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertIsInstance(response.data, dict)
            self.assertIn("results", response.data)
            self.assertEqual(len(response.data["results"]), self._get_queryset().count())

            for response_data in response.data["results"]:
                for field in depth_fields:
                    self.assertIn(field, response_data)
                    if isinstance(response_data[field], list):
                        for entry in response_data[field]:
                            self.assertIsInstance(entry, dict)
                            self.assertTrue(is_uuid(entry["id"]))
                    else:
                        if response_data[field] is not None:
                            self.assertIsInstance(response_data[field], dict)
                            url = response_data[field]["url"]
                            pk = response_data[field]["id"]
                            object_type = response_data[field]["object_type"]
                            # The response should be a brief API object, containing an ID, object_type, and URL ending in the UUID of the relevant object
                            # http://nautobot.example.com/api/circuits/providers/<uuid>/
                            #                                                    ^^^^^^
                            self.assertTrue(is_uuid(url.split("/")[-2]))
                            self.assertTrue(is_uuid(pk))

                            with self.subTest(f"Assert object_type {object_type} is valid"):
                                app_label, model_name = object_type.split(".")
                                ContentType.objects.get(app_label=app_label, model=model_name)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects_depth_1(self):
            """
            GET a list of objects using the "?depth=1" parameter.
            """
            depth_fields = self.get_depth_fields()
            self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
            url = f"{self._get_list_url()}?depth=1"
            response = self.client.get(url, **self.header)

            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertIsInstance(response.data, dict)
            self.assertIn("results", response.data)
            self.assertEqual(len(response.data["results"]), self._get_queryset().count())

            for response_data in response.data["results"]:
                for field in depth_fields:
                    self.assertIn(field, response_data)
                    if isinstance(response_data[field], list):
                        for entry in response_data[field]:
                            self.assertIsInstance(entry, dict)
                            self.assertTrue(is_uuid(entry["id"]))
                    else:
                        if response_data[field] is not None:
                            self.assertIsInstance(response_data[field], dict)
                            self.assertTrue(is_uuid(response_data[field]["id"]))

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects_without_permission(self):
            """
            GET a list of objects as an authenticated user without the required permission.
            """
            url = self._get_list_url()

            # Try GET without permission
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects(self):
            """
            GET a list of objects as an authenticated user with permission to view the objects.
            """
            self.assertGreaterEqual(
                self._get_queryset().count(),
                3,
                f"Test requires the creation of at least three {self.model} instances",
            )
            instance1, instance2 = self._get_queryset()[:2]

            # Add object-level permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"pk__in": [instance1.pk, instance2.pk]},
                actions=["view"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try GET to permitted objects
            response = self.client.get(self._get_list_url(), **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertIsInstance(response.data, dict)
            self.assertIn("results", response.data)
            self.assertEqual(len(response.data["results"]), 2)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects_filtered(self):
            """
            GET a list of objects filtered by ID.
            """
            self.assertGreaterEqual(
                self._get_queryset().count(),
                3,
                f"Test requires the creation of at least three {self.model} instances",
            )
            self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
            instance1, instance2 = self._get_queryset()[:2]
            response = self.client.get(f"{self._get_list_url()}?id={instance1.pk}&id={instance2.pk}", **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertIsInstance(response.data, dict)
            self.assertIn("results", response.data)
            self.assertEqual(len(response.data["results"]), 2)
            for entry in response.data["results"]:
                self.assertIn(str(entry["id"]), [str(instance1.pk), str(instance2.pk)])

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects_ascending_ordered(self):
            # Simple sorting check for models with a "name" field
            # TreeModels don't support sorting at this time (order_by is not supported by TreeQuerySet)
            #   They will pass api == queryset tests below but will fail the user expected sort test
            if hasattr(self.model, "name") and not issubclass(self.model, TreeModel):
                self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
                response = self.client.get(f"{self._get_list_url()}?sort=name&limit=3", **self.header)
                self.assertHttpStatus(response, status.HTTP_200_OK)
                result_list = list(map(lambda p: p["name"], response.data["results"]))
                self.assertEqual(
                    result_list,
                    list(self._get_queryset().order_by("name").values_list("name", flat=True)[:3]),
                    "API sort not identical to QuerySet.order_by",
                )

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects_descending_ordered(self):
            # Simple sorting check for models with a "name" field
            # TreeModels don't support sorting at this time (order_by is not supported by TreeQuerySet)
            #   They will pass api == queryset tests below but will fail the user expected sort test
            if hasattr(self.model, "name") and not issubclass(self.model, TreeModel):
                self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
                response = self.client.get(f"{self._get_list_url()}?sort=-name&limit=3", **self.header)
                self.assertHttpStatus(response, status.HTTP_200_OK)
                result_list = list(map(lambda p: p["name"], response.data["results"]))
                self.assertEqual(
                    result_list,
                    list(self._get_queryset().order_by("-name").values_list("name", flat=True)[:3]),
                    "API sort not identical to QuerySet.order_by",
                )

                response_ascending = self.client.get(f"{self._get_list_url()}?sort=name&limit=3", **self.header)
                self.assertHttpStatus(response, status.HTTP_200_OK)
                result_list_ascending = list(map(lambda p: p["name"], response_ascending.data["results"]))

                self.assertNotEqual(
                    result_list,
                    result_list_ascending,
                    "Same results obtained when sorting by name and by -name (QuerySet not ordering)",
                )

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[], STRICT_FILTERING=True)
        def test_list_objects_unknown_filter_strict_filtering(self):
            """
            GET a list of objects with an unknown filter parameter and strict filtering, expect a 400 response.
            """
            self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
            with testing.disable_warnings("django.request"):
                response = self.client.get(f"{self._get_list_url()}?ice_cream_flavor=rocky-road", **self.header)
            self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
            self.assertIsInstance(response.data, dict)
            self.assertIn("ice_cream_flavor", response.data)
            self.assertIsInstance(response.data["ice_cream_flavor"], list)
            self.assertEqual("Unknown filter field", str(response.data["ice_cream_flavor"][0]))

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[], STRICT_FILTERING=False)
        def test_list_objects_unknown_filter_no_strict_filtering(self):
            """
            GET a list of objects with an unknown filter parameter and no strict filtering, expect it to be ignored.
            """
            self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
            with self.assertLogs("nautobot.core.filters") as cm:
                response = self.client.get(f"{self._get_list_url()}?ice_cream_flavor=rocky-road", **self.header)
            self.assertEqual(
                cm.output,
                [
                    f"WARNING:nautobot.core.filters:{self.get_filterset().__name__}: "
                    'Unknown filter field "ice_cream_flavor"',
                ],
            )
            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertIsInstance(response.data, dict)
            self.assertIn("results", response.data)
            self.assertEqual(len(response.data["results"]), self._get_queryset().count())

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_options_objects(self):
            """
            Make an OPTIONS request for a list endpoint.
            """
            response = self.client.options(self._get_list_url(), **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects_csv(self):
            """
            GET a list of objects in CSV format as an authenticated user with permission to view some objects.
            """
            self.assertGreaterEqual(
                self._get_queryset().count(),
                3,
                f"Test requires the creation of at least three {self.model} instances",
            )
            instance1, instance2, instance3 = self._get_queryset()[:3]

            # Add object-level permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"pk__in": [instance1.pk, instance2.pk]},
                actions=["view"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try filtered GET to objects specifying CSV format as a query parameter
            response_1 = self.client.get(
                f"{self._get_list_url()}?format=csv&id={instance1.pk}&id={instance3.pk}", **self.header
            )
            self.assertHttpStatus(response_1, status.HTTP_200_OK)
            self.assertEqual(response_1.get("Content-Type"), "text/csv; charset=UTF-8")
            self.assertEqual(
                response_1.get("Content-Disposition"),
                f'attachment; filename="nautobot_{self.model.__name__.lower()}_data.csv"',
            )

            # Try same request specifying CSV format via the ACCEPT header
            response_2 = self.client.get(
                f"{self._get_list_url()}?id={instance1.pk}&id={instance3.pk}", **self.header, HTTP_ACCEPT="text/csv"
            )
            self.assertHttpStatus(response_2, status.HTTP_200_OK)
            self.assertEqual(response_2.get("Content-Type"), "text/csv; charset=UTF-8")
            self.assertEqual(
                response_2.get("Content-Disposition"),
                f'attachment; filename="nautobot_{self.model.__name__.lower()}_data.csv"',
            )

            self.maxDiff = None
            # This check is more useful than it might seem. Any related object that wasn't CSV-converted correctly
            # will likely be rendered incorrectly as an API URL, and that API URL *will* differ between the
            # two responses based on the inclusion or omission of the "?format=csv" parameter. If
            # you run into this, make sure all serializers have `Meta.fields = "__all__"` set.
            self.assertEqual(
                response_1.content.decode(response_1.charset), response_2.content.decode(response_2.charset)
            )

            # Load the csv data back into a list of object dicts
            reader = csv.DictReader(StringIO(response_1.content.decode(response_1.charset)))
            rows = list(reader)
            # Should only have one entry (instance1) since we filtered out instance2 and permissions block instance3
            self.assertEqual(1, len(rows))
            self.assertEqual(rows[0]["id"], str(instance1.pk))
            self.assertEqual(rows[0]["display"], getattr(instance1, "display", str(instance1)))
            if hasattr(self.model, "_custom_field_data"):
                custom_fields = extras_models.CustomField.objects.get_for_model(self.model)
                for cf in custom_fields:
                    self.assertIn(f"cf_{cf.key}", rows[0])
                    self.assertEqual(rows[0][f"cf_{cf.key}"], instance1._custom_field_data.get(cf.key) or "")
            # TODO what other generic tests should we run on the data?

    class CreateObjectViewTestCase(APITestCase):
        create_data = []
        validation_excluded_fields = []
        slug_source: Optional[Union[str, Sequence[str]]] = None
        slugify_function = staticmethod(slugify)

        def test_create_object_without_permission(self):
            """
            POST a single object without permission.
            """
            url = self._get_list_url()

            # Try POST without permission
            with testing.disable_warnings("django.request"):
                response = self.client.post(url, self.create_data[0], format="json", **self.header)
                self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

        def check_expected_slug(self, obj):
            slug_source = self.slug_source if isinstance(self.slug_source, (list, tuple)) else [self.slug_source]
            expected_slug = ""
            for source_item in slug_source:
                # e.g. self.slug_source = ["parent__name", "name"]
                source_keys = source_item.split("__")
                try:
                    val = getattr(obj, source_keys[0])
                    for key in source_keys[1:]:
                        val = getattr(val, key)
                except AttributeError:
                    val = ""
                if val:
                    if expected_slug != "":
                        expected_slug += "-"
                    expected_slug += self.slugify_function(val)

            self.assertNotEqual(expected_slug, "")
            if hasattr(obj, "slug"):
                self.assertEqual(obj.slug, expected_slug)
            else:
                self.assertEqual(obj.key, expected_slug)

        def test_create_object(self):
            """
            POST a single object with permission.
            """
            # Add object-level permission
            obj_perm = users_models.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))

            initial_count = self._get_queryset().count()
            for i, create_data in enumerate(self.create_data):
                if i == len(self.create_data) - 1:
                    # Test to see if depth parameter is ignored in POST request.
                    response = self.client.post(
                        self._get_list_url() + "?depth=3", create_data, format="json", **self.header
                    )
                else:
                    response = self.client.post(self._get_list_url(), create_data, format="json", **self.header)
                self.assertHttpStatus(response, status.HTTP_201_CREATED)
                self.assertEqual(self._get_queryset().count(), initial_count + i + 1)
                instance = self._get_queryset().get(pk=response.data["id"])
                self.assertInstanceEqual(
                    instance,
                    create_data,
                    exclude=self.validation_excluded_fields,
                    api=True,
                )

                # Check if Slug field is automatically created
                if self.slug_source is not None and "slug" not in create_data:
                    self.check_expected_slug(self._get_queryset().get(pk=response.data["id"]))

                # Verify ObjectChange creation
                if hasattr(self.model, "to_objectchange"):
                    objectchanges = lookup.get_changes_for_model(instance)
                    self.assertEqual(len(objectchanges), 1)
                    self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_CREATE)

        def test_recreate_object_csv(self):
            """CSV export an object, delete it, and recreate it via CSV import."""
            if hasattr(self, "get_deletable_object"):
                # provided by DeleteObjectViewTestCase mixin
                instance = self.get_deletable_object()
            else:
                # try to do it ourselves
                instance = testing.get_deletable_objects(self.model, self._get_queryset()).first()
            if instance is None:
                self.fail("Couldn't find a single deletable object!")

            # Add object-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["add", "view"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            response = self.client.get(self._get_detail_url(instance) + "?format=csv", **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertEqual(
                response.get("Content-Disposition"),
                f'attachment; filename="nautobot_{self.model.__name__.lower()}_data.csv"',
            )
            csv_data = response.content.decode(response.charset)

            serializer_class = get_serializer_for_model(self.model)
            old_serializer = serializer_class(instance, context={"request": None})
            old_data = old_serializer.data
            instance.delete()

            response = self.client.post(self._get_list_url(), csv_data, content_type="text/csv", **self.header)
            self.assertHttpStatus(response, status.HTTP_201_CREATED, csv_data)
            # Note that create via CSV is always treated as a bulk-create, and so the response is always a list of dicts
            new_instance = self._get_queryset().get(pk=response.data[0]["id"])
            self.assertNotEqual(new_instance.pk, instance.pk)

            new_serializer = serializer_class(new_instance, context={"request": None})
            new_data = new_serializer.data
            for field_name, field in new_serializer.fields.items():
                if field.read_only or field.write_only:
                    continue
                if field_name in ["created", "last_updated"]:
                    self.assertNotEqual(
                        old_data[field_name],
                        new_data[field_name],
                        f"{field_name} should have been updated on delete/recreate but it didn't change!",
                    )
                else:
                    self.assertEqual(
                        old_data[field_name],
                        new_data[field_name],
                        f"{field_name} should have been unchanged on delete/recreate but it differs!",
                    )

        def test_bulk_create_objects(self):
            """
            POST a set of objects in a single request.
            """
            # Add object-level permission
            obj_perm = users_models.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))

            initial_count = self._get_queryset().count()
            response = self.client.post(self._get_list_url(), self.create_data, format="json", **self.header)
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            self.assertEqual(len(response.data), len(self.create_data))
            self.assertEqual(self._get_queryset().count(), initial_count + len(self.create_data))
            for i, obj in enumerate(response.data):
                for field in self.create_data[i]:
                    if field not in self.validation_excluded_fields:
                        self.assertIn(
                            field,
                            obj,
                            f"Bulk create field '{field}' missing from object {i} in response",
                        )
            for i, obj in enumerate(response.data):
                self.assertInstanceEqual(
                    self._get_queryset().get(pk=obj["id"]),
                    self.create_data[i],
                    exclude=self.validation_excluded_fields,
                    api=True,
                )
                if self.slug_source is not None and "slug" not in self.create_data[i]:
                    self.check_expected_slug(self._get_queryset().get(pk=obj["id"]))

    class UpdateObjectViewTestCase(APITestCase):
        update_data = {}
        bulk_update_data: Optional[dict] = None
        validation_excluded_fields = []
        choices_fields = None

        def test_update_object_without_permission(self):
            """
            PATCH a single object without permission.
            """
            url = self._get_detail_url(self._get_queryset().first())
            update_data = self.update_data or getattr(self, "create_data")[0]

            # Try PATCH without permission
            with testing.disable_warnings("django.request"):
                response = self.client.patch(url, update_data, format="json", **self.header)
                self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

        def test_update_object(self):
            """
            PATCH a single object identified by its ID.
            """

            def strip_serialized_object(this_object):
                """
                Only here to work around acceptable differences in PATCH response vs GET response which are known bugs.
                """
                # Work around for https://github.com/nautobot/nautobot/issues/3321
                this_object.pop("last_updated", None)
                # PATCH response always includes "opt-in" fields, but GET response does not.
                this_object.pop("computed_fields", None)
                this_object.pop("config_context", None)
                this_object.pop("relationships", None)

                for value in this_object.values():
                    if isinstance(value, dict):
                        strip_serialized_object(value)
                    elif isinstance(value, list):
                        for list_dict in value:
                            if isinstance(list_dict, dict):
                                strip_serialized_object(list_dict)

            self.maxDiff = None
            instance = self._get_queryset().first()
            url = self._get_detail_url(instance)
            update_data = self.update_data or getattr(self, "create_data")[0]

            # Add object-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Verify that an empty PATCH results in no change to the object.
            # This is to catch issues like https://github.com/nautobot/nautobot/issues/3533

            # Add object-level permission for GET
            obj_perm.actions = ["view"]
            obj_perm.save()
            # Get initial serialized object representation
            get_response = self.client.get(url, **self.header)
            self.assertHttpStatus(get_response, status.HTTP_200_OK)
            initial_serialized_object = get_response.json()
            strip_serialized_object(initial_serialized_object)

            # Redefine object-level permission for PATCH
            obj_perm.actions = ["change"]
            obj_perm.save()

            # Send empty PATCH request
            response = self.client.patch(url, {}, format="json", **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            serialized_object = response.json()
            strip_serialized_object(serialized_object)
            self.assertEqual(initial_serialized_object, serialized_object)

            # Verify ObjectChange creation -- yes, even though nothing actually changed
            # This may change (hah) at some point -- see https://github.com/nautobot/nautobot/issues/3321
            if hasattr(self.model, "to_objectchange"):
                objectchanges = lookup.get_changes_for_model(instance)
                self.assertEqual(len(objectchanges), 1)
                self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_UPDATE)
                objectchanges.delete()

            # Verify that a PATCH with some data updates that data correctly.
            response = self.client.patch(url, update_data, format="json", **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            # Check for unexpected side effects on fields we DIDN'T intend to update
            for field in initial_serialized_object:
                if field not in update_data:
                    self.assertEqual(initial_serialized_object[field], serialized_object[field])
            instance.refresh_from_db()
            self.assertInstanceEqual(instance, update_data, exclude=self.validation_excluded_fields, api=True)

            # Verify ObjectChange creation
            if hasattr(self.model, "to_objectchange"):
                objectchanges = lookup.get_changes_for_model(instance)
                self.assertEqual(len(objectchanges), 1)
                self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_UPDATE)

        def test_get_put_round_trip(self):
            """GET and then PUT an object and verify that it's accepted and unchanged."""
            self.maxDiff = None
            # Add object-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view", "change"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            instance = self._get_queryset().first()
            url = self._get_detail_url(instance)

            # GET object representation
            opt_in_fields = getattr(get_serializer_for_model(self.model).Meta, "opt_in_fields", None)
            if opt_in_fields:
                url += "?" + "&".join([f"include={field}" for field in opt_in_fields])
            get_response = self.client.get(url, **self.header)
            self.assertHttpStatus(get_response, status.HTTP_200_OK)
            initial_serialized_object = get_response.json()

            # PUT same object representation
            put_response = self.client.put(url, initial_serialized_object, format="json", **self.header)
            self.assertHttpStatus(put_response, status.HTTP_200_OK, initial_serialized_object)
            updated_serialized_object = put_response.json()

            # Work around for https://github.com/nautobot/nautobot/issues/3321
            initial_serialized_object.pop("last_updated", None)
            updated_serialized_object.pop("last_updated", None)
            self.assertEqual(initial_serialized_object, updated_serialized_object)

        def test_bulk_update_objects(self):
            """
            PATCH a set of objects in a single request.
            """
            if self.bulk_update_data is None:
                self.skipTest("Bulk update data not set")

            # Add object-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            id_list = list(self._get_queryset().values_list("id", flat=True)[:3])
            self.assertEqual(len(id_list), 3, "Insufficient number of objects to test bulk update")
            data = [{"id": id, **self.bulk_update_data} for id in id_list]

            response = self.client.patch(self._get_list_url(), data, format="json", **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            for i, obj in enumerate(response.data):
                for field, _value in self.bulk_update_data.items():
                    self.assertIn(
                        field,
                        obj,
                        f"Bulk update field '{field}' missing from object {i} in response",
                    )
                    # TODO(Glenn): shouldn't we also check that obj[field] == value?
            for instance in self._get_queryset().filter(pk__in=id_list):
                self.assertInstanceEqual(
                    instance,
                    self.bulk_update_data,
                    exclude=self.validation_excluded_fields,
                    api=True,
                )

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_options_returns_expected_choices(self):
            """
            Make an OPTIONS request for a list endpoint and validate choices match expected choices for serializer.
            """
            # Set self.choices_fields as empty set to compare classes that shouldn't have any choices on serializer.
            if not self.choices_fields:
                self.choices_fields = set()

            # Save self.user as superuser to be able to view available choices on list views.
            self.user.is_superuser = True
            self.user.save()

            response = self.client.options(self._get_list_url(), **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            data = response.json()

            self.assertIn("actions", data)

            # Grab any field that has choices defined (fields with enums)
            if any(
                [
                    "POST" in data["actions"],
                    "PUT" in data["actions"],
                ]
            ):
                schema = data["schema"]
                props = schema["properties"]
                fields = props.keys()
                field_choices = set()
                for field_name in fields:
                    obj = props[field_name]
                    if "enum" in obj and "enumNames" in obj:
                        enum = obj["enum"]
                        # Zipping to assert that the enum and the mapping have the same number of items.
                        model_field_choices = dict(zip(obj["enumNames"], enum))
                        self.assertEqual(len(enum), len(model_field_choices))
                        field_choices.add(field_name)
            else:
                self.fail(f"Neither PUT nor POST are available actions in: {data['actions']}")

            self.assertEqual(
                set(self.choices_fields),
                field_choices,
                "All field names of choice fields for a given model serializer need to be manually added to "
                "self.choices_fields. If this is already the case, perhaps the serializer is implemented incorrectly?",
            )

    class DeleteObjectViewTestCase(APITestCase):
        def get_deletable_object(self):
            """
            Get an instance that can be deleted.

            For some models this may just be any random object, but when we have FKs with `on_delete=models.PROTECT`
            (as is often the case) we need to find or create an instance that doesn't have such entanglements.
            """
            instance = testing.get_deletable_objects(self.model, self._get_queryset()).first()
            if instance is None:
                self.fail("Couldn't find a single deletable object!")
            return instance

        def get_deletable_object_pks(self):
            """
            Get a list of PKs corresponding to objects that can be safely bulk-deleted.

            For some models this may just be any random objects, but when we have FKs with `on_delete=models.PROTECT`
            (as is often the case) we need to find or create an instance that doesn't have such entanglements.
            """
            instances = testing.get_deletable_objects(self.model, self._get_queryset()).values_list("pk", flat=True)[:3]
            if len(instances) < 3:
                self.fail(f"Couldn't find 3 deletable objects, only found {len(instances)}!")
            return instances

        def test_delete_object_without_permission(self):
            """
            DELETE a single object without permission.
            """
            url = self._get_detail_url(self.get_deletable_object())

            # Try DELETE without permission
            with testing.disable_warnings("django.request"):
                response = self.client.delete(url, **self.header)
                self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

        def test_delete_object(self):
            """
            DELETE a single object identified by its primary key.
            """
            instance = self.get_deletable_object()
            url = self._get_detail_url(instance)

            # Add object-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            response = self.client.delete(url, **self.header)
            self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
            self.assertFalse(self._get_queryset().filter(pk=instance.pk).exists())

            # Verify ObjectChange creation
            if hasattr(self.model, "to_objectchange"):
                objectchanges = lookup.get_changes_for_model(instance)
                self.assertEqual(len(objectchanges), 1)
                self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_DELETE)

        def test_bulk_delete_objects(self):
            """
            DELETE a set of objects in a single request.
            """
            id_list = self.get_deletable_object_pks()
            # Add object-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            data = [{"id": id} for id in id_list]

            initial_count = self._get_queryset().count()
            response = self.client.delete(self._get_list_url(), data, format="json", **self.header)
            self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
            self.assertEqual(self._get_queryset().count(), initial_count - len(id_list))

    class NotesURLViewTestCase(APITestCase):
        """Validate Notes URL on objects that have the Note model Mixin."""

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_notes_url_on_object(self):
            notes = getattr(self.model, "notes", None)
            if notes and isinstance(notes, ForeignKey):
                instance1 = self._get_queryset().first()
                # Add object-level permission
                obj_perm = users_models.ObjectPermission(
                    name="Test permission",
                    constraints={"pk": instance1.pk},
                    actions=["view"],
                )
                obj_perm.save()
                obj_perm.users.add(self.user)
                obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))
                url = self._get_detail_url(instance1)
                response = self.client.get(url, **self.header)
                self.assertHttpStatus(response, status.HTTP_200_OK)
                self.assertIn("notes_url", response.data)
                self.assertIn(f"{url}notes/", str(response.data["notes_url"]))

    class TreeModelAPIViewTestCaseMixin:
        """Test `?depth=2` query parameter for TreeModel"""

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects_depth_2(self):
            """
            GET a list of objects using the "?depth=2" parameter.
            TreeModel Only
            """
            field = "parent"

            self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
            url = f"{self._get_list_url()}?depth=2"
            response = self.client.get(url, **self.header)

            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertIsInstance(response.data, dict)
            self.assertIn("results", response.data)
            self.assertEqual(len(response.data["results"]), self._get_queryset().count())

            response_data = response.data["results"]
            for data in response_data:
                # First Level Parent
                self.assertEqual(field in data, True)
                if data[field] is not None:
                    self.assertIsInstance(data[field], dict)
                    self.assertTrue(is_uuid(data[field]["id"]))
                    # Second Level Parent
                    self.assertIn(field, data[field])
                    if data[field][field] is not None:
                        self.assertIsInstance(data[field][field], dict)
                        self.assertTrue(is_uuid(data[field][field]["id"]))

    class APIViewTestCase(
        GetObjectViewTestCase,
        ListObjectsViewTestCase,
        CreateObjectViewTestCase,
        UpdateObjectViewTestCase,
        DeleteObjectViewTestCase,
        NotesURLViewTestCase,
    ):
        pass

CreateObjectViewTestCase

Bases: APITestCase

Source code in nautobot/core/testing/api.py

class CreateObjectViewTestCase(APITestCase):
    create_data = []
    validation_excluded_fields = []
    slug_source: Optional[Union[str, Sequence[str]]] = None
    slugify_function = staticmethod(slugify)

    def test_create_object_without_permission(self):
        """
        POST a single object without permission.
        """
        url = self._get_list_url()

        # Try POST without permission
        with testing.disable_warnings("django.request"):
            response = self.client.post(url, self.create_data[0], format="json", **self.header)
            self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

    def check_expected_slug(self, obj):
        slug_source = self.slug_source if isinstance(self.slug_source, (list, tuple)) else [self.slug_source]
        expected_slug = ""
        for source_item in slug_source:
            # e.g. self.slug_source = ["parent__name", "name"]
            source_keys = source_item.split("__")
            try:
                val = getattr(obj, source_keys[0])
                for key in source_keys[1:]:
                    val = getattr(val, key)
            except AttributeError:
                val = ""
            if val:
                if expected_slug != "":
                    expected_slug += "-"
                expected_slug += self.slugify_function(val)

        self.assertNotEqual(expected_slug, "")
        if hasattr(obj, "slug"):
            self.assertEqual(obj.slug, expected_slug)
        else:
            self.assertEqual(obj.key, expected_slug)

    def test_create_object(self):
        """
        POST a single object with permission.
        """
        # Add object-level permission
        obj_perm = users_models.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))

        initial_count = self._get_queryset().count()
        for i, create_data in enumerate(self.create_data):
            if i == len(self.create_data) - 1:
                # Test to see if depth parameter is ignored in POST request.
                response = self.client.post(
                    self._get_list_url() + "?depth=3", create_data, format="json", **self.header
                )
            else:
                response = self.client.post(self._get_list_url(), create_data, format="json", **self.header)
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            self.assertEqual(self._get_queryset().count(), initial_count + i + 1)
            instance = self._get_queryset().get(pk=response.data["id"])
            self.assertInstanceEqual(
                instance,
                create_data,
                exclude=self.validation_excluded_fields,
                api=True,
            )

            # Check if Slug field is automatically created
            if self.slug_source is not None and "slug" not in create_data:
                self.check_expected_slug(self._get_queryset().get(pk=response.data["id"]))

            # Verify ObjectChange creation
            if hasattr(self.model, "to_objectchange"):
                objectchanges = lookup.get_changes_for_model(instance)
                self.assertEqual(len(objectchanges), 1)
                self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_CREATE)

    def test_recreate_object_csv(self):
        """CSV export an object, delete it, and recreate it via CSV import."""
        if hasattr(self, "get_deletable_object"):
            # provided by DeleteObjectViewTestCase mixin
            instance = self.get_deletable_object()
        else:
            # try to do it ourselves
            instance = testing.get_deletable_objects(self.model, self._get_queryset()).first()
        if instance is None:
            self.fail("Couldn't find a single deletable object!")

        # Add object-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["add", "view"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        response = self.client.get(self._get_detail_url(instance) + "?format=csv", **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertEqual(
            response.get("Content-Disposition"),
            f'attachment; filename="nautobot_{self.model.__name__.lower()}_data.csv"',
        )
        csv_data = response.content.decode(response.charset)

        serializer_class = get_serializer_for_model(self.model)
        old_serializer = serializer_class(instance, context={"request": None})
        old_data = old_serializer.data
        instance.delete()

        response = self.client.post(self._get_list_url(), csv_data, content_type="text/csv", **self.header)
        self.assertHttpStatus(response, status.HTTP_201_CREATED, csv_data)
        # Note that create via CSV is always treated as a bulk-create, and so the response is always a list of dicts
        new_instance = self._get_queryset().get(pk=response.data[0]["id"])
        self.assertNotEqual(new_instance.pk, instance.pk)

        new_serializer = serializer_class(new_instance, context={"request": None})
        new_data = new_serializer.data
        for field_name, field in new_serializer.fields.items():
            if field.read_only or field.write_only:
                continue
            if field_name in ["created", "last_updated"]:
                self.assertNotEqual(
                    old_data[field_name],
                    new_data[field_name],
                    f"{field_name} should have been updated on delete/recreate but it didn't change!",
                )
            else:
                self.assertEqual(
                    old_data[field_name],
                    new_data[field_name],
                    f"{field_name} should have been unchanged on delete/recreate but it differs!",
                )

    def test_bulk_create_objects(self):
        """
        POST a set of objects in a single request.
        """
        # Add object-level permission
        obj_perm = users_models.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))

        initial_count = self._get_queryset().count()
        response = self.client.post(self._get_list_url(), self.create_data, format="json", **self.header)
        self.assertHttpStatus(response, status.HTTP_201_CREATED)
        self.assertEqual(len(response.data), len(self.create_data))
        self.assertEqual(self._get_queryset().count(), initial_count + len(self.create_data))
        for i, obj in enumerate(response.data):
            for field in self.create_data[i]:
                if field not in self.validation_excluded_fields:
                    self.assertIn(
                        field,
                        obj,
                        f"Bulk create field '{field}' missing from object {i} in response",
                    )
        for i, obj in enumerate(response.data):
            self.assertInstanceEqual(
                self._get_queryset().get(pk=obj["id"]),
                self.create_data[i],
                exclude=self.validation_excluded_fields,
                api=True,
            )
            if self.slug_source is not None and "slug" not in self.create_data[i]:
                self.check_expected_slug(self._get_queryset().get(pk=obj["id"]))

test_bulk_create_objects()

POST a set of objects in a single request.

Source code in nautobot/core/testing/api.py

def test_bulk_create_objects(self):
    """
    POST a set of objects in a single request.
    """
    # Add object-level permission
    obj_perm = users_models.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))

    initial_count = self._get_queryset().count()
    response = self.client.post(self._get_list_url(), self.create_data, format="json", **self.header)
    self.assertHttpStatus(response, status.HTTP_201_CREATED)
    self.assertEqual(len(response.data), len(self.create_data))
    self.assertEqual(self._get_queryset().count(), initial_count + len(self.create_data))
    for i, obj in enumerate(response.data):
        for field in self.create_data[i]:
            if field not in self.validation_excluded_fields:
                self.assertIn(
                    field,
                    obj,
                    f"Bulk create field '{field}' missing from object {i} in response",
                )
    for i, obj in enumerate(response.data):
        self.assertInstanceEqual(
            self._get_queryset().get(pk=obj["id"]),
            self.create_data[i],
            exclude=self.validation_excluded_fields,
            api=True,
        )
        if self.slug_source is not None and "slug" not in self.create_data[i]:
            self.check_expected_slug(self._get_queryset().get(pk=obj["id"]))

test_create_object()

POST a single object with permission.

Source code in nautobot/core/testing/api.py

def test_create_object(self):
    """
    POST a single object with permission.
    """
    # Add object-level permission
    obj_perm = users_models.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))

    initial_count = self._get_queryset().count()
    for i, create_data in enumerate(self.create_data):
        if i == len(self.create_data) - 1:
            # Test to see if depth parameter is ignored in POST request.
            response = self.client.post(
                self._get_list_url() + "?depth=3", create_data, format="json", **self.header
            )
        else:
            response = self.client.post(self._get_list_url(), create_data, format="json", **self.header)
        self.assertHttpStatus(response, status.HTTP_201_CREATED)
        self.assertEqual(self._get_queryset().count(), initial_count + i + 1)
        instance = self._get_queryset().get(pk=response.data["id"])
        self.assertInstanceEqual(
            instance,
            create_data,
            exclude=self.validation_excluded_fields,
            api=True,
        )

        # Check if Slug field is automatically created
        if self.slug_source is not None and "slug" not in create_data:
            self.check_expected_slug(self._get_queryset().get(pk=response.data["id"]))

        # Verify ObjectChange creation
        if hasattr(self.model, "to_objectchange"):
            objectchanges = lookup.get_changes_for_model(instance)
            self.assertEqual(len(objectchanges), 1)
            self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_CREATE)

test_create_object_without_permission()

POST a single object without permission.

Source code in nautobot/core/testing/api.py

def test_create_object_without_permission(self):
    """
    POST a single object without permission.
    """
    url = self._get_list_url()

    # Try POST without permission
    with testing.disable_warnings("django.request"):
        response = self.client.post(url, self.create_data[0], format="json", **self.header)
        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

test_recreate_object_csv()

CSV export an object, delete it, and recreate it via CSV import.

Source code in nautobot/core/testing/api.py

def test_recreate_object_csv(self):
    """CSV export an object, delete it, and recreate it via CSV import."""
    if hasattr(self, "get_deletable_object"):
        # provided by DeleteObjectViewTestCase mixin
        instance = self.get_deletable_object()
    else:
        # try to do it ourselves
        instance = testing.get_deletable_objects(self.model, self._get_queryset()).first()
    if instance is None:
        self.fail("Couldn't find a single deletable object!")

    # Add object-level permission
    obj_perm = users_models.ObjectPermission(name="Test permission", actions=["add", "view"])
    obj_perm.save()
    obj_perm.users.add(self.user)
    obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

    response = self.client.get(self._get_detail_url(instance) + "?format=csv", **self.header)
    self.assertHttpStatus(response, status.HTTP_200_OK)
    self.assertEqual(
        response.get("Content-Disposition"),
        f'attachment; filename="nautobot_{self.model.__name__.lower()}_data.csv"',
    )
    csv_data = response.content.decode(response.charset)

    serializer_class = get_serializer_for_model(self.model)
    old_serializer = serializer_class(instance, context={"request": None})
    old_data = old_serializer.data
    instance.delete()

    response = self.client.post(self._get_list_url(), csv_data, content_type="text/csv", **self.header)
    self.assertHttpStatus(response, status.HTTP_201_CREATED, csv_data)
    # Note that create via CSV is always treated as a bulk-create, and so the response is always a list of dicts
    new_instance = self._get_queryset().get(pk=response.data[0]["id"])
    self.assertNotEqual(new_instance.pk, instance.pk)

    new_serializer = serializer_class(new_instance, context={"request": None})
    new_data = new_serializer.data
    for field_name, field in new_serializer.fields.items():
        if field.read_only or field.write_only:
            continue
        if field_name in ["created", "last_updated"]:
            self.assertNotEqual(
                old_data[field_name],
                new_data[field_name],
                f"{field_name} should have been updated on delete/recreate but it didn't change!",
            )
        else:
            self.assertEqual(
                old_data[field_name],
                new_data[field_name],
                f"{field_name} should have been unchanged on delete/recreate but it differs!",
            )

DeleteObjectViewTestCase

Bases: APITestCase

Source code in nautobot/core/testing/api.py

class DeleteObjectViewTestCase(APITestCase):
    def get_deletable_object(self):
        """
        Get an instance that can be deleted.

        For some models this may just be any random object, but when we have FKs with `on_delete=models.PROTECT`
        (as is often the case) we need to find or create an instance that doesn't have such entanglements.
        """
        instance = testing.get_deletable_objects(self.model, self._get_queryset()).first()
        if instance is None:
            self.fail("Couldn't find a single deletable object!")
        return instance

    def get_deletable_object_pks(self):
        """
        Get a list of PKs corresponding to objects that can be safely bulk-deleted.

        For some models this may just be any random objects, but when we have FKs with `on_delete=models.PROTECT`
        (as is often the case) we need to find or create an instance that doesn't have such entanglements.
        """
        instances = testing.get_deletable_objects(self.model, self._get_queryset()).values_list("pk", flat=True)[:3]
        if len(instances) < 3:
            self.fail(f"Couldn't find 3 deletable objects, only found {len(instances)}!")
        return instances

    def test_delete_object_without_permission(self):
        """
        DELETE a single object without permission.
        """
        url = self._get_detail_url(self.get_deletable_object())

        # Try DELETE without permission
        with testing.disable_warnings("django.request"):
            response = self.client.delete(url, **self.header)
            self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

    def test_delete_object(self):
        """
        DELETE a single object identified by its primary key.
        """
        instance = self.get_deletable_object()
        url = self._get_detail_url(instance)

        # Add object-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        response = self.client.delete(url, **self.header)
        self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
        self.assertFalse(self._get_queryset().filter(pk=instance.pk).exists())

        # Verify ObjectChange creation
        if hasattr(self.model, "to_objectchange"):
            objectchanges = lookup.get_changes_for_model(instance)
            self.assertEqual(len(objectchanges), 1)
            self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_DELETE)

    def test_bulk_delete_objects(self):
        """
        DELETE a set of objects in a single request.
        """
        id_list = self.get_deletable_object_pks()
        # Add object-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        data = [{"id": id} for id in id_list]

        initial_count = self._get_queryset().count()
        response = self.client.delete(self._get_list_url(), data, format="json", **self.header)
        self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
        self.assertEqual(self._get_queryset().count(), initial_count - len(id_list))

get_deletable_object()

Get an instance that can be deleted.

For some models this may just be any random object, but when we have FKs with on_delete=models.PROTECT (as is often the case) we need to find or create an instance that doesn't have such entanglements.

Source code in nautobot/core/testing/api.py

def get_deletable_object(self):
    """
    Get an instance that can be deleted.

    For some models this may just be any random object, but when we have FKs with `on_delete=models.PROTECT`
    (as is often the case) we need to find or create an instance that doesn't have such entanglements.
    """
    instance = testing.get_deletable_objects(self.model, self._get_queryset()).first()
    if instance is None:
        self.fail("Couldn't find a single deletable object!")
    return instance

get_deletable_object_pks()

Get a list of PKs corresponding to objects that can be safely bulk-deleted.

For some models this may just be any random objects, but when we have FKs with on_delete=models.PROTECT (as is often the case) we need to find or create an instance that doesn't have such entanglements.

Source code in nautobot/core/testing/api.py

def get_deletable_object_pks(self):
    """
    Get a list of PKs corresponding to objects that can be safely bulk-deleted.

    For some models this may just be any random objects, but when we have FKs with `on_delete=models.PROTECT`
    (as is often the case) we need to find or create an instance that doesn't have such entanglements.
    """
    instances = testing.get_deletable_objects(self.model, self._get_queryset()).values_list("pk", flat=True)[:3]
    if len(instances) < 3:
        self.fail(f"Couldn't find 3 deletable objects, only found {len(instances)}!")
    return instances

test_bulk_delete_objects()

DELETE a set of objects in a single request.

Source code in nautobot/core/testing/api.py

def test_bulk_delete_objects(self):
    """
    DELETE a set of objects in a single request.
    """
    id_list = self.get_deletable_object_pks()
    # Add object-level permission
    obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
    obj_perm.save()
    obj_perm.users.add(self.user)
    obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

    data = [{"id": id} for id in id_list]

    initial_count = self._get_queryset().count()
    response = self.client.delete(self._get_list_url(), data, format="json", **self.header)
    self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
    self.assertEqual(self._get_queryset().count(), initial_count - len(id_list))

test_delete_object()

DELETE a single object identified by its primary key.

Source code in nautobot/core/testing/api.py

def test_delete_object(self):
    """
    DELETE a single object identified by its primary key.
    """
    instance = self.get_deletable_object()
    url = self._get_detail_url(instance)

    # Add object-level permission
    obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
    obj_perm.save()
    obj_perm.users.add(self.user)
    obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

    response = self.client.delete(url, **self.header)
    self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
    self.assertFalse(self._get_queryset().filter(pk=instance.pk).exists())

    # Verify ObjectChange creation
    if hasattr(self.model, "to_objectchange"):
        objectchanges = lookup.get_changes_for_model(instance)
        self.assertEqual(len(objectchanges), 1)
        self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_DELETE)

test_delete_object_without_permission()

DELETE a single object without permission.

Source code in nautobot/core/testing/api.py

def test_delete_object_without_permission(self):
    """
    DELETE a single object without permission.
    """
    url = self._get_detail_url(self.get_deletable_object())

    # Try DELETE without permission
    with testing.disable_warnings("django.request"):
        response = self.client.delete(url, **self.header)
        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

GetObjectViewTestCase

Bases: APITestCase

Source code in nautobot/core/testing/api.py

class GetObjectViewTestCase(APITestCase):
    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_get_object_anonymous(self):
        """
        GET a single object as an unauthenticated user.
        """
        url = self._get_detail_url(self._get_queryset().first())
        if (
            self.model._meta.app_label,
            self.model._meta.model_name,
        ) in settings.EXEMPT_EXCLUDE_MODELS:
            # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)
        else:
            response = self.client.get(url, **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_get_object_without_permission(self):
        """
        GET a single object as an authenticated user without the required permission.
        """
        url = self._get_detail_url(self._get_queryset().first())

        # Try GET without permission
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_get_object(self):
        """
        GET a single object as an authenticated user with permission to view the object.
        """
        self.assertGreaterEqual(
            self._get_queryset().count(),
            2,
            f"Test requires the creation of at least two {self.model} instances",
        )
        instance1, instance2 = self._get_queryset()[:2]

        # Add object-level permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={"pk": instance1.pk},
            actions=["view"],
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try GET to non-permitted object
        url = self._get_detail_url(instance2)
        self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_404_NOT_FOUND)

        # Try GET to permitted object
        url = self._get_detail_url(instance1)
        response = self.client.get(url, **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertIsInstance(response.data, dict)
        # Fields that should be present in *ALL* model serializers:
        self.assertIn("id", response.data)
        self.assertEqual(str(response.data["id"]), str(instance1.pk))  # coerce to str to handle both int and uuid
        self.assertIn("url", response.data)
        self.assertIn("display", response.data)
        self.assertIsInstance(response.data["display"], str)
        # Fields that should be present in appropriate model serializers:
        if issubclass(self.model, extras_models.ChangeLoggedModel):
            self.assertIn("created", response.data)
            self.assertIn("last_updated", response.data)
        # Fields that should be absent by default (opt-in fields):
        self.assertNotIn("computed_fields", response.data)
        self.assertNotIn("relationships", response.data)

        # If opt-in fields are supported on this model, make sure they can be opted into

        custom_fields_registry = registry.registry["model_features"]["custom_fields"]
        # computed fields and custom fields use the same registry
        cf_supported = self.model._meta.model_name in custom_fields_registry.get(self.model._meta.app_label, {})
        if cf_supported:  # custom_fields is not an opt-in field, it should always be present if supported
            self.assertIn("custom_fields", response.data)
            self.assertIsInstance(response.data["custom_fields"], dict)

        relationships_registry = registry.registry["model_features"]["relationships"]
        rel_supported = self.model._meta.model_name in relationships_registry.get(self.model._meta.app_label, {})
        if cf_supported or rel_supported:
            query_params = []
            if cf_supported:
                query_params.append("include=computed_fields")
            if rel_supported:
                query_params.append("include=relationships")
            query_string = "&".join(query_params)
            url = f"{url}?{query_string}"

            response = self.client.get(url, **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertIsInstance(response.data, dict)
            if cf_supported:
                self.assertIn("computed_fields", response.data)
                self.assertIsInstance(response.data["computed_fields"], dict)
            else:
                self.assertNotIn("computed_fields", response.data)
            if rel_supported:
                self.assertIn("relationships", response.data)
                self.assertIsInstance(response.data["relationships"], dict)
            else:
                self.assertNotIn("relationships", response.data)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_options_object(self):
        """
        Make an OPTIONS request for a single object.
        """
        url = self._get_detail_url(self._get_queryset().first())
        response = self.client.options(url, **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)

        with self.subTest("Assert Detail View Config is generated well"):
            # Namings Help
            # 1. detail_view_config: This is the detail view config set in the serializer.Meta.detail_view_config
            # 2. detail_view_schema: This is the retrieve schema generated from an OPTIONS request.
            # 3. advanced_view_config: This is the advanced view config set in the serializer.Meta.advanced_view_config
            # 4. advanced_view_schema: This is the advanced tab schema generated from an OPTIONS request.
            serializer = get_serializer_for_model(self._get_queryset().model)
            advanced_view_schema = response.data["view_options"]["advanced"]

            if getattr(serializer.Meta, "advanced_view_config", None):
                # Get all custom advanced tab fields
                advanced_view_config = serializer.Meta.advanced_view_config
                self.assertGreaterEqual(len(advanced_view_schema), 1)
                advanced_tab_fields = []
                advanced_view_layout = advanced_view_config["layout"]
                for section in advanced_view_layout:
                    for value in section.values():
                        advanced_tab_fields += value["fields"]
            else:
                # Get default advanced tab fields
                self.assertEqual(len(advanced_view_schema), 1)
                self.assertIn("Object Details", advanced_view_schema[0])
                advanced_tab_fields = advanced_view_schema[0].get("Object Details")["fields"]

            if detail_view_config := getattr(serializer.Meta, "detail_view_config", None):
                detail_view_schema = response.data["view_options"]["retrieve"]
                self.assertHttpStatus(response, status.HTTP_200_OK)

                # According to convention, fields in the advanced tab fields should not exist in
                # the `detail_view_schema`. Assert this is True.
                with self.subTest("Assert advanced tab fields should not exist in the detail_view_schema."):
                    if detail_view_config.get("include_others"):
                        # Handle `Other Fields` specially as `Other Field` is dynamically added by nautobot and not part of the serializer view config
                        other_fields = detail_view_schema[0]["Other Fields"]["fields"]
                        for field in advanced_tab_fields:
                            self.assertNotIn(field, other_fields)

                    for col_idx, col in enumerate(detail_view_schema):
                        for group_idx, (group_title, group) in enumerate(col.items()):
                            if group_title == "Other Fields":
                                continue
                            group_fields = group["fields"]
                            # Config on the serializer
                            fields = detail_view_config["layout"][col_idx][group_title]["fields"]
                            # According to convention, advanced_tab_fields should only exist at the end of the first col group and should be
                            # deleted from any other places it may appear in the layout. Assert this is True.
                            if group_idx == 0 == col_idx:
                                self.assertFalse(
                                    any(
                                        field in advanced_tab_fields
                                        for field in group_fields[: -len(advanced_tab_fields)]
                                    )
                                )
                            else:
                                for field in group_fields:
                                    self.assertNotIn(field, advanced_tab_fields)
                            # Assert response from options correspond to the view config set on the serializer
                            for field in fields:
                                if field not in advanced_tab_fields:
                                    self.assertIn(field, group_fields)

test_get_object()

GET a single object as an authenticated user with permission to view the object.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_get_object(self):
    """
    GET a single object as an authenticated user with permission to view the object.
    """
    self.assertGreaterEqual(
        self._get_queryset().count(),
        2,
        f"Test requires the creation of at least two {self.model} instances",
    )
    instance1, instance2 = self._get_queryset()[:2]

    # Add object-level permission
    obj_perm = users_models.ObjectPermission(
        name="Test permission",
        constraints={"pk": instance1.pk},
        actions=["view"],
    )
    obj_perm.save()
    obj_perm.users.add(self.user)
    obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

    # Try GET to non-permitted object
    url = self._get_detail_url(instance2)
    self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_404_NOT_FOUND)

    # Try GET to permitted object
    url = self._get_detail_url(instance1)
    response = self.client.get(url, **self.header)
    self.assertHttpStatus(response, status.HTTP_200_OK)
    self.assertIsInstance(response.data, dict)
    # Fields that should be present in *ALL* model serializers:
    self.assertIn("id", response.data)
    self.assertEqual(str(response.data["id"]), str(instance1.pk))  # coerce to str to handle both int and uuid
    self.assertIn("url", response.data)
    self.assertIn("display", response.data)
    self.assertIsInstance(response.data["display"], str)
    # Fields that should be present in appropriate model serializers:
    if issubclass(self.model, extras_models.ChangeLoggedModel):
        self.assertIn("created", response.data)
        self.assertIn("last_updated", response.data)
    # Fields that should be absent by default (opt-in fields):
    self.assertNotIn("computed_fields", response.data)
    self.assertNotIn("relationships", response.data)

    # If opt-in fields are supported on this model, make sure they can be opted into

    custom_fields_registry = registry.registry["model_features"]["custom_fields"]
    # computed fields and custom fields use the same registry
    cf_supported = self.model._meta.model_name in custom_fields_registry.get(self.model._meta.app_label, {})
    if cf_supported:  # custom_fields is not an opt-in field, it should always be present if supported
        self.assertIn("custom_fields", response.data)
        self.assertIsInstance(response.data["custom_fields"], dict)

    relationships_registry = registry.registry["model_features"]["relationships"]
    rel_supported = self.model._meta.model_name in relationships_registry.get(self.model._meta.app_label, {})
    if cf_supported or rel_supported:
        query_params = []
        if cf_supported:
            query_params.append("include=computed_fields")
        if rel_supported:
            query_params.append("include=relationships")
        query_string = "&".join(query_params)
        url = f"{url}?{query_string}"

        response = self.client.get(url, **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertIsInstance(response.data, dict)
        if cf_supported:
            self.assertIn("computed_fields", response.data)
            self.assertIsInstance(response.data["computed_fields"], dict)
        else:
            self.assertNotIn("computed_fields", response.data)
        if rel_supported:
            self.assertIn("relationships", response.data)
            self.assertIsInstance(response.data["relationships"], dict)
        else:
            self.assertNotIn("relationships", response.data)

test_get_object_anonymous()

GET a single object as an unauthenticated user.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
def test_get_object_anonymous(self):
    """
    GET a single object as an unauthenticated user.
    """
    url = self._get_detail_url(self._get_queryset().first())
    if (
        self.model._meta.app_label,
        self.model._meta.model_name,
    ) in settings.EXEMPT_EXCLUDE_MODELS:
        # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)
    else:
        response = self.client.get(url, **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)

test_get_object_without_permission()

GET a single object as an authenticated user without the required permission.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_get_object_without_permission(self):
    """
    GET a single object as an authenticated user without the required permission.
    """
    url = self._get_detail_url(self._get_queryset().first())

    # Try GET without permission
    with testing.disable_warnings("django.request"):
        self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)

test_options_object()

Make an OPTIONS request for a single object.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
def test_options_object(self):
    """
    Make an OPTIONS request for a single object.
    """
    url = self._get_detail_url(self._get_queryset().first())
    response = self.client.options(url, **self.header)
    self.assertHttpStatus(response, status.HTTP_200_OK)

    with self.subTest("Assert Detail View Config is generated well"):
        # Namings Help
        # 1. detail_view_config: This is the detail view config set in the serializer.Meta.detail_view_config
        # 2. detail_view_schema: This is the retrieve schema generated from an OPTIONS request.
        # 3. advanced_view_config: This is the advanced view config set in the serializer.Meta.advanced_view_config
        # 4. advanced_view_schema: This is the advanced tab schema generated from an OPTIONS request.
        serializer = get_serializer_for_model(self._get_queryset().model)
        advanced_view_schema = response.data["view_options"]["advanced"]

        if getattr(serializer.Meta, "advanced_view_config", None):
            # Get all custom advanced tab fields
            advanced_view_config = serializer.Meta.advanced_view_config
            self.assertGreaterEqual(len(advanced_view_schema), 1)
            advanced_tab_fields = []
            advanced_view_layout = advanced_view_config["layout"]
            for section in advanced_view_layout:
                for value in section.values():
                    advanced_tab_fields += value["fields"]
        else:
            # Get default advanced tab fields
            self.assertEqual(len(advanced_view_schema), 1)
            self.assertIn("Object Details", advanced_view_schema[0])
            advanced_tab_fields = advanced_view_schema[0].get("Object Details")["fields"]

        if detail_view_config := getattr(serializer.Meta, "detail_view_config", None):
            detail_view_schema = response.data["view_options"]["retrieve"]
            self.assertHttpStatus(response, status.HTTP_200_OK)

            # According to convention, fields in the advanced tab fields should not exist in
            # the `detail_view_schema`. Assert this is True.
            with self.subTest("Assert advanced tab fields should not exist in the detail_view_schema."):
                if detail_view_config.get("include_others"):
                    # Handle `Other Fields` specially as `Other Field` is dynamically added by nautobot and not part of the serializer view config
                    other_fields = detail_view_schema[0]["Other Fields"]["fields"]
                    for field in advanced_tab_fields:
                        self.assertNotIn(field, other_fields)

                for col_idx, col in enumerate(detail_view_schema):
                    for group_idx, (group_title, group) in enumerate(col.items()):
                        if group_title == "Other Fields":
                            continue
                        group_fields = group["fields"]
                        # Config on the serializer
                        fields = detail_view_config["layout"][col_idx][group_title]["fields"]
                        # According to convention, advanced_tab_fields should only exist at the end of the first col group and should be
                        # deleted from any other places it may appear in the layout. Assert this is True.
                        if group_idx == 0 == col_idx:
                            self.assertFalse(
                                any(
                                    field in advanced_tab_fields
                                    for field in group_fields[: -len(advanced_tab_fields)]
                                )
                            )
                        else:
                            for field in group_fields:
                                self.assertNotIn(field, advanced_tab_fields)
                        # Assert response from options correspond to the view config set on the serializer
                        for field in fields:
                            if field not in advanced_tab_fields:
                                self.assertIn(field, group_fields)

ListObjectsViewTestCase

Bases: APITestCase

Source code in nautobot/core/testing/api.py

class ListObjectsViewTestCase(APITestCase):
    choices_fields = None
    filterset = None

    def get_filterset(self):
        return self.filterset or lookup.get_filterset_for_model(self.model)

    def get_depth_fields(self):
        """Get a list of model fields that could be tested with the ?depth query parameter"""
        depth_fields = []
        for field in self.model._meta.fields:
            if not field.name.startswith("_"):
                if isinstance(field, (ForeignKey, GenericForeignKey, ManyToManyField, core_fields.TagsField)) and (
                    # we represent content-types as "app_label.modelname" rather than as FKs
                    field.related_model != ContentType
                    # user is a model field on Token but not a field on TokenSerializer
                    and not (field.name == "user" and self.model == users_models.Token)
                ):
                    depth_fields.append(field.name)
        return depth_fields

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_list_objects_anonymous(self):
        """
        GET a list of objects as an unauthenticated user.
        """
        url = self._get_list_url()
        if (
            self.model._meta.app_label,
            self.model._meta.model_name,
        ) in settings.EXEMPT_EXCLUDE_MODELS:
            # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)
        else:
            # TODO(Glenn): if we're passing **self.header, we are *by definition* **NOT** anonymous!!
            response = self.client.get(url, **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertIsInstance(response.data, dict)
            self.assertIn("results", response.data)
            self.assertEqual(len(response.data["results"]), self._get_queryset().count())

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects_depth_0(self):
        """
        GET a list of objects using the "?depth=0" parameter.
        """
        depth_fields = self.get_depth_fields()
        self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
        url = f"{self._get_list_url()}?depth=0"
        response = self.client.get(url, **self.header)

        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertIsInstance(response.data, dict)
        self.assertIn("results", response.data)
        self.assertEqual(len(response.data["results"]), self._get_queryset().count())

        for response_data in response.data["results"]:
            for field in depth_fields:
                self.assertIn(field, response_data)
                if isinstance(response_data[field], list):
                    for entry in response_data[field]:
                        self.assertIsInstance(entry, dict)
                        self.assertTrue(is_uuid(entry["id"]))
                else:
                    if response_data[field] is not None:
                        self.assertIsInstance(response_data[field], dict)
                        url = response_data[field]["url"]
                        pk = response_data[field]["id"]
                        object_type = response_data[field]["object_type"]
                        # The response should be a brief API object, containing an ID, object_type, and URL ending in the UUID of the relevant object
                        # http://nautobot.example.com/api/circuits/providers/<uuid>/
                        #                                                    ^^^^^^
                        self.assertTrue(is_uuid(url.split("/")[-2]))
                        self.assertTrue(is_uuid(pk))

                        with self.subTest(f"Assert object_type {object_type} is valid"):
                            app_label, model_name = object_type.split(".")
                            ContentType.objects.get(app_label=app_label, model=model_name)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects_depth_1(self):
        """
        GET a list of objects using the "?depth=1" parameter.
        """
        depth_fields = self.get_depth_fields()
        self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
        url = f"{self._get_list_url()}?depth=1"
        response = self.client.get(url, **self.header)

        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertIsInstance(response.data, dict)
        self.assertIn("results", response.data)
        self.assertEqual(len(response.data["results"]), self._get_queryset().count())

        for response_data in response.data["results"]:
            for field in depth_fields:
                self.assertIn(field, response_data)
                if isinstance(response_data[field], list):
                    for entry in response_data[field]:
                        self.assertIsInstance(entry, dict)
                        self.assertTrue(is_uuid(entry["id"]))
                else:
                    if response_data[field] is not None:
                        self.assertIsInstance(response_data[field], dict)
                        self.assertTrue(is_uuid(response_data[field]["id"]))

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects_without_permission(self):
        """
        GET a list of objects as an authenticated user without the required permission.
        """
        url = self._get_list_url()

        # Try GET without permission
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects(self):
        """
        GET a list of objects as an authenticated user with permission to view the objects.
        """
        self.assertGreaterEqual(
            self._get_queryset().count(),
            3,
            f"Test requires the creation of at least three {self.model} instances",
        )
        instance1, instance2 = self._get_queryset()[:2]

        # Add object-level permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={"pk__in": [instance1.pk, instance2.pk]},
            actions=["view"],
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try GET to permitted objects
        response = self.client.get(self._get_list_url(), **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertIsInstance(response.data, dict)
        self.assertIn("results", response.data)
        self.assertEqual(len(response.data["results"]), 2)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects_filtered(self):
        """
        GET a list of objects filtered by ID.
        """
        self.assertGreaterEqual(
            self._get_queryset().count(),
            3,
            f"Test requires the creation of at least three {self.model} instances",
        )
        self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
        instance1, instance2 = self._get_queryset()[:2]
        response = self.client.get(f"{self._get_list_url()}?id={instance1.pk}&id={instance2.pk}", **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertIsInstance(response.data, dict)
        self.assertIn("results", response.data)
        self.assertEqual(len(response.data["results"]), 2)
        for entry in response.data["results"]:
            self.assertIn(str(entry["id"]), [str(instance1.pk), str(instance2.pk)])

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects_ascending_ordered(self):
        # Simple sorting check for models with a "name" field
        # TreeModels don't support sorting at this time (order_by is not supported by TreeQuerySet)
        #   They will pass api == queryset tests below but will fail the user expected sort test
        if hasattr(self.model, "name") and not issubclass(self.model, TreeModel):
            self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
            response = self.client.get(f"{self._get_list_url()}?sort=name&limit=3", **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            result_list = list(map(lambda p: p["name"], response.data["results"]))
            self.assertEqual(
                result_list,
                list(self._get_queryset().order_by("name").values_list("name", flat=True)[:3]),
                "API sort not identical to QuerySet.order_by",
            )

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects_descending_ordered(self):
        # Simple sorting check for models with a "name" field
        # TreeModels don't support sorting at this time (order_by is not supported by TreeQuerySet)
        #   They will pass api == queryset tests below but will fail the user expected sort test
        if hasattr(self.model, "name") and not issubclass(self.model, TreeModel):
            self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
            response = self.client.get(f"{self._get_list_url()}?sort=-name&limit=3", **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            result_list = list(map(lambda p: p["name"], response.data["results"]))
            self.assertEqual(
                result_list,
                list(self._get_queryset().order_by("-name").values_list("name", flat=True)[:3]),
                "API sort not identical to QuerySet.order_by",
            )

            response_ascending = self.client.get(f"{self._get_list_url()}?sort=name&limit=3", **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            result_list_ascending = list(map(lambda p: p["name"], response_ascending.data["results"]))

            self.assertNotEqual(
                result_list,
                result_list_ascending,
                "Same results obtained when sorting by name and by -name (QuerySet not ordering)",
            )

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[], STRICT_FILTERING=True)
    def test_list_objects_unknown_filter_strict_filtering(self):
        """
        GET a list of objects with an unknown filter parameter and strict filtering, expect a 400 response.
        """
        self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
        with testing.disable_warnings("django.request"):
            response = self.client.get(f"{self._get_list_url()}?ice_cream_flavor=rocky-road", **self.header)
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertIsInstance(response.data, dict)
        self.assertIn("ice_cream_flavor", response.data)
        self.assertIsInstance(response.data["ice_cream_flavor"], list)
        self.assertEqual("Unknown filter field", str(response.data["ice_cream_flavor"][0]))

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[], STRICT_FILTERING=False)
    def test_list_objects_unknown_filter_no_strict_filtering(self):
        """
        GET a list of objects with an unknown filter parameter and no strict filtering, expect it to be ignored.
        """
        self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
        with self.assertLogs("nautobot.core.filters") as cm:
            response = self.client.get(f"{self._get_list_url()}?ice_cream_flavor=rocky-road", **self.header)
        self.assertEqual(
            cm.output,
            [
                f"WARNING:nautobot.core.filters:{self.get_filterset().__name__}: "
                'Unknown filter field "ice_cream_flavor"',
            ],
        )
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertIsInstance(response.data, dict)
        self.assertIn("results", response.data)
        self.assertEqual(len(response.data["results"]), self._get_queryset().count())

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_options_objects(self):
        """
        Make an OPTIONS request for a list endpoint.
        """
        response = self.client.options(self._get_list_url(), **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects_csv(self):
        """
        GET a list of objects in CSV format as an authenticated user with permission to view some objects.
        """
        self.assertGreaterEqual(
            self._get_queryset().count(),
            3,
            f"Test requires the creation of at least three {self.model} instances",
        )
        instance1, instance2, instance3 = self._get_queryset()[:3]

        # Add object-level permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={"pk__in": [instance1.pk, instance2.pk]},
            actions=["view"],
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try filtered GET to objects specifying CSV format as a query parameter
        response_1 = self.client.get(
            f"{self._get_list_url()}?format=csv&id={instance1.pk}&id={instance3.pk}", **self.header
        )
        self.assertHttpStatus(response_1, status.HTTP_200_OK)
        self.assertEqual(response_1.get("Content-Type"), "text/csv; charset=UTF-8")
        self.assertEqual(
            response_1.get("Content-Disposition"),
            f'attachment; filename="nautobot_{self.model.__name__.lower()}_data.csv"',
        )

        # Try same request specifying CSV format via the ACCEPT header
        response_2 = self.client.get(
            f"{self._get_list_url()}?id={instance1.pk}&id={instance3.pk}", **self.header, HTTP_ACCEPT="text/csv"
        )
        self.assertHttpStatus(response_2, status.HTTP_200_OK)
        self.assertEqual(response_2.get("Content-Type"), "text/csv; charset=UTF-8")
        self.assertEqual(
            response_2.get("Content-Disposition"),
            f'attachment; filename="nautobot_{self.model.__name__.lower()}_data.csv"',
        )

        self.maxDiff = None
        # This check is more useful than it might seem. Any related object that wasn't CSV-converted correctly
        # will likely be rendered incorrectly as an API URL, and that API URL *will* differ between the
        # two responses based on the inclusion or omission of the "?format=csv" parameter. If
        # you run into this, make sure all serializers have `Meta.fields = "__all__"` set.
        self.assertEqual(
            response_1.content.decode(response_1.charset), response_2.content.decode(response_2.charset)
        )

        # Load the csv data back into a list of object dicts
        reader = csv.DictReader(StringIO(response_1.content.decode(response_1.charset)))
        rows = list(reader)
        # Should only have one entry (instance1) since we filtered out instance2 and permissions block instance3
        self.assertEqual(1, len(rows))
        self.assertEqual(rows[0]["id"], str(instance1.pk))
        self.assertEqual(rows[0]["display"], getattr(instance1, "display", str(instance1)))
        if hasattr(self.model, "_custom_field_data"):
            custom_fields = extras_models.CustomField.objects.get_for_model(self.model)
            for cf in custom_fields:
                self.assertIn(f"cf_{cf.key}", rows[0])
                self.assertEqual(rows[0][f"cf_{cf.key}"], instance1._custom_field_data.get(cf.key) or "")

get_depth_fields()

Get a list of model fields that could be tested with the ?depth query parameter

Source code in nautobot/core/testing/api.py

def get_depth_fields(self):
    """Get a list of model fields that could be tested with the ?depth query parameter"""
    depth_fields = []
    for field in self.model._meta.fields:
        if not field.name.startswith("_"):
            if isinstance(field, (ForeignKey, GenericForeignKey, ManyToManyField, core_fields.TagsField)) and (
                # we represent content-types as "app_label.modelname" rather than as FKs
                field.related_model != ContentType
                # user is a model field on Token but not a field on TokenSerializer
                and not (field.name == "user" and self.model == users_models.Token)
            ):
                depth_fields.append(field.name)
    return depth_fields

test_list_objects()

GET a list of objects as an authenticated user with permission to view the objects.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_list_objects(self):
    """
    GET a list of objects as an authenticated user with permission to view the objects.
    """
    self.assertGreaterEqual(
        self._get_queryset().count(),
        3,
        f"Test requires the creation of at least three {self.model} instances",
    )
    instance1, instance2 = self._get_queryset()[:2]

    # Add object-level permission
    obj_perm = users_models.ObjectPermission(
        name="Test permission",
        constraints={"pk__in": [instance1.pk, instance2.pk]},
        actions=["view"],
    )
    obj_perm.save()
    obj_perm.users.add(self.user)
    obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

    # Try GET to permitted objects
    response = self.client.get(self._get_list_url(), **self.header)
    self.assertHttpStatus(response, status.HTTP_200_OK)
    self.assertIsInstance(response.data, dict)
    self.assertIn("results", response.data)
    self.assertEqual(len(response.data["results"]), 2)

test_list_objects_anonymous()

GET a list of objects as an unauthenticated user.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
def test_list_objects_anonymous(self):
    """
    GET a list of objects as an unauthenticated user.
    """
    url = self._get_list_url()
    if (
        self.model._meta.app_label,
        self.model._meta.model_name,
    ) in settings.EXEMPT_EXCLUDE_MODELS:
        # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)
    else:
        # TODO(Glenn): if we're passing **self.header, we are *by definition* **NOT** anonymous!!
        response = self.client.get(url, **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertIsInstance(response.data, dict)
        self.assertIn("results", response.data)
        self.assertEqual(len(response.data["results"]), self._get_queryset().count())

test_list_objects_csv()

GET a list of objects in CSV format as an authenticated user with permission to view some objects.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_list_objects_csv(self):
    """
    GET a list of objects in CSV format as an authenticated user with permission to view some objects.
    """
    self.assertGreaterEqual(
        self._get_queryset().count(),
        3,
        f"Test requires the creation of at least three {self.model} instances",
    )
    instance1, instance2, instance3 = self._get_queryset()[:3]

    # Add object-level permission
    obj_perm = users_models.ObjectPermission(
        name="Test permission",
        constraints={"pk__in": [instance1.pk, instance2.pk]},
        actions=["view"],
    )
    obj_perm.save()
    obj_perm.users.add(self.user)
    obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

    # Try filtered GET to objects specifying CSV format as a query parameter
    response_1 = self.client.get(
        f"{self._get_list_url()}?format=csv&id={instance1.pk}&id={instance3.pk}", **self.header
    )
    self.assertHttpStatus(response_1, status.HTTP_200_OK)
    self.assertEqual(response_1.get("Content-Type"), "text/csv; charset=UTF-8")
    self.assertEqual(
        response_1.get("Content-Disposition"),
        f'attachment; filename="nautobot_{self.model.__name__.lower()}_data.csv"',
    )

    # Try same request specifying CSV format via the ACCEPT header
    response_2 = self.client.get(
        f"{self._get_list_url()}?id={instance1.pk}&id={instance3.pk}", **self.header, HTTP_ACCEPT="text/csv"
    )
    self.assertHttpStatus(response_2, status.HTTP_200_OK)
    self.assertEqual(response_2.get("Content-Type"), "text/csv; charset=UTF-8")
    self.assertEqual(
        response_2.get("Content-Disposition"),
        f'attachment; filename="nautobot_{self.model.__name__.lower()}_data.csv"',
    )

    self.maxDiff = None
    # This check is more useful than it might seem. Any related object that wasn't CSV-converted correctly
    # will likely be rendered incorrectly as an API URL, and that API URL *will* differ between the
    # two responses based on the inclusion or omission of the "?format=csv" parameter. If
    # you run into this, make sure all serializers have `Meta.fields = "__all__"` set.
    self.assertEqual(
        response_1.content.decode(response_1.charset), response_2.content.decode(response_2.charset)
    )

    # Load the csv data back into a list of object dicts
    reader = csv.DictReader(StringIO(response_1.content.decode(response_1.charset)))
    rows = list(reader)
    # Should only have one entry (instance1) since we filtered out instance2 and permissions block instance3
    self.assertEqual(1, len(rows))
    self.assertEqual(rows[0]["id"], str(instance1.pk))
    self.assertEqual(rows[0]["display"], getattr(instance1, "display", str(instance1)))
    if hasattr(self.model, "_custom_field_data"):
        custom_fields = extras_models.CustomField.objects.get_for_model(self.model)
        for cf in custom_fields:
            self.assertIn(f"cf_{cf.key}", rows[0])
            self.assertEqual(rows[0][f"cf_{cf.key}"], instance1._custom_field_data.get(cf.key) or "")

test_list_objects_depth_0()

GET a list of objects using the "?depth=0" parameter.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_list_objects_depth_0(self):
    """
    GET a list of objects using the "?depth=0" parameter.
    """
    depth_fields = self.get_depth_fields()
    self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
    url = f"{self._get_list_url()}?depth=0"
    response = self.client.get(url, **self.header)

    self.assertHttpStatus(response, status.HTTP_200_OK)
    self.assertIsInstance(response.data, dict)
    self.assertIn("results", response.data)
    self.assertEqual(len(response.data["results"]), self._get_queryset().count())

    for response_data in response.data["results"]:
        for field in depth_fields:
            self.assertIn(field, response_data)
            if isinstance(response_data[field], list):
                for entry in response_data[field]:
                    self.assertIsInstance(entry, dict)
                    self.assertTrue(is_uuid(entry["id"]))
            else:
                if response_data[field] is not None:
                    self.assertIsInstance(response_data[field], dict)
                    url = response_data[field]["url"]
                    pk = response_data[field]["id"]
                    object_type = response_data[field]["object_type"]
                    # The response should be a brief API object, containing an ID, object_type, and URL ending in the UUID of the relevant object
                    # http://nautobot.example.com/api/circuits/providers/<uuid>/
                    #                                                    ^^^^^^
                    self.assertTrue(is_uuid(url.split("/")[-2]))
                    self.assertTrue(is_uuid(pk))

                    with self.subTest(f"Assert object_type {object_type} is valid"):
                        app_label, model_name = object_type.split(".")
                        ContentType.objects.get(app_label=app_label, model=model_name)

test_list_objects_depth_1()

GET a list of objects using the "?depth=1" parameter.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_list_objects_depth_1(self):
    """
    GET a list of objects using the "?depth=1" parameter.
    """
    depth_fields = self.get_depth_fields()
    self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
    url = f"{self._get_list_url()}?depth=1"
    response = self.client.get(url, **self.header)

    self.assertHttpStatus(response, status.HTTP_200_OK)
    self.assertIsInstance(response.data, dict)
    self.assertIn("results", response.data)
    self.assertEqual(len(response.data["results"]), self._get_queryset().count())

    for response_data in response.data["results"]:
        for field in depth_fields:
            self.assertIn(field, response_data)
            if isinstance(response_data[field], list):
                for entry in response_data[field]:
                    self.assertIsInstance(entry, dict)
                    self.assertTrue(is_uuid(entry["id"]))
            else:
                if response_data[field] is not None:
                    self.assertIsInstance(response_data[field], dict)
                    self.assertTrue(is_uuid(response_data[field]["id"]))

test_list_objects_filtered()

GET a list of objects filtered by ID.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_list_objects_filtered(self):
    """
    GET a list of objects filtered by ID.
    """
    self.assertGreaterEqual(
        self._get_queryset().count(),
        3,
        f"Test requires the creation of at least three {self.model} instances",
    )
    self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
    instance1, instance2 = self._get_queryset()[:2]
    response = self.client.get(f"{self._get_list_url()}?id={instance1.pk}&id={instance2.pk}", **self.header)
    self.assertHttpStatus(response, status.HTTP_200_OK)
    self.assertIsInstance(response.data, dict)
    self.assertIn("results", response.data)
    self.assertEqual(len(response.data["results"]), 2)
    for entry in response.data["results"]:
        self.assertIn(str(entry["id"]), [str(instance1.pk), str(instance2.pk)])

test_list_objects_unknown_filter_no_strict_filtering()

GET a list of objects with an unknown filter parameter and no strict filtering, expect it to be ignored.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[], STRICT_FILTERING=False)
def test_list_objects_unknown_filter_no_strict_filtering(self):
    """
    GET a list of objects with an unknown filter parameter and no strict filtering, expect it to be ignored.
    """
    self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
    with self.assertLogs("nautobot.core.filters") as cm:
        response = self.client.get(f"{self._get_list_url()}?ice_cream_flavor=rocky-road", **self.header)
    self.assertEqual(
        cm.output,
        [
            f"WARNING:nautobot.core.filters:{self.get_filterset().__name__}: "
            'Unknown filter field "ice_cream_flavor"',
        ],
    )
    self.assertHttpStatus(response, status.HTTP_200_OK)
    self.assertIsInstance(response.data, dict)
    self.assertIn("results", response.data)
    self.assertEqual(len(response.data["results"]), self._get_queryset().count())

test_list_objects_unknown_filter_strict_filtering()

GET a list of objects with an unknown filter parameter and strict filtering, expect a 400 response.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[], STRICT_FILTERING=True)
def test_list_objects_unknown_filter_strict_filtering(self):
    """
    GET a list of objects with an unknown filter parameter and strict filtering, expect a 400 response.
    """
    self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
    with testing.disable_warnings("django.request"):
        response = self.client.get(f"{self._get_list_url()}?ice_cream_flavor=rocky-road", **self.header)
    self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
    self.assertIsInstance(response.data, dict)
    self.assertIn("ice_cream_flavor", response.data)
    self.assertIsInstance(response.data["ice_cream_flavor"], list)
    self.assertEqual("Unknown filter field", str(response.data["ice_cream_flavor"][0]))

test_list_objects_without_permission()

GET a list of objects as an authenticated user without the required permission.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_list_objects_without_permission(self):
    """
    GET a list of objects as an authenticated user without the required permission.
    """
    url = self._get_list_url()

    # Try GET without permission
    with testing.disable_warnings("django.request"):
        self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN)

test_options_objects()

Make an OPTIONS request for a list endpoint.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
def test_options_objects(self):
    """
    Make an OPTIONS request for a list endpoint.
    """
    response = self.client.options(self._get_list_url(), **self.header)
    self.assertHttpStatus(response, status.HTTP_200_OK)

NotesURLViewTestCase

Bases: APITestCase

Validate Notes URL on objects that have the Note model Mixin.

Source code in nautobot/core/testing/api.py

class NotesURLViewTestCase(APITestCase):
    """Validate Notes URL on objects that have the Note model Mixin."""

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_notes_url_on_object(self):
        notes = getattr(self.model, "notes", None)
        if notes and isinstance(notes, ForeignKey):
            instance1 = self._get_queryset().first()
            # Add object-level permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"pk": instance1.pk},
                actions=["view"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))
            url = self._get_detail_url(instance1)
            response = self.client.get(url, **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertIn("notes_url", response.data)
            self.assertIn(f"{url}notes/", str(response.data["notes_url"]))

TreeModelAPIViewTestCaseMixin

Test ?depth=2 query parameter for TreeModel

Source code in nautobot/core/testing/api.py

class TreeModelAPIViewTestCaseMixin:
    """Test `?depth=2` query parameter for TreeModel"""

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects_depth_2(self):
        """
        GET a list of objects using the "?depth=2" parameter.
        TreeModel Only
        """
        field = "parent"

        self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
        url = f"{self._get_list_url()}?depth=2"
        response = self.client.get(url, **self.header)

        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertIsInstance(response.data, dict)
        self.assertIn("results", response.data)
        self.assertEqual(len(response.data["results"]), self._get_queryset().count())

        response_data = response.data["results"]
        for data in response_data:
            # First Level Parent
            self.assertEqual(field in data, True)
            if data[field] is not None:
                self.assertIsInstance(data[field], dict)
                self.assertTrue(is_uuid(data[field]["id"]))
                # Second Level Parent
                self.assertIn(field, data[field])
                if data[field][field] is not None:
                    self.assertIsInstance(data[field][field], dict)
                    self.assertTrue(is_uuid(data[field][field]["id"]))

test_list_objects_depth_2()

GET a list of objects using the "?depth=2" parameter. TreeModel Only

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_list_objects_depth_2(self):
    """
    GET a list of objects using the "?depth=2" parameter.
    TreeModel Only
    """
    field = "parent"

    self.add_permissions(f"{self.model._meta.app_label}.view_{self.model._meta.model_name}")
    url = f"{self._get_list_url()}?depth=2"
    response = self.client.get(url, **self.header)

    self.assertHttpStatus(response, status.HTTP_200_OK)
    self.assertIsInstance(response.data, dict)
    self.assertIn("results", response.data)
    self.assertEqual(len(response.data["results"]), self._get_queryset().count())

    response_data = response.data["results"]
    for data in response_data:
        # First Level Parent
        self.assertEqual(field in data, True)
        if data[field] is not None:
            self.assertIsInstance(data[field], dict)
            self.assertTrue(is_uuid(data[field]["id"]))
            # Second Level Parent
            self.assertIn(field, data[field])
            if data[field][field] is not None:
                self.assertIsInstance(data[field][field], dict)
                self.assertTrue(is_uuid(data[field][field]["id"]))

UpdateObjectViewTestCase

Bases: APITestCase

Source code in nautobot/core/testing/api.py

class UpdateObjectViewTestCase(APITestCase):
    update_data = {}
    bulk_update_data: Optional[dict] = None
    validation_excluded_fields = []
    choices_fields = None

    def test_update_object_without_permission(self):
        """
        PATCH a single object without permission.
        """
        url = self._get_detail_url(self._get_queryset().first())
        update_data = self.update_data or getattr(self, "create_data")[0]

        # Try PATCH without permission
        with testing.disable_warnings("django.request"):
            response = self.client.patch(url, update_data, format="json", **self.header)
            self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

    def test_update_object(self):
        """
        PATCH a single object identified by its ID.
        """

        def strip_serialized_object(this_object):
            """
            Only here to work around acceptable differences in PATCH response vs GET response which are known bugs.
            """
            # Work around for https://github.com/nautobot/nautobot/issues/3321
            this_object.pop("last_updated", None)
            # PATCH response always includes "opt-in" fields, but GET response does not.
            this_object.pop("computed_fields", None)
            this_object.pop("config_context", None)
            this_object.pop("relationships", None)

            for value in this_object.values():
                if isinstance(value, dict):
                    strip_serialized_object(value)
                elif isinstance(value, list):
                    for list_dict in value:
                        if isinstance(list_dict, dict):
                            strip_serialized_object(list_dict)

        self.maxDiff = None
        instance = self._get_queryset().first()
        url = self._get_detail_url(instance)
        update_data = self.update_data or getattr(self, "create_data")[0]

        # Add object-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Verify that an empty PATCH results in no change to the object.
        # This is to catch issues like https://github.com/nautobot/nautobot/issues/3533

        # Add object-level permission for GET
        obj_perm.actions = ["view"]
        obj_perm.save()
        # Get initial serialized object representation
        get_response = self.client.get(url, **self.header)
        self.assertHttpStatus(get_response, status.HTTP_200_OK)
        initial_serialized_object = get_response.json()
        strip_serialized_object(initial_serialized_object)

        # Redefine object-level permission for PATCH
        obj_perm.actions = ["change"]
        obj_perm.save()

        # Send empty PATCH request
        response = self.client.patch(url, {}, format="json", **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        serialized_object = response.json()
        strip_serialized_object(serialized_object)
        self.assertEqual(initial_serialized_object, serialized_object)

        # Verify ObjectChange creation -- yes, even though nothing actually changed
        # This may change (hah) at some point -- see https://github.com/nautobot/nautobot/issues/3321
        if hasattr(self.model, "to_objectchange"):
            objectchanges = lookup.get_changes_for_model(instance)
            self.assertEqual(len(objectchanges), 1)
            self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_UPDATE)
            objectchanges.delete()

        # Verify that a PATCH with some data updates that data correctly.
        response = self.client.patch(url, update_data, format="json", **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        # Check for unexpected side effects on fields we DIDN'T intend to update
        for field in initial_serialized_object:
            if field not in update_data:
                self.assertEqual(initial_serialized_object[field], serialized_object[field])
        instance.refresh_from_db()
        self.assertInstanceEqual(instance, update_data, exclude=self.validation_excluded_fields, api=True)

        # Verify ObjectChange creation
        if hasattr(self.model, "to_objectchange"):
            objectchanges = lookup.get_changes_for_model(instance)
            self.assertEqual(len(objectchanges), 1)
            self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_UPDATE)

    def test_get_put_round_trip(self):
        """GET and then PUT an object and verify that it's accepted and unchanged."""
        self.maxDiff = None
        # Add object-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view", "change"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        instance = self._get_queryset().first()
        url = self._get_detail_url(instance)

        # GET object representation
        opt_in_fields = getattr(get_serializer_for_model(self.model).Meta, "opt_in_fields", None)
        if opt_in_fields:
            url += "?" + "&".join([f"include={field}" for field in opt_in_fields])
        get_response = self.client.get(url, **self.header)
        self.assertHttpStatus(get_response, status.HTTP_200_OK)
        initial_serialized_object = get_response.json()

        # PUT same object representation
        put_response = self.client.put(url, initial_serialized_object, format="json", **self.header)
        self.assertHttpStatus(put_response, status.HTTP_200_OK, initial_serialized_object)
        updated_serialized_object = put_response.json()

        # Work around for https://github.com/nautobot/nautobot/issues/3321
        initial_serialized_object.pop("last_updated", None)
        updated_serialized_object.pop("last_updated", None)
        self.assertEqual(initial_serialized_object, updated_serialized_object)

    def test_bulk_update_objects(self):
        """
        PATCH a set of objects in a single request.
        """
        if self.bulk_update_data is None:
            self.skipTest("Bulk update data not set")

        # Add object-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        id_list = list(self._get_queryset().values_list("id", flat=True)[:3])
        self.assertEqual(len(id_list), 3, "Insufficient number of objects to test bulk update")
        data = [{"id": id, **self.bulk_update_data} for id in id_list]

        response = self.client.patch(self._get_list_url(), data, format="json", **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        for i, obj in enumerate(response.data):
            for field, _value in self.bulk_update_data.items():
                self.assertIn(
                    field,
                    obj,
                    f"Bulk update field '{field}' missing from object {i} in response",
                )
                # TODO(Glenn): shouldn't we also check that obj[field] == value?
        for instance in self._get_queryset().filter(pk__in=id_list):
            self.assertInstanceEqual(
                instance,
                self.bulk_update_data,
                exclude=self.validation_excluded_fields,
                api=True,
            )

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_options_returns_expected_choices(self):
        """
        Make an OPTIONS request for a list endpoint and validate choices match expected choices for serializer.
        """
        # Set self.choices_fields as empty set to compare classes that shouldn't have any choices on serializer.
        if not self.choices_fields:
            self.choices_fields = set()

        # Save self.user as superuser to be able to view available choices on list views.
        self.user.is_superuser = True
        self.user.save()

        response = self.client.options(self._get_list_url(), **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        data = response.json()

        self.assertIn("actions", data)

        # Grab any field that has choices defined (fields with enums)
        if any(
            [
                "POST" in data["actions"],
                "PUT" in data["actions"],
            ]
        ):
            schema = data["schema"]
            props = schema["properties"]
            fields = props.keys()
            field_choices = set()
            for field_name in fields:
                obj = props[field_name]
                if "enum" in obj and "enumNames" in obj:
                    enum = obj["enum"]
                    # Zipping to assert that the enum and the mapping have the same number of items.
                    model_field_choices = dict(zip(obj["enumNames"], enum))
                    self.assertEqual(len(enum), len(model_field_choices))
                    field_choices.add(field_name)
        else:
            self.fail(f"Neither PUT nor POST are available actions in: {data['actions']}")

        self.assertEqual(
            set(self.choices_fields),
            field_choices,
            "All field names of choice fields for a given model serializer need to be manually added to "
            "self.choices_fields. If this is already the case, perhaps the serializer is implemented incorrectly?",
        )

test_bulk_update_objects()

PATCH a set of objects in a single request.

Source code in nautobot/core/testing/api.py

def test_bulk_update_objects(self):
    """
    PATCH a set of objects in a single request.
    """
    if self.bulk_update_data is None:
        self.skipTest("Bulk update data not set")

    # Add object-level permission
    obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
    obj_perm.save()
    obj_perm.users.add(self.user)
    obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

    id_list = list(self._get_queryset().values_list("id", flat=True)[:3])
    self.assertEqual(len(id_list), 3, "Insufficient number of objects to test bulk update")
    data = [{"id": id, **self.bulk_update_data} for id in id_list]

    response = self.client.patch(self._get_list_url(), data, format="json", **self.header)
    self.assertHttpStatus(response, status.HTTP_200_OK)
    for i, obj in enumerate(response.data):
        for field, _value in self.bulk_update_data.items():
            self.assertIn(
                field,
                obj,
                f"Bulk update field '{field}' missing from object {i} in response",
            )
            # TODO(Glenn): shouldn't we also check that obj[field] == value?
    for instance in self._get_queryset().filter(pk__in=id_list):
        self.assertInstanceEqual(
            instance,
            self.bulk_update_data,
            exclude=self.validation_excluded_fields,
            api=True,
        )

test_get_put_round_trip()

GET and then PUT an object and verify that it's accepted and unchanged.

Source code in nautobot/core/testing/api.py

def test_get_put_round_trip(self):
    """GET and then PUT an object and verify that it's accepted and unchanged."""
    self.maxDiff = None
    # Add object-level permission
    obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view", "change"])
    obj_perm.save()
    obj_perm.users.add(self.user)
    obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

    instance = self._get_queryset().first()
    url = self._get_detail_url(instance)

    # GET object representation
    opt_in_fields = getattr(get_serializer_for_model(self.model).Meta, "opt_in_fields", None)
    if opt_in_fields:
        url += "?" + "&".join([f"include={field}" for field in opt_in_fields])
    get_response = self.client.get(url, **self.header)
    self.assertHttpStatus(get_response, status.HTTP_200_OK)
    initial_serialized_object = get_response.json()

    # PUT same object representation
    put_response = self.client.put(url, initial_serialized_object, format="json", **self.header)
    self.assertHttpStatus(put_response, status.HTTP_200_OK, initial_serialized_object)
    updated_serialized_object = put_response.json()

    # Work around for https://github.com/nautobot/nautobot/issues/3321
    initial_serialized_object.pop("last_updated", None)
    updated_serialized_object.pop("last_updated", None)
    self.assertEqual(initial_serialized_object, updated_serialized_object)

test_options_returns_expected_choices()

Make an OPTIONS request for a list endpoint and validate choices match expected choices for serializer.

Source code in nautobot/core/testing/api.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_options_returns_expected_choices(self):
    """
    Make an OPTIONS request for a list endpoint and validate choices match expected choices for serializer.
    """
    # Set self.choices_fields as empty set to compare classes that shouldn't have any choices on serializer.
    if not self.choices_fields:
        self.choices_fields = set()

    # Save self.user as superuser to be able to view available choices on list views.
    self.user.is_superuser = True
    self.user.save()

    response = self.client.options(self._get_list_url(), **self.header)
    self.assertHttpStatus(response, status.HTTP_200_OK)
    data = response.json()

    self.assertIn("actions", data)

    # Grab any field that has choices defined (fields with enums)
    if any(
        [
            "POST" in data["actions"],
            "PUT" in data["actions"],
        ]
    ):
        schema = data["schema"]
        props = schema["properties"]
        fields = props.keys()
        field_choices = set()
        for field_name in fields:
            obj = props[field_name]
            if "enum" in obj and "enumNames" in obj:
                enum = obj["enum"]
                # Zipping to assert that the enum and the mapping have the same number of items.
                model_field_choices = dict(zip(obj["enumNames"], enum))
                self.assertEqual(len(enum), len(model_field_choices))
                field_choices.add(field_name)
    else:
        self.fail(f"Neither PUT nor POST are available actions in: {data['actions']}")

    self.assertEqual(
        set(self.choices_fields),
        field_choices,
        "All field names of choice fields for a given model serializer need to be manually added to "
        "self.choices_fields. If this is already the case, perhaps the serializer is implemented incorrectly?",
    )

test_update_object()

PATCH a single object identified by its ID.

Source code in nautobot/core/testing/api.py

def test_update_object(self):
    """
    PATCH a single object identified by its ID.
    """

    def strip_serialized_object(this_object):
        """
        Only here to work around acceptable differences in PATCH response vs GET response which are known bugs.
        """
        # Work around for https://github.com/nautobot/nautobot/issues/3321
        this_object.pop("last_updated", None)
        # PATCH response always includes "opt-in" fields, but GET response does not.
        this_object.pop("computed_fields", None)
        this_object.pop("config_context", None)
        this_object.pop("relationships", None)

        for value in this_object.values():
            if isinstance(value, dict):
                strip_serialized_object(value)
            elif isinstance(value, list):
                for list_dict in value:
                    if isinstance(list_dict, dict):
                        strip_serialized_object(list_dict)

    self.maxDiff = None
    instance = self._get_queryset().first()
    url = self._get_detail_url(instance)
    update_data = self.update_data or getattr(self, "create_data")[0]

    # Add object-level permission
    obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
    obj_perm.save()
    obj_perm.users.add(self.user)
    obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

    # Verify that an empty PATCH results in no change to the object.
    # This is to catch issues like https://github.com/nautobot/nautobot/issues/3533

    # Add object-level permission for GET
    obj_perm.actions = ["view"]
    obj_perm.save()
    # Get initial serialized object representation
    get_response = self.client.get(url, **self.header)
    self.assertHttpStatus(get_response, status.HTTP_200_OK)
    initial_serialized_object = get_response.json()
    strip_serialized_object(initial_serialized_object)

    # Redefine object-level permission for PATCH
    obj_perm.actions = ["change"]
    obj_perm.save()

    # Send empty PATCH request
    response = self.client.patch(url, {}, format="json", **self.header)
    self.assertHttpStatus(response, status.HTTP_200_OK)
    serialized_object = response.json()
    strip_serialized_object(serialized_object)
    self.assertEqual(initial_serialized_object, serialized_object)

    # Verify ObjectChange creation -- yes, even though nothing actually changed
    # This may change (hah) at some point -- see https://github.com/nautobot/nautobot/issues/3321
    if hasattr(self.model, "to_objectchange"):
        objectchanges = lookup.get_changes_for_model(instance)
        self.assertEqual(len(objectchanges), 1)
        self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_UPDATE)
        objectchanges.delete()

    # Verify that a PATCH with some data updates that data correctly.
    response = self.client.patch(url, update_data, format="json", **self.header)
    self.assertHttpStatus(response, status.HTTP_200_OK)
    # Check for unexpected side effects on fields we DIDN'T intend to update
    for field in initial_serialized_object:
        if field not in update_data:
            self.assertEqual(initial_serialized_object[field], serialized_object[field])
    instance.refresh_from_db()
    self.assertInstanceEqual(instance, update_data, exclude=self.validation_excluded_fields, api=True)

    # Verify ObjectChange creation
    if hasattr(self.model, "to_objectchange"):
        objectchanges = lookup.get_changes_for_model(instance)
        self.assertEqual(len(objectchanges), 1)
        self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_UPDATE)

test_update_object_without_permission()

PATCH a single object without permission.

Source code in nautobot/core/testing/api.py

def test_update_object_without_permission(self):
    """
    PATCH a single object without permission.
    """
    url = self._get_detail_url(self._get_queryset().first())
    update_data = self.update_data or getattr(self, "create_data")[0]

    # Try PATCH without permission
    with testing.disable_warnings("django.request"):
        response = self.client.patch(url, update_data, format="json", **self.header)
        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

nautobot.apps.testing.FilterTestCases

Source code in nautobot/core/testing/filters.py

@tag("unit")
class FilterTestCases:
    class BaseFilterTestCase(views.TestCase):
        """Base class for testing of FilterSets."""

        def get_filterset_test_values(self, field_name, queryset=None):
            """Returns a list of distinct values from the requested queryset field to use in filterset tests.

            Returns a list for use in testing multiple choice filters. The size of the returned list is random
            but will contain at minimum 2 unique values. The list of values will match at least 2 instances when
            passed to the queryset's filter(field_name__in=[]) method but will fail to match at least one instance.

            Args:
                field_name (str): The name of the field to retrieve test values from.
                queryset (QuerySet): The queryset to retrieve test values. Defaults to `self.queryset`.

            Returns:
                (list): A list of unique values derived from the queryset.

            Raises:
                ValueError: Raised if unable to find a combination of 2 or more unique values
                    to filter the queryset to a subset of the total instances.
            """
            test_values = []
            if queryset is None:
                queryset = self.queryset
            qs_count = queryset.count()
            values_with_count = queryset.values(field_name).annotate(count=Count(field_name)).order_by("count")
            for value in values_with_count:
                # randomly break out of loop after 2 values have been selected
                if len(test_values) > 1 and random.choice([True, False]):
                    break
                if value[field_name] and value["count"] < qs_count:
                    qs_count -= value["count"]
                    test_values.append(str(value[field_name]))

            if len(test_values) < 2:
                raise ValueError(
                    f"Cannot find enough valid test data for {queryset.model._meta.object_name} field {field_name} "
                    f"(found {len(test_values)} option(s): {test_values}) but need at least 2 of them"
                )
            return test_values

    class FilterTestCase(BaseFilterTestCase):
        """Add common tests for all FilterSets."""

        queryset = None
        filterset = None

        # list of filters to be tested by `test_filters_generic`
        # list of iterables with filter name and optional field name
        # example:
        #   generic_filter_tests = [
        #       ["filter1"],
        #       ["filter2", "field2__name"],
        #   ]
        generic_filter_tests = []

        def test_id(self):
            """Verify that the filterset supports filtering by id."""
            params = {"id": self.queryset.values_list("pk", flat=True)[:2]}
            filterset = self.filterset(params, self.queryset)
            self.assertTrue(filterset.is_valid())
            self.assertEqual(filterset.qs.count(), 2)

        def test_invalid_filter(self):
            """Verify that the filterset reports as invalid when initialized with an unsupported filter parameter."""
            params = {"ice_cream_flavor": ["chocolate"]}
            self.assertFalse(self.filterset(params, self.queryset).is_valid())

        def test_filters_generic(self):
            """Test all multiple choice filters declared in `self.generic_filter_tests`.

            This test uses `get_filterset_test_values()` to retrieve a valid set of test data and asserts
            that the filterset filter output matches the corresponding queryset filter.
            The majority of Nautobot filters use conjoined=False, so the extra logic to support conjoined=True has not
            been implemented here. TagFilter and similar "AND" filters are not supported.

            Examples:
                Multiple tests can be performed for the same filter by adding multiple entries in
                `generic_filter_tests` with explicit field names.
                For example, to test a NaturalKeyOrPKMultipleChoiceFilter, use:
                    generic_filter_tests = (
                        ["filter_name", "field_name__name"],
                        ["filter_name", "field_name__id"],
                    )

                If a field name is not declared, the filter name will be used for the field name:
                    generic_filter_tests = (
                        ["devices"],
                    )
                This expects a field named `devices` on the model and a filter named `devices` on the filterset.
            """
            if not self.generic_filter_tests:
                self.skipTest("No generic_filter_tests defined?")

            for test in self.generic_filter_tests:
                filter_name = test[0]
                field_name = test[-1]  # default to filter_name if a second list item was not supplied
                with self.subTest(f"{self.filterset.__name__} filter {filter_name} ({field_name})"):
                    test_data = self.get_filterset_test_values(field_name)
                    params = {filter_name: test_data}
                    filterset_result = self.filterset(params, self.queryset).qs
                    qs_result = self.queryset.filter(**{f"{field_name}__in": test_data}).distinct()
                    self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)

        def test_boolean_filters_generic(self):
            """Test all `RelatedMembershipBooleanFilter` filters found in `self.filterset.get_filters()`
            except for the ones with custom filter logic defined in its `method` attribute.

            This test asserts that `filter=True` matches `self.queryset.filter(field__isnull=False)` and
            that `filter=False` matches `self.queryset.filter(field__isnull=True)`.
            """
            for filter_name, filter_object in self.filterset.get_filters().items():
                if not isinstance(filter_object, RelatedMembershipBooleanFilter):
                    continue
                if filter_object.method is not None:
                    continue
                field_name = filter_object.field_name
                with self.subTest(f"{self.filterset.__name__} RelatedMembershipBooleanFilter {filter_name} (True)"):
                    filterset_result = self.filterset({filter_name: True}, self.queryset).qs
                    qs_result = self.queryset.filter(**{f"{field_name}__isnull": False}).distinct()
                    self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)
                with self.subTest(f"{self.filterset.__name__} RelatedMembershipBooleanFilter {filter_name} (False)"):
                    filterset_result = self.filterset({filter_name: False}, self.queryset).qs
                    qs_result = self.queryset.filter(**{f"{field_name}__isnull": True}).distinct()
                    self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)

        def test_tags_filter(self):
            """Test the `tags` filter which should be present on all PrimaryModel filtersets."""
            if not issubclass(self.queryset.model, PrimaryModel):
                self.skipTest("Not a PrimaryModel")

            # Find an instance with at least two tags (should be common given our factory design)
            for instance in list(self.queryset):
                if len(instance.tags.all()) >= 2:
                    tags = list(instance.tags.all()[:2])
                    break
            else:
                self.fail(f"Couldn't find any {self.queryset.model._meta.object_name} with at least two Tags.")
            params = {"tags": [tags[0].name, tags[1].pk]}
            filterset_result = self.filterset(params, self.queryset).qs
            # Tags is an AND filter not an OR filter
            qs_result = self.queryset.filter(tags=tags[0]).filter(tags=tags[1]).distinct()
            self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)

    class NameOnlyFilterTestCase(FilterTestCase):
        """Add simple tests for filtering by name."""

        def test_name(self):
            """Verify that the filterset supports filtering by name."""
            params = {"name": list(self.queryset.values_list("name", flat=True)[:2])}
            filterset = self.filterset(params, self.queryset)
            self.assertTrue(filterset.is_valid())
            self.assertQuerysetEqualAndNotEmpty(
                filterset.qs.order_by("name"), self.queryset.filter(name__in=params["name"]).order_by("name")
            )

    class NameSlugFilterTestCase(NameOnlyFilterTestCase):
        """Add simple tests for filtering by name and by slug."""

        def test_slug(self):
            """Verify that the filterset supports filtering by slug."""
            params = {"slug": self.queryset.values_list("slug", flat=True)[:2]}
            filterset = self.filterset(params, self.queryset)
            self.assertTrue(filterset.is_valid())
            self.assertEqual(filterset.qs.count(), 2)

    class TenancyFilterTestCaseMixin(views.TestCase):
        """Add test cases for tenant and tenant-group filters."""

        tenancy_related_name = ""

        def test_tenant(self):
            tenants = list(models.Tenant.objects.filter(**{f"{self.tenancy_related_name}__isnull": False}))[:2]
            params = {"tenant_id": [tenants[0].pk, tenants[1].pk]}
            self.assertQuerysetEqual(
                self.filterset(params, self.queryset).qs, self.queryset.filter(tenant__in=tenants), ordered=False
            )
            params = {"tenant": [tenants[0].name, tenants[1].name]}
            self.assertQuerysetEqual(
                self.filterset(params, self.queryset).qs, self.queryset.filter(tenant__in=tenants), ordered=False
            )

        def test_tenant_group(self):
            tenant_groups = list(
                models.TenantGroup.objects.filter(
                    tenants__isnull=False, **{f"tenants__{self.tenancy_related_name}__isnull": False}
                )
            )[:2]
            tenant_groups_including_children = []
            for tenant_group in tenant_groups:
                tenant_groups_including_children += tenant_group.descendants(include_self=True)

            params = {"tenant_group": [tenant_groups[0].pk, tenant_groups[1].pk]}
            self.assertQuerysetEqual(
                self.filterset(params, self.queryset).qs,
                self.queryset.filter(tenant__tenant_group__in=tenant_groups_including_children),
                ordered=False,
            )

            params = {"tenant_group": [tenant_groups[0].name, tenant_groups[1].name]}
            self.assertQuerysetEqual(
                self.filterset(params, self.queryset).qs,
                self.queryset.filter(tenant__tenant_group__in=tenant_groups_including_children),
                ordered=False,
            )

BaseFilterTestCase

Bases: views.TestCase

Base class for testing of FilterSets.

Source code in nautobot/core/testing/filters.py

class BaseFilterTestCase(views.TestCase):
    """Base class for testing of FilterSets."""

    def get_filterset_test_values(self, field_name, queryset=None):
        """Returns a list of distinct values from the requested queryset field to use in filterset tests.

        Returns a list for use in testing multiple choice filters. The size of the returned list is random
        but will contain at minimum 2 unique values. The list of values will match at least 2 instances when
        passed to the queryset's filter(field_name__in=[]) method but will fail to match at least one instance.

        Args:
            field_name (str): The name of the field to retrieve test values from.
            queryset (QuerySet): The queryset to retrieve test values. Defaults to `self.queryset`.

        Returns:
            (list): A list of unique values derived from the queryset.

        Raises:
            ValueError: Raised if unable to find a combination of 2 or more unique values
                to filter the queryset to a subset of the total instances.
        """
        test_values = []
        if queryset is None:
            queryset = self.queryset
        qs_count = queryset.count()
        values_with_count = queryset.values(field_name).annotate(count=Count(field_name)).order_by("count")
        for value in values_with_count:
            # randomly break out of loop after 2 values have been selected
            if len(test_values) > 1 and random.choice([True, False]):
                break
            if value[field_name] and value["count"] < qs_count:
                qs_count -= value["count"]
                test_values.append(str(value[field_name]))

        if len(test_values) < 2:
            raise ValueError(
                f"Cannot find enough valid test data for {queryset.model._meta.object_name} field {field_name} "
                f"(found {len(test_values)} option(s): {test_values}) but need at least 2 of them"
            )
        return test_values

get_filterset_test_values(field_name, queryset=None)

Returns a list of distinct values from the requested queryset field to use in filterset tests.

Returns a list for use in testing multiple choice filters. The size of the returned list is random but will contain at minimum 2 unique values. The list of values will match at least 2 instances when passed to the queryset's filter(field_name__in=[]) method but will fail to match at least one instance.

Parameters:

Name	Type	Description	Default
field_name	str	The name of the field to retrieve test values from.	required
queryset	QuerySet	The queryset to retrieve test values. Defaults to self.queryset.	None

Returns:

Type	Description
list	A list of unique values derived from the queryset.

Raises:

Type	Description
ValueError	Raised if unable to find a combination of 2 or more unique values to filter the queryset to a subset of the total instances.

Source code in nautobot/core/testing/filters.py

def get_filterset_test_values(self, field_name, queryset=None):
    """Returns a list of distinct values from the requested queryset field to use in filterset tests.

    Returns a list for use in testing multiple choice filters. The size of the returned list is random
    but will contain at minimum 2 unique values. The list of values will match at least 2 instances when
    passed to the queryset's filter(field_name__in=[]) method but will fail to match at least one instance.

    Args:
        field_name (str): The name of the field to retrieve test values from.
        queryset (QuerySet): The queryset to retrieve test values. Defaults to `self.queryset`.

    Returns:
        (list): A list of unique values derived from the queryset.

    Raises:
        ValueError: Raised if unable to find a combination of 2 or more unique values
            to filter the queryset to a subset of the total instances.
    """
    test_values = []
    if queryset is None:
        queryset = self.queryset
    qs_count = queryset.count()
    values_with_count = queryset.values(field_name).annotate(count=Count(field_name)).order_by("count")
    for value in values_with_count:
        # randomly break out of loop after 2 values have been selected
        if len(test_values) > 1 and random.choice([True, False]):
            break
        if value[field_name] and value["count"] < qs_count:
            qs_count -= value["count"]
            test_values.append(str(value[field_name]))

    if len(test_values) < 2:
        raise ValueError(
            f"Cannot find enough valid test data for {queryset.model._meta.object_name} field {field_name} "
            f"(found {len(test_values)} option(s): {test_values}) but need at least 2 of them"
        )
    return test_values

FilterTestCase

Bases: BaseFilterTestCase

Add common tests for all FilterSets.

Source code in nautobot/core/testing/filters.py

class FilterTestCase(BaseFilterTestCase):
    """Add common tests for all FilterSets."""

    queryset = None
    filterset = None

    # list of filters to be tested by `test_filters_generic`
    # list of iterables with filter name and optional field name
    # example:
    #   generic_filter_tests = [
    #       ["filter1"],
    #       ["filter2", "field2__name"],
    #   ]
    generic_filter_tests = []

    def test_id(self):
        """Verify that the filterset supports filtering by id."""
        params = {"id": self.queryset.values_list("pk", flat=True)[:2]}
        filterset = self.filterset(params, self.queryset)
        self.assertTrue(filterset.is_valid())
        self.assertEqual(filterset.qs.count(), 2)

    def test_invalid_filter(self):
        """Verify that the filterset reports as invalid when initialized with an unsupported filter parameter."""
        params = {"ice_cream_flavor": ["chocolate"]}
        self.assertFalse(self.filterset(params, self.queryset).is_valid())

    def test_filters_generic(self):
        """Test all multiple choice filters declared in `self.generic_filter_tests`.

        This test uses `get_filterset_test_values()` to retrieve a valid set of test data and asserts
        that the filterset filter output matches the corresponding queryset filter.
        The majority of Nautobot filters use conjoined=False, so the extra logic to support conjoined=True has not
        been implemented here. TagFilter and similar "AND" filters are not supported.

        Examples:
            Multiple tests can be performed for the same filter by adding multiple entries in
            `generic_filter_tests` with explicit field names.
            For example, to test a NaturalKeyOrPKMultipleChoiceFilter, use:
                generic_filter_tests = (
                    ["filter_name", "field_name__name"],
                    ["filter_name", "field_name__id"],
                )

            If a field name is not declared, the filter name will be used for the field name:
                generic_filter_tests = (
                    ["devices"],
                )
            This expects a field named `devices` on the model and a filter named `devices` on the filterset.
        """
        if not self.generic_filter_tests:
            self.skipTest("No generic_filter_tests defined?")

        for test in self.generic_filter_tests:
            filter_name = test[0]
            field_name = test[-1]  # default to filter_name if a second list item was not supplied
            with self.subTest(f"{self.filterset.__name__} filter {filter_name} ({field_name})"):
                test_data = self.get_filterset_test_values(field_name)
                params = {filter_name: test_data}
                filterset_result = self.filterset(params, self.queryset).qs
                qs_result = self.queryset.filter(**{f"{field_name}__in": test_data}).distinct()
                self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)

    def test_boolean_filters_generic(self):
        """Test all `RelatedMembershipBooleanFilter` filters found in `self.filterset.get_filters()`
        except for the ones with custom filter logic defined in its `method` attribute.

        This test asserts that `filter=True` matches `self.queryset.filter(field__isnull=False)` and
        that `filter=False` matches `self.queryset.filter(field__isnull=True)`.
        """
        for filter_name, filter_object in self.filterset.get_filters().items():
            if not isinstance(filter_object, RelatedMembershipBooleanFilter):
                continue
            if filter_object.method is not None:
                continue
            field_name = filter_object.field_name
            with self.subTest(f"{self.filterset.__name__} RelatedMembershipBooleanFilter {filter_name} (True)"):
                filterset_result = self.filterset({filter_name: True}, self.queryset).qs
                qs_result = self.queryset.filter(**{f"{field_name}__isnull": False}).distinct()
                self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)
            with self.subTest(f"{self.filterset.__name__} RelatedMembershipBooleanFilter {filter_name} (False)"):
                filterset_result = self.filterset({filter_name: False}, self.queryset).qs
                qs_result = self.queryset.filter(**{f"{field_name}__isnull": True}).distinct()
                self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)

    def test_tags_filter(self):
        """Test the `tags` filter which should be present on all PrimaryModel filtersets."""
        if not issubclass(self.queryset.model, PrimaryModel):
            self.skipTest("Not a PrimaryModel")

        # Find an instance with at least two tags (should be common given our factory design)
        for instance in list(self.queryset):
            if len(instance.tags.all()) >= 2:
                tags = list(instance.tags.all()[:2])
                break
        else:
            self.fail(f"Couldn't find any {self.queryset.model._meta.object_name} with at least two Tags.")
        params = {"tags": [tags[0].name, tags[1].pk]}
        filterset_result = self.filterset(params, self.queryset).qs
        # Tags is an AND filter not an OR filter
        qs_result = self.queryset.filter(tags=tags[0]).filter(tags=tags[1]).distinct()
        self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)

test_boolean_filters_generic()

Test all RelatedMembershipBooleanFilter filters found in self.filterset.get_filters() except for the ones with custom filter logic defined in its method attribute.

This test asserts that filter=True matches self.queryset.filter(field__isnull=False) and that filter=False matches self.queryset.filter(field__isnull=True).

Source code in nautobot/core/testing/filters.py

def test_boolean_filters_generic(self):
    """Test all `RelatedMembershipBooleanFilter` filters found in `self.filterset.get_filters()`
    except for the ones with custom filter logic defined in its `method` attribute.

    This test asserts that `filter=True` matches `self.queryset.filter(field__isnull=False)` and
    that `filter=False` matches `self.queryset.filter(field__isnull=True)`.
    """
    for filter_name, filter_object in self.filterset.get_filters().items():
        if not isinstance(filter_object, RelatedMembershipBooleanFilter):
            continue
        if filter_object.method is not None:
            continue
        field_name = filter_object.field_name
        with self.subTest(f"{self.filterset.__name__} RelatedMembershipBooleanFilter {filter_name} (True)"):
            filterset_result = self.filterset({filter_name: True}, self.queryset).qs
            qs_result = self.queryset.filter(**{f"{field_name}__isnull": False}).distinct()
            self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)
        with self.subTest(f"{self.filterset.__name__} RelatedMembershipBooleanFilter {filter_name} (False)"):
            filterset_result = self.filterset({filter_name: False}, self.queryset).qs
            qs_result = self.queryset.filter(**{f"{field_name}__isnull": True}).distinct()
            self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)

test_filters_generic()

Test all multiple choice filters declared in self.generic_filter_tests.

This test uses get_filterset_test_values() to retrieve a valid set of test data and asserts that the filterset filter output matches the corresponding queryset filter. The majority of Nautobot filters use conjoined=False, so the extra logic to support conjoined=True has not been implemented here. TagFilter and similar "AND" filters are not supported.

Examples:

Multiple tests can be performed for the same filter by adding multiple entries in generic_filter_tests with explicit field names. For example, to test a NaturalKeyOrPKMultipleChoiceFilter, use: generic_filter_tests = ( ["filter_name", "field_name__name"], ["filter_name", "field_name__id"], )

If a field name is not declared, the filter name will be used for the field name: generic_filter_tests = ( ["devices"], ) This expects a field named devices on the model and a filter named devices on the filterset.

Source code in nautobot/core/testing/filters.py

def test_filters_generic(self):
    """Test all multiple choice filters declared in `self.generic_filter_tests`.

    This test uses `get_filterset_test_values()` to retrieve a valid set of test data and asserts
    that the filterset filter output matches the corresponding queryset filter.
    The majority of Nautobot filters use conjoined=False, so the extra logic to support conjoined=True has not
    been implemented here. TagFilter and similar "AND" filters are not supported.

    Examples:
        Multiple tests can be performed for the same filter by adding multiple entries in
        `generic_filter_tests` with explicit field names.
        For example, to test a NaturalKeyOrPKMultipleChoiceFilter, use:
            generic_filter_tests = (
                ["filter_name", "field_name__name"],
                ["filter_name", "field_name__id"],
            )

        If a field name is not declared, the filter name will be used for the field name:
            generic_filter_tests = (
                ["devices"],
            )
        This expects a field named `devices` on the model and a filter named `devices` on the filterset.
    """
    if not self.generic_filter_tests:
        self.skipTest("No generic_filter_tests defined?")

    for test in self.generic_filter_tests:
        filter_name = test[0]
        field_name = test[-1]  # default to filter_name if a second list item was not supplied
        with self.subTest(f"{self.filterset.__name__} filter {filter_name} ({field_name})"):
            test_data = self.get_filterset_test_values(field_name)
            params = {filter_name: test_data}
            filterset_result = self.filterset(params, self.queryset).qs
            qs_result = self.queryset.filter(**{f"{field_name}__in": test_data}).distinct()
            self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)

test_id()

Verify that the filterset supports filtering by id.

Source code in nautobot/core/testing/filters.py

def test_id(self):
    """Verify that the filterset supports filtering by id."""
    params = {"id": self.queryset.values_list("pk", flat=True)[:2]}
    filterset = self.filterset(params, self.queryset)
    self.assertTrue(filterset.is_valid())
    self.assertEqual(filterset.qs.count(), 2)

test_invalid_filter()

Verify that the filterset reports as invalid when initialized with an unsupported filter parameter.

Source code in nautobot/core/testing/filters.py

def test_invalid_filter(self):
    """Verify that the filterset reports as invalid when initialized with an unsupported filter parameter."""
    params = {"ice_cream_flavor": ["chocolate"]}
    self.assertFalse(self.filterset(params, self.queryset).is_valid())

test_tags_filter()

Test the tags filter which should be present on all PrimaryModel filtersets.

Source code in nautobot/core/testing/filters.py

def test_tags_filter(self):
    """Test the `tags` filter which should be present on all PrimaryModel filtersets."""
    if not issubclass(self.queryset.model, PrimaryModel):
        self.skipTest("Not a PrimaryModel")

    # Find an instance with at least two tags (should be common given our factory design)
    for instance in list(self.queryset):
        if len(instance.tags.all()) >= 2:
            tags = list(instance.tags.all()[:2])
            break
    else:
        self.fail(f"Couldn't find any {self.queryset.model._meta.object_name} with at least two Tags.")
    params = {"tags": [tags[0].name, tags[1].pk]}
    filterset_result = self.filterset(params, self.queryset).qs
    # Tags is an AND filter not an OR filter
    qs_result = self.queryset.filter(tags=tags[0]).filter(tags=tags[1]).distinct()
    self.assertQuerysetEqualAndNotEmpty(filterset_result, qs_result)

NameOnlyFilterTestCase

Bases: FilterTestCase

Add simple tests for filtering by name.

Source code in nautobot/core/testing/filters.py

class NameOnlyFilterTestCase(FilterTestCase):
    """Add simple tests for filtering by name."""

    def test_name(self):
        """Verify that the filterset supports filtering by name."""
        params = {"name": list(self.queryset.values_list("name", flat=True)[:2])}
        filterset = self.filterset(params, self.queryset)
        self.assertTrue(filterset.is_valid())
        self.assertQuerysetEqualAndNotEmpty(
            filterset.qs.order_by("name"), self.queryset.filter(name__in=params["name"]).order_by("name")
        )

test_name()

Verify that the filterset supports filtering by name.

Source code in nautobot/core/testing/filters.py

def test_name(self):
    """Verify that the filterset supports filtering by name."""
    params = {"name": list(self.queryset.values_list("name", flat=True)[:2])}
    filterset = self.filterset(params, self.queryset)
    self.assertTrue(filterset.is_valid())
    self.assertQuerysetEqualAndNotEmpty(
        filterset.qs.order_by("name"), self.queryset.filter(name__in=params["name"]).order_by("name")
    )

NameSlugFilterTestCase

Bases: NameOnlyFilterTestCase

Add simple tests for filtering by name and by slug.

Source code in nautobot/core/testing/filters.py

class NameSlugFilterTestCase(NameOnlyFilterTestCase):
    """Add simple tests for filtering by name and by slug."""

    def test_slug(self):
        """Verify that the filterset supports filtering by slug."""
        params = {"slug": self.queryset.values_list("slug", flat=True)[:2]}
        filterset = self.filterset(params, self.queryset)
        self.assertTrue(filterset.is_valid())
        self.assertEqual(filterset.qs.count(), 2)

test_slug()

Verify that the filterset supports filtering by slug.

Source code in nautobot/core/testing/filters.py

def test_slug(self):
    """Verify that the filterset supports filtering by slug."""
    params = {"slug": self.queryset.values_list("slug", flat=True)[:2]}
    filterset = self.filterset(params, self.queryset)
    self.assertTrue(filterset.is_valid())
    self.assertEqual(filterset.qs.count(), 2)

TenancyFilterTestCaseMixin

Bases: views.TestCase

Add test cases for tenant and tenant-group filters.

Source code in nautobot/core/testing/filters.py

class TenancyFilterTestCaseMixin(views.TestCase):
    """Add test cases for tenant and tenant-group filters."""

    tenancy_related_name = ""

    def test_tenant(self):
        tenants = list(models.Tenant.objects.filter(**{f"{self.tenancy_related_name}__isnull": False}))[:2]
        params = {"tenant_id": [tenants[0].pk, tenants[1].pk]}
        self.assertQuerysetEqual(
            self.filterset(params, self.queryset).qs, self.queryset.filter(tenant__in=tenants), ordered=False
        )
        params = {"tenant": [tenants[0].name, tenants[1].name]}
        self.assertQuerysetEqual(
            self.filterset(params, self.queryset).qs, self.queryset.filter(tenant__in=tenants), ordered=False
        )

    def test_tenant_group(self):
        tenant_groups = list(
            models.TenantGroup.objects.filter(
                tenants__isnull=False, **{f"tenants__{self.tenancy_related_name}__isnull": False}
            )
        )[:2]
        tenant_groups_including_children = []
        for tenant_group in tenant_groups:
            tenant_groups_including_children += tenant_group.descendants(include_self=True)

        params = {"tenant_group": [tenant_groups[0].pk, tenant_groups[1].pk]}
        self.assertQuerysetEqual(
            self.filterset(params, self.queryset).qs,
            self.queryset.filter(tenant__tenant_group__in=tenant_groups_including_children),
            ordered=False,
        )

        params = {"tenant_group": [tenant_groups[0].name, tenant_groups[1].name]}
        self.assertQuerysetEqual(
            self.filterset(params, self.queryset).qs,
            self.queryset.filter(tenant__tenant_group__in=tenant_groups_including_children),
            ordered=False,
        )

nautobot.apps.testing.ViewTestCases

We keep any TestCases with test_* methods inside a class to prevent unittest from trying to run them.

Source code in nautobot/core/testing/views.py

@tag("unit")
class ViewTestCases:
    """
    We keep any TestCases with test_* methods inside a class to prevent unittest from trying to run them.
    """

    class GetObjectViewTestCase(ModelViewTestCase):
        """
        Retrieve a single instance.
        """

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_get_object_anonymous(self):
            # Make the request as an unauthenticated user
            self.client.logout()
            response = self.client.get(self._get_queryset().first().get_absolute_url())
            self.assertHttpStatus(response, 200)
            response_body = response.content.decode(response.charset)
            self.assertIn(
                "/login/?next=" + self._get_queryset().first().get_absolute_url(), response_body, msg=response_body
            )

            # The "Change Log" tab should appear in the response since we have all exempt permissions
            if issubclass(self.model, extras_models.ChangeLoggedModel):
                response_body = testing.extract_page_body(response.content.decode(response.charset))
                self.assertIn("Change Log", response_body, msg=response_body)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_get_object_without_permission(self):
            instance = self._get_queryset().first()

            # Try GET without permission
            with testing.disable_warnings("django.request"):
                response = self.client.get(instance.get_absolute_url())
                self.assertHttpStatus(response, [403, 404])
                response_body = response.content.decode(response.charset)
                self.assertNotIn("/login/", response_body, msg=response_body)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_get_object_with_permission(self):
            instance = self._get_queryset().first()

            # Add model-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view"])
            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
            response = self.client.get(instance.get_absolute_url())
            self.assertHttpStatus(response, 200)

            response_body = testing.extract_page_body(response.content.decode(response.charset))

            # The object's display name or string representation should appear in the response
            self.assertIn(escape(getattr(instance, "display", str(instance))), response_body, msg=response_body)

            # If any Relationships are defined, they should appear in the response
            if self.relationships is not None:
                for relationship in self.relationships:  # false positive pylint: disable=not-an-iterable
                    content_type = ContentType.objects.get_for_model(instance)
                    if content_type == relationship.source_type:
                        self.assertIn(
                            escape(relationship.get_label(extras_choices.RelationshipSideChoices.SIDE_SOURCE)),
                            response_body,
                            msg=response_body,
                        )
                    if content_type == relationship.destination_type:
                        self.assertIn(
                            escape(relationship.get_label(extras_choices.RelationshipSideChoices.SIDE_DESTINATION)),
                            response_body,
                            msg=response_body,
                        )

            # If any Custom Fields are defined, they should appear in the response
            if self.custom_fields is not None:
                for custom_field in self.custom_fields:  # false positive pylint: disable=not-an-iterable
                    self.assertIn(escape(str(custom_field)), response_body, msg=response_body)
                    if custom_field.type == extras_choices.CustomFieldTypeChoices.TYPE_MULTISELECT:
                        for value in instance.cf.get(custom_field.key):
                            self.assertIn(escape(str(value)), response_body, msg=response_body)
                    else:
                        self.assertIn(
                            escape(str(instance.cf.get(custom_field.key) or "")), response_body, msg=response_body
                        )

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_get_object_with_constrained_permission(self):
            instance1, instance2 = self._get_queryset().all()[:2]

            # Add object-level permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"pk": instance1.pk},
                # To get a different rendering flow than the `test_get_object_with_permission` test above,
                # enable additional permissions for this object so that add/edit/delete buttons are rendered.
                actions=["view", "add", "change", "delete"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try GET to permitted object
            self.assertHttpStatus(self.client.get(instance1.get_absolute_url()), 200)

            # Try GET to non-permitted object
            self.assertHttpStatus(self.client.get(instance2.get_absolute_url()), 404)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_has_advanced_tab(self):
            instance = self._get_queryset().first()

            # Add model-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            response = self.client.get(instance.get_absolute_url())
            response_body = testing.extract_page_body(response.content.decode(response.charset))
            advanced_tab_href = f"{instance.get_absolute_url()}#advanced"

            self.assertIn(advanced_tab_href, response_body)
            self.assertIn("Advanced", response_body)

    class GetObjectChangelogViewTestCase(ModelViewTestCase):
        """
        View the changelog for an instance.
        """

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_get_object_changelog(self):
            url = self._get_url("changelog", self._get_queryset().first())
            response = self.client.get(url)
            self.assertHttpStatus(response, 200)

    class GetObjectNotesViewTestCase(ModelViewTestCase):
        """
        View the notes for an instance.
        """

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_get_object_notes(self):
            if hasattr(self.model, "notes"):
                url = self._get_url("notes", self._get_queryset().first())
                response = self.client.get(url)
                self.assertHttpStatus(response, 200)

    class CreateObjectViewTestCase(ModelViewTestCase):
        """
        Create a single new instance.

        :form_data: Data to be used when creating a new object.
        """

        form_data = {}
        slug_source = None
        slugify_function = staticmethod(slugify)
        slug_test_object = ""

        def test_create_object_without_permission(self):
            # Try GET without permission
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.get(self._get_url("add")), 403)

            # Try POST without permission
            request = {
                "path": self._get_url("add"),
                "data": testing.post_data(self.form_data),
            }
            response = self.client.post(**request)
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(response, 403)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_create_object_with_permission(self):
            initial_count = self._get_queryset().count()

            # Assign unconstrained permission
            obj_perm = users_models.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("add")), 200)

            # Try POST with model-level permission
            request = {
                "path": self._get_url("add"),
                "data": testing.post_data(self.form_data),
            }
            self.assertHttpStatus(self.client.post(**request), 302)
            self.assertEqual(initial_count + 1, self._get_queryset().count())
            # order_by() is no supported by django TreeNode,
            # So we directly retrieve the instance by "slug" or "name".
            if isinstance(self._get_queryset().first(), TreeNode):
                filter_by = self.slug_source if getattr(self, "slug_source", None) else "name"
                instance = self._get_queryset().get(**{filter_by: self.form_data.get(filter_by)})
                self.assertInstanceEqual(instance, self.form_data)
            else:
                if hasattr(self.model, "last_updated"):
                    instance = self._get_queryset().order_by("last_updated").last()
                    self.assertInstanceEqual(instance, self.form_data)
                else:
                    instance = self._get_queryset().last()
                    self.assertInstanceEqual(instance, self.form_data)

            if hasattr(self.model, "to_objectchange"):
                # Verify ObjectChange creation
                objectchanges = lookup.get_changes_for_model(instance)
                self.assertEqual(len(objectchanges), 1)
                self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_CREATE)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_create_object_with_constrained_permission(self):
            initial_count = self._get_queryset().count()

            # Assign constrained permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"pk": str(uuid.uuid4())},  # Match a non-existent pk (i.e., deny all)
                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 object-level permission
            self.assertHttpStatus(self.client.get(self._get_url("add")), 200)

            # Try to create an object (not permitted)
            request = {
                "path": self._get_url("add"),
                "data": testing.post_data(self.form_data),
            }
            self.assertHttpStatus(self.client.post(**request), 200)
            self.assertEqual(initial_count, self._get_queryset().count())  # Check that no object was created

            # Update the ObjectPermission to allow creation
            obj_perm.constraints = {"pk__isnull": False}
            obj_perm.save()

            # Try to create an object (permitted)
            request = {
                "path": self._get_url("add"),
                "data": testing.post_data(self.form_data),
            }
            self.assertHttpStatus(self.client.post(**request), 302)
            self.assertEqual(initial_count + 1, self._get_queryset().count())
            # order_by() is no supported by django TreeNode,
            # So we directly retrieve the instance by "slug".
            if isinstance(self._get_queryset().first(), TreeNode):
                filter_by = self.slug_source if getattr(self, "slug_source", None) else "name"
                instance = self._get_queryset().get(**{filter_by: self.form_data.get(filter_by)})
                self.assertInstanceEqual(instance, self.form_data)
            else:
                if hasattr(self.model, "last_updated"):
                    self.assertInstanceEqual(self._get_queryset().order_by("last_updated").last(), self.form_data)
                else:
                    self.assertInstanceEqual(self._get_queryset().last(), self.form_data)

        def test_slug_autocreation(self):
            """Test that slug is autocreated through ORM."""
            # This really should go on a models test page, but we don't have test structures for models.
            if getattr(self.model, "slug_source", None) is not None:
                obj = self.model.objects.get(**{self.slug_source: self.slug_test_object})
                expected_slug = self.slugify_function(getattr(obj, self.slug_source))
                self.assertEqual(obj.slug, expected_slug)

        def test_slug_not_modified(self):
            """Ensure save method does not modify slug that is passed in."""
            # This really should go on a models test page, but we don't have test structures for models.
            if getattr(self.model, "slug_source", None) is not None:
                new_slug_source_value = "kwyjibo"

                obj = self.model.objects.get(**{self.slug_source: self.slug_test_object})
                expected_slug = self.slugify_function(getattr(obj, self.slug_source))
                # Update slug source field str
                filter_ = self.slug_source + "__exact"
                self.model.objects.filter(**{filter_: self.slug_test_object}).update(
                    **{self.slug_source: new_slug_source_value}
                )

                obj.refresh_from_db()
                self.assertEqual(getattr(obj, self.slug_source), new_slug_source_value)
                self.assertEqual(obj.slug, expected_slug)

    class EditObjectViewTestCase(ModelViewTestCase):
        """
        Edit a single existing instance.

        :form_data: Data to be used when updating the first existing object.
        """

        form_data = {}

        def test_edit_object_without_permission(self):
            instance = self._get_queryset().first()

            # Try GET without permission
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.get(self._get_url("edit", instance)), [403, 404])

            # Try POST without permission
            request = {
                "path": self._get_url("edit", instance),
                "data": testing.post_data(self.form_data),
            }
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.post(**request), [403, 404])

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_edit_object_with_permission(self):
            instance = self._get_queryset().first()

            # Assign model-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
            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("edit", instance)), 200)

            # Try POST with model-level permission
            request = {
                "path": self._get_url("edit", instance),
                "data": testing.post_data(self.form_data),
            }
            self.assertHttpStatus(self.client.post(**request), 302)
            self.assertInstanceEqual(self._get_queryset().get(pk=instance.pk), self.form_data)

            if hasattr(self.model, "to_objectchange"):
                # Verify ObjectChange creation
                objectchanges = lookup.get_changes_for_model(instance)
                self.assertEqual(len(objectchanges), 1)
                self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_UPDATE)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_edit_object_with_constrained_permission(self):
            instance1, instance2 = self._get_queryset().all()[:2]

            # Assign constrained permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"pk": instance1.pk},
                actions=["change"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try GET with a permitted object
            self.assertHttpStatus(self.client.get(self._get_url("edit", instance1)), 200)

            # Try GET with a non-permitted object
            self.assertHttpStatus(self.client.get(self._get_url("edit", instance2)), 404)

            # Try to edit a permitted object
            request = {
                "path": self._get_url("edit", instance1),
                "data": testing.post_data(self.form_data),
            }
            self.assertHttpStatus(self.client.post(**request), 302)
            self.assertInstanceEqual(self._get_queryset().get(pk=instance1.pk), self.form_data)

            # Try to edit a non-permitted object
            request = {
                "path": self._get_url("edit", instance2),
                "data": testing.post_data(self.form_data),
            }
            self.assertHttpStatus(self.client.post(**request), 404)

    class DeleteObjectViewTestCase(ModelViewTestCase):
        """
        Delete a single instance.
        """

        def get_deletable_object(self):
            """
            Get an instance that can be deleted.

            For some models this may just be any random object, but when we have FKs with `on_delete=models.PROTECT`
            (as is often the case) we need to find or create an instance that doesn't have such entanglements.
            """
            instance = testing.get_deletable_objects(self.model, self._get_queryset()).first()
            if instance is None:
                self.fail("Couldn't find a single deletable object!")
            return instance

        def test_delete_object_without_permission(self):
            instance = self.get_deletable_object()

            # Try GET without permission
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.get(self._get_url("delete", instance)), [403, 404])

            # Try POST without permission
            request = {
                "path": self._get_url("delete", instance),
                "data": testing.post_data({"confirm": True}),
            }
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.post(**request), [403, 404])

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_delete_object_with_permission(self):
            instance = self.get_deletable_object()

            # Assign model-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
            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("delete", instance)), 200)

            # Try POST with model-level permission
            request = {
                "path": self._get_url("delete", instance),
                "data": testing.post_data({"confirm": True}),
            }
            self.assertHttpStatus(self.client.post(**request), 302)
            with self.assertRaises(ObjectDoesNotExist):
                self._get_queryset().get(pk=instance.pk)

            if hasattr(self.model, "to_objectchange"):
                # Verify ObjectChange creation
                objectchanges = lookup.get_changes_for_model(instance)
                self.assertEqual(len(objectchanges), 1)
                self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_DELETE)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_delete_object_with_permission_and_xwwwformurlencoded(self):
            instance = self.get_deletable_object()

            # Assign model-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
            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("delete", instance)), 200)

            # Try POST with model-level permission
            request = {
                "path": self._get_url("delete", instance),
                "data": urlencode({"confirm": True}),
                "content_type": "application/x-www-form-urlencoded",
            }
            self.assertHttpStatus(self.client.post(**request), 302)
            with self.assertRaises(ObjectDoesNotExist):
                self._get_queryset().get(pk=instance.pk)

            if hasattr(self.model, "to_objectchange"):
                # Verify ObjectChange creation
                objectchanges = lookup.get_changes_for_model(instance)
                self.assertEqual(len(objectchanges), 1)
                self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_DELETE)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_delete_object_with_constrained_permission(self):
            instance1 = self.get_deletable_object()
            instance2 = self._get_queryset().exclude(pk=instance1.pk)[0]

            # Assign object-level permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"pk": instance1.pk},
                actions=["delete"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try GET with a permitted object
            self.assertHttpStatus(self.client.get(self._get_url("delete", instance1)), 200)

            # Try GET with a non-permitted object
            self.assertHttpStatus(self.client.get(self._get_url("delete", instance2)), 404)

            # Try to delete a permitted object
            request = {
                "path": self._get_url("delete", instance1),
                "data": testing.post_data({"confirm": True}),
            }
            self.assertHttpStatus(self.client.post(**request), 302)
            with self.assertRaises(ObjectDoesNotExist):
                self._get_queryset().get(pk=instance1.pk)

            # Try to delete a non-permitted object
            # Note that in the case of tree models, deleting instance1 above may have cascade-deleted to instance2,
            # so to be safe, we need to get another object instance that definitely exists:
            instance3 = self._get_queryset().first()
            request = {
                "path": self._get_url("delete", instance3),
                "data": testing.post_data({"confirm": True}),
            }
            self.assertHttpStatus(self.client.post(**request), 404)
            self.assertTrue(self._get_queryset().filter(pk=instance3.pk).exists())

    class ListObjectsViewTestCase(ModelViewTestCase):
        """
        Retrieve multiple instances.
        """

        filterset = None

        def get_filterset(self):
            return self.filterset or lookup.get_filterset_for_model(self.model)

        # Helper methods to be overriden by special cases.
        # See ConsoleConnectionsTestCase, InterfaceConnectionsTestCase and PowerConnectionsTestCase
        def get_list_url(self):
            return reverse(helpers.validated_viewname(self.model, "list"))

        def get_title(self):
            return helpers.bettertitle(self.model._meta.verbose_name_plural)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_list_objects_anonymous(self):
            # Make the request as an unauthenticated user
            self.client.logout()
            response = self.client.get(self._get_url("list"))
            self.assertHttpStatus(response, 200)
            response_body = response.content.decode(response.charset)
            self.assertIn("/login/?next=" + self._get_url("list"), response_body, msg=response_body)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_list_objects_filtered(self):
            instance1, instance2 = self._get_queryset().all()[:2]
            response = self.client.get(f"{self._get_url('list')}?id={instance1.pk}")
            self.assertHttpStatus(response, 200)
            content = testing.extract_page_body(response.content.decode(response.charset))
            # TODO: it'd make test failures more readable if we strip the page headers/footers from the content
            if hasattr(self.model, "name"):
                self.assertRegex(content, r">\s*" + re.escape(escape(instance1.name)) + r"\s*<", msg=content)
                self.assertNotRegex(content, r">\s*" + re.escape(escape(instance2.name)) + r"\s*<", msg=content)
            if instance1.get_absolute_url() in content:
                self.assertNotIn(instance2.get_absolute_url(), content, msg=content)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"], STRICT_FILTERING=True)
        def test_list_objects_unknown_filter_strict_filtering(self):
            """Verify that with STRICT_FILTERING, an unknown filter results in an error message and no matches."""
            response = self.client.get(f"{self._get_url('list')}?ice_cream_flavor=chocolate")
            self.assertHttpStatus(response, 200)
            content = testing.extract_page_body(response.content.decode(response.charset))
            # TODO: it'd make test failures more readable if we strip the page headers/footers from the content
            self.assertIn("Unknown filter field", content, msg=content)
            # There should be no table rows displayed except for the empty results row
            self.assertIn(f"No {self.model._meta.verbose_name_plural} found", content, msg=content)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"], STRICT_FILTERING=False)
        def test_list_objects_unknown_filter_no_strict_filtering(self):
            """Verify that without STRICT_FILTERING, an unknown filter is ignored."""
            instance1, instance2 = self._get_queryset().all()[:2]
            with self.assertLogs("nautobot.core.filters") as cm:
                response = self.client.get(f"{self._get_url('list')}?ice_cream_flavor=chocolate")
            filterset = self.get_filterset()
            if not filterset:
                self.fail(
                    f"Couldn't find filterset for model {self.model}. The FilterSet class is expected to be in the "
                    "filters module within the application associated with the model and its name is expected to be "
                    f"{self.model.__name__}FilterSet."
                )
            self.assertEqual(
                cm.output,
                [
                    f'WARNING:nautobot.core.filters:{filterset.__name__}: Unknown filter field "ice_cream_flavor"',
                ],
            )
            self.assertHttpStatus(response, 200)
            content = testing.extract_page_body(response.content.decode(response.charset))
            # TODO: it'd make test failures more readable if we strip the page headers/footers from the content
            self.assertNotIn("Unknown filter field", content, msg=content)
            self.assertNotIn(f"No {self.model._meta.verbose_name_plural} found", content, msg=content)
            if hasattr(self.model, "name"):
                self.assertRegex(content, r">\s*" + re.escape(instance1.name) + r"\s*<", msg=content)
                self.assertRegex(content, r">\s*" + re.escape(instance2.name) + r"\s*<", msg=content)
            if instance1.get_absolute_url() in content:
                self.assertIn(instance2.get_absolute_url(), content, msg=content)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects_without_permission(self):
            # Try GET without permission
            with testing.disable_warnings("django.request"):
                response = self.client.get(self._get_url("list"))
                self.assertHttpStatus(response, 403)
                response_body = response.content.decode(response.charset)
                self.assertNotIn("/login/", response_body, msg=response_body)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects_with_permission(self):
            # Add model-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view"])
            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
            response = self.client.get(self._get_url("list"))
            self.assertHttpStatus(response, 200)
            response_body = response.content.decode(response.charset)

            list_url = self.get_list_url()
            title = self.get_title()

            # Check if breadcrumb is rendered correctly
            self.assertIn(
                f'<a href="{list_url}">{title}</a>',
                response_body,
            )

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_objects_with_constrained_permission(self):
            instance1, instance2 = self._get_queryset().all()[:2]

            # Add object-level permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"pk": instance1.pk},
                actions=["view"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try GET with object-level permission
            response = self.client.get(self._get_url("list"))
            self.assertHttpStatus(response, 200)
            content = testing.extract_page_body(response.content.decode(response.charset))
            # TODO: it'd make test failures more readable if we strip the page headers/footers from the content
            if hasattr(self.model, "name"):
                self.assertRegex(content, r">\s*" + re.escape(instance1.name) + r"\s*<", msg=content)
                self.assertNotRegex(content, r">\s*" + re.escape(instance2.name) + r"\s*<", msg=content)
            elif hasattr(self.model, "get_absolute_url"):
                self.assertIn(instance1.get_absolute_url(), content, msg=content)
                self.assertNotIn(instance2.get_absolute_url(), content, msg=content)

        @skipIf(
            "example_plugin" not in settings.PLUGINS,
            "example_plugin not in settings.PLUGINS",
        )
        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_list_view_plugin_banner(self):
            """
            If example plugin is installed, check if the plugin banner is rendered correctly in ObjectListView.
            """
            # Add model-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view"])
            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
            response = self.client.get(self._get_url("list"))
            self.assertHttpStatus(response, 200)
            response_body = response.content.decode(response.charset)

            # Check plugin banner is rendered correctly
            self.assertIn(
                f"<div>You are viewing a table of {self.model._meta.verbose_name_plural}</div>", response_body
            )

    class CreateMultipleObjectsViewTestCase(ModelViewTestCase):
        """
        Create multiple instances using a single form. Expects the creation of three new instances by default.

        :bulk_create_count: The number of objects expected to be created (default: 3).
        :bulk_create_data: A dictionary of data to be used for bulk object creation.
        """

        bulk_create_count = 3
        bulk_create_data = {}

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_create_multiple_objects_without_permission(self):
            request = {
                "path": self._get_url("add"),
                "data": testing.post_data(self.bulk_create_data),
            }

            # Try POST without permission
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.post(**request), 403)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_create_multiple_objects_with_permission(self):
            initial_count = self._get_queryset().count()
            request = {
                "path": self._get_url("add"),
                "data": testing.post_data(self.bulk_create_data),
            }

            # Assign non-constrained permission
            obj_perm = users_models.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))

            # Bulk create objects
            response = self.client.post(**request)
            self.assertHttpStatus(response, 302)
            self.assertEqual(initial_count + self.bulk_create_count, self._get_queryset().count())
            matching_count = 0
            for instance in self._get_queryset().all():
                try:
                    self.assertInstanceEqual(instance, self.bulk_create_data)
                    matching_count += 1
                except AssertionError:
                    pass
            self.assertEqual(matching_count, self.bulk_create_count)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_create_multiple_objects_with_constrained_permission(self):
            initial_count = self._get_queryset().count()
            request = {
                "path": self._get_url("add"),
                "data": testing.post_data(self.bulk_create_data),
            }

            # Assign constrained permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                actions=["add"],
                constraints={"pk": uuid.uuid4()},  # Match a non-existent pk (i.e., deny all)
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Attempt to make the request with unmet constraints
            self.assertHttpStatus(self.client.post(**request), 200)
            self.assertEqual(self._get_queryset().count(), initial_count)

            # Update the ObjectPermission to allow creation
            obj_perm.constraints = {"pk__isnull": False}  # Set constraint to allow all
            obj_perm.save()

            response = self.client.post(**request)
            self.assertHttpStatus(response, 302)
            self.assertEqual(initial_count + self.bulk_create_count, self._get_queryset().count())

            matching_count = 0
            for instance in self._get_queryset().all():
                try:
                    self.assertInstanceEqual(instance, self.bulk_create_data)
                    matching_count += 1
                except AssertionError:
                    pass
            self.assertEqual(matching_count, self.bulk_create_count)

    class BulkImportObjectsViewTestCase(ModelViewTestCase):
        """
        Create multiple instances from imported data.

        Note that CSV import, since it's now implemented via the REST API,
        is also exercised by APIViewTestCases.CreateObjectViewTestCase.test_recreate_object_csv().

        :csv_data: A list of CSV-formatted lines (starting with the headers) to be used for bulk object import.
        """

        csv_data = ()

        def _get_csv_data(self):
            return "\n".join(self.csv_data)

        def test_bulk_import_objects_without_permission(self):
            data = {
                "csv_data": self._get_csv_data(),
            }

            # Test GET without permission
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.get(self._get_url("import")), 403)

            # Try POST without permission
            response = self.client.post(self._get_url("import"), data)
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(response, 403)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_bulk_import_objects_with_permission(self):
            initial_count = self._get_queryset().count()
            data = {
                "csv_data": self._get_csv_data(),
            }

            # Assign model-level permission
            obj_perm = users_models.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(self._get_queryset().count(), initial_count + len(self.csv_data) - 1)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_bulk_import_objects_with_permission_csv_file(self):
            initial_count = self._get_queryset().count()
            self.file_contents = bytes(self._get_csv_data(), "utf-8")
            self.bulk_import_file = SimpleUploadedFile(name="bulk_import_data.csv", content=self.file_contents)
            data = {
                "csv_file": self.bulk_import_file,
            }

            # Assign model-level permission
            obj_perm = users_models.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
            response = self.client.post(self._get_url("import"), data)
            self.assertHttpStatus(response, 200)
            self.assertEqual(
                self._get_queryset().count(),
                initial_count + len(self.csv_data) - 1,
                testing.extract_page_body(response.content.decode(response.charset)),
            )

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_bulk_import_objects_with_constrained_permission(self):
            initial_count = self._get_queryset().count()
            data = {
                "csv_data": self._get_csv_data(),
            }

            # Assign constrained permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"pk": str(uuid.uuid4())},  # Match a non-existent pk (i.e., deny all)
                actions=["add"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Attempt to import non-permitted objects
            self.assertHttpStatus(self.client.post(self._get_url("import"), data), 200)
            self.assertEqual(self._get_queryset().count(), initial_count)

            # Update permission constraints
            obj_perm.constraints = {"pk__isnull": False}  # Set permission to allow all
            obj_perm.save()

            # Import permitted objects
            self.assertHttpStatus(self.client.post(self._get_url("import"), data), 200)
            self.assertEqual(self._get_queryset().count(), initial_count + len(self.csv_data) - 1)

    class BulkEditObjectsViewTestCase(ModelViewTestCase):
        """
        Edit multiple instances.

        :bulk_edit_data: A dictionary of data to be used when bulk editing a set of objects. This data should differ
                         from that used for initial object creation within setUpTestData().
        """

        bulk_edit_data = {}

        def test_bulk_edit_objects_without_permission(self):
            pk_list = list(self._get_queryset().values_list("pk", flat=True)[:3])
            data = {
                "pk": pk_list,
                "_apply": True,  # Form button
            }

            # Try POST without permission
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.post(self._get_url("bulk_edit"), data), 403)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_bulk_edit_objects_with_permission(self):
            pk_list = list(self._get_queryset().values_list("pk", flat=True)[:3])
            data = {
                "pk": pk_list,
                "_apply": True,  # Form button
            }

            # Append the form data to the request
            data.update(testing.post_data(self.bulk_edit_data))

            # Assign model-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try POST with model-level permission
            self.assertHttpStatus(self.client.post(self._get_url("bulk_edit"), data), 302)
            for instance in self._get_queryset().filter(pk__in=pk_list):
                self.assertInstanceEqual(instance, self.bulk_edit_data)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_bulk_edit_form_contains_all_pks(self):
            # We are testing the intermediary step of bulk_edit with pagination applied.
            # i.e. "_all" passed in the form.
            pk_list = self._get_queryset().values_list("pk", flat=True)
            # We only pass in one pk to test the functionality of "_all"
            # which should grab all instance pks regardless of "pk"
            selected_data = {
                "pk": pk_list[:1],
                "_all": "on",
            }
            # Assign model-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try POST with model-level permission
            response = self.client.post(self._get_url("bulk_edit"), selected_data)
            # Expect a 200 status cause we are only rendering the bulk edit table.
            # after pressing Edit Selected button.
            self.assertHttpStatus(response, 200)
            response_body = testing.extract_page_body(response.content.decode(response.charset))
            # Check if all the pks are passed into the BulkEditForm/BulkUpdateForm
            for pk in pk_list:
                self.assertIn(f'<input type="hidden" name="pk" value="{pk}"', response_body)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_bulk_edit_objects_with_constrained_permission(self):
            # Select some objects that are *not* already set to match the first value in self.bulk_edit_data or null.
            # We have to exclude null cases because Django filter()/exclude() doesn't like `__in=[None]` as a case.
            attr_name = list(self.bulk_edit_data.keys())[0]
            objects = (
                self._get_queryset()
                .exclude(**{attr_name: self.bulk_edit_data[attr_name]})
                .exclude(**{f"{attr_name}__isnull": True})
            )[:3]
            self.assertEqual(objects.count(), 3)
            pk_list = list(objects.values_list("pk", flat=True))

            # Define a permission that permits the above objects, but will not permit them after updating them.
            field = self.model._meta.get_field(attr_name)
            values = [field.value_from_object(obj) for obj in objects]

            # Assign constrained permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={f"{attr_name}__in": values},
                actions=["change"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Build form data
            data = {
                "pk": pk_list,
                "_apply": True,  # Form button
            }
            data.update(testing.post_data(self.bulk_edit_data))

            # Attempt to bulk edit permitted objects into a non-permitted state
            response = self.client.post(self._get_url("bulk_edit"), data)
            # 200 because we're sent back to the edit form to try again; if the update were successful it'd be a 302
            self.assertHttpStatus(response, 200)
            # Assert that the objects are NOT updated
            for instance in self._get_queryset().filter(pk__in=pk_list):
                self.assertIn(field.value_from_object(instance), values)
                self.assertNotEqual(field.value_from_object(instance), self.bulk_edit_data[attr_name])

            # Update permission constraints to permit all objects
            obj_perm.constraints = {"pk__gt": 0}
            obj_perm.save()

            # Bulk edit permitted objects and expect a redirect back to the list view
            self.assertHttpStatus(self.client.post(self._get_url("bulk_edit"), data), 302)
            # Assert that the objects were all updated correctly
            for instance in self._get_queryset().filter(pk__in=pk_list):
                self.assertInstanceEqual(instance, self.bulk_edit_data)

    class BulkDeleteObjectsViewTestCase(ModelViewTestCase):
        """
        Delete multiple instances.
        """

        def get_deletable_object_pks(self):
            """
            Get a list of PKs corresponding to objects that can be safely bulk-deleted.

            For some models this may just be any random objects, but when we have FKs with `on_delete=models.PROTECT`
            (as is often the case) we need to find or create an instance that doesn't have such entanglements.
            """
            return testing.get_deletable_objects(self.model, self._get_queryset()).values_list("pk", flat=True)[:3]

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_bulk_delete_objects_without_permission(self):
            pk_list = self.get_deletable_object_pks()
            data = {
                "pk": pk_list,
                "confirm": True,
                "_confirm": True,  # Form button
            }

            # Try POST without permission
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.post(self._get_url("bulk_delete"), data), 403)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_bulk_delete_objects_with_permission(self):
            pk_list = self.get_deletable_object_pks()
            initial_count = self._get_queryset().count()
            data = {
                "pk": pk_list,
                "confirm": True,
                "_confirm": True,  # Form button
            }

            # Assign unconstrained permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try POST with model-level permission
            self.assertHttpStatus(self.client.post(self._get_url("bulk_delete"), data), 302)
            self.assertEqual(self._get_queryset().count(), initial_count - len(pk_list))

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_bulk_delete_form_contains_all_pks(self):
            # We are testing the intermediary step of bulk_delete with pagination applied.
            # i.e. "_all" passed in the form.
            pk_list = self._get_queryset().values_list("pk", flat=True)
            # We only pass in one pk to test the functionality of "_all"
            # which should grab all instance pks regardless of "pks".
            selected_data = {
                "pk": pk_list[:1],
                "confirm": True,
                "_all": "on",
            }

            # Assign unconstrained permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try POST with the selected data first. Emulating selecting all -> pressing Delete Selected button.
            response = self.client.post(self._get_url("bulk_delete"), selected_data)
            self.assertHttpStatus(response, 200)
            response_body = testing.extract_page_body(response.content.decode(response.charset))
            # Check if all the pks are passed into the BulkDeleteForm/BulkDestroyForm
            for pk in pk_list:
                self.assertIn(f'<input type="hidden" name="pk" value="{pk}"', response_body)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
        def test_bulk_delete_objects_with_constrained_permission(self):
            pk_list = self.get_deletable_object_pks()
            initial_count = self._get_queryset().count()
            data = {
                "pk": pk_list,
                "confirm": True,
                "_confirm": True,  # Form button
            }

            # Assign constrained permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"pk": str(uuid.uuid4())},  # Match a non-existent pk (i.e., deny all)
                actions=["delete"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Attempt to bulk delete non-permitted objects
            self.assertHttpStatus(self.client.post(self._get_url("bulk_delete"), data), 302)
            self.assertEqual(self._get_queryset().count(), initial_count)

            # Update permission constraints
            obj_perm.constraints = {"pk__isnull": False}  # Match a non-existent pk (i.e., allow all)
            obj_perm.save()

            # Bulk delete permitted objects
            self.assertHttpStatus(self.client.post(self._get_url("bulk_delete"), data), 302)
            self.assertEqual(self._get_queryset().count(), initial_count - len(pk_list))

    class BulkRenameObjectsViewTestCase(ModelViewTestCase):
        """
        Rename multiple instances.
        """

        rename_data = {
            "find": "^(.*)$",
            "replace": "\\1X",  # Append an X to the original value
            "use_regex": True,
        }

        def test_bulk_rename_objects_without_permission(self):
            pk_list = list(self._get_queryset().values_list("pk", flat=True)[:3])
            data = {
                "pk": pk_list,
                "_apply": True,  # Form button
            }
            data.update(self.rename_data)

            # Test GET without permission
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.get(self._get_url("bulk_rename")), 403)

            # Try POST without permission
            with testing.disable_warnings("django.request"):
                self.assertHttpStatus(self.client.post(self._get_url("bulk_rename"), data), 403)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_bulk_rename_objects_with_permission(self):
            objects = list(self._get_queryset().all()[:3])
            pk_list = [obj.pk for obj in objects]
            data = {
                "pk": pk_list,
                "_apply": True,  # Form button
            }
            data.update(self.rename_data)

            # Assign model-level permission
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Try POST with model-level permission
            self.assertHttpStatus(self.client.post(self._get_url("bulk_rename"), data), 302)
            for i, instance in enumerate(self._get_queryset().filter(pk__in=pk_list)):
                self.assertEqual(instance.name, f"{objects[i].name}X")

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_bulk_rename_objects_with_constrained_permission(self):
            objects = list(self._get_queryset().all()[:3])
            pk_list = [obj.pk for obj in objects]
            data = {
                "pk": pk_list,
                "_apply": True,  # Form button
            }
            data.update(self.rename_data)

            # Assign constrained permission
            obj_perm = users_models.ObjectPermission(
                name="Test permission",
                constraints={"name__regex": "[^X]$"},
                actions=["change"],
            )
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            # Attempt to bulk edit permitted objects into a non-permitted state
            response = self.client.post(self._get_url("bulk_rename"), data)
            self.assertHttpStatus(response, 200)

            # Update permission constraints
            obj_perm.constraints = {"pk__gt": 0}
            obj_perm.save()

            # Bulk rename permitted objects
            self.assertHttpStatus(self.client.post(self._get_url("bulk_rename"), data), 302)
            for i, instance in enumerate(self._get_queryset().filter(pk__in=pk_list)):
                self.assertEqual(instance.name, f"{objects[i].name}X")

    class PrimaryObjectViewTestCase(
        GetObjectViewTestCase,
        GetObjectChangelogViewTestCase,
        GetObjectNotesViewTestCase,
        CreateObjectViewTestCase,
        EditObjectViewTestCase,
        DeleteObjectViewTestCase,
        ListObjectsViewTestCase,
        BulkImportObjectsViewTestCase,
        BulkEditObjectsViewTestCase,
        BulkDeleteObjectsViewTestCase,
    ):
        """
        TestCase suitable for testing all standard View functions for primary objects
        """

        maxDiff = None

    class OrganizationalObjectViewTestCase(
        GetObjectViewTestCase,
        GetObjectChangelogViewTestCase,
        GetObjectNotesViewTestCase,
        CreateObjectViewTestCase,
        EditObjectViewTestCase,
        DeleteObjectViewTestCase,
        ListObjectsViewTestCase,
        BulkImportObjectsViewTestCase,
        BulkDeleteObjectsViewTestCase,
    ):
        """
        TestCase suitable for all organizational objects
        """

        maxDiff = None

    class DeviceComponentTemplateViewTestCase(
        EditObjectViewTestCase,
        DeleteObjectViewTestCase,
        CreateMultipleObjectsViewTestCase,
        BulkEditObjectsViewTestCase,
        BulkRenameObjectsViewTestCase,
        BulkDeleteObjectsViewTestCase,
    ):
        """
        TestCase suitable for testing device component template models (ConsolePortTemplates, InterfaceTemplates, etc.)
        """

        maxDiff = None

    class DeviceComponentViewTestCase(
        GetObjectViewTestCase,
        GetObjectChangelogViewTestCase,
        GetObjectNotesViewTestCase,
        EditObjectViewTestCase,
        DeleteObjectViewTestCase,
        ListObjectsViewTestCase,
        CreateMultipleObjectsViewTestCase,
        BulkImportObjectsViewTestCase,
        BulkEditObjectsViewTestCase,
        BulkRenameObjectsViewTestCase,
        BulkDeleteObjectsViewTestCase,
    ):
        """
        TestCase suitable for testing device component models (ConsolePorts, Interfaces, etc.)
        """

        maxDiff = None
        bulk_add_data = None
        """Used for bulk-add (distinct from bulk-create) view testing; self.bulk_create_data will be used if unset."""

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_bulk_add_component(self):
            """Test bulk-adding this component to devices/virtual-machines."""
            obj_perm = users_models.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))

            initial_count = self._get_queryset().count()

            data = (self.bulk_add_data or self.bulk_create_data).copy()

            # Load the device-bulk-add or virtualmachine-bulk-add form
            if "device" in data:
                url = reverse(f"dcim:device_bulk_add_{self.model._meta.model_name}")
                request = {
                    "path": url,
                    "data": testing.post_data({"pk": data["device"]}),
                }
            else:
                url = reverse(f"virtualization:virtualmachine_bulk_add_{self.model._meta.model_name}")
                request = {
                    "path": url,
                    "data": testing.post_data({"pk": data["virtual_machine"]}),
                }
            self.assertHttpStatus(self.client.post(**request), 200)

            # Post to the device-bulk-add or virtualmachine-bulk-add form to create records
            if "device" in data:
                data["pk"] = data.pop("device")
            else:
                data["pk"] = data.pop("virtual_machine")
            data["_create"] = ""
            request["data"] = testing.post_data(data)
            self.assertHttpStatus(self.client.post(**request), 302)

            updated_count = self._get_queryset().count()
            self.assertEqual(updated_count, initial_count + self.bulk_create_count)

            matching_count = 0
            for instance in self._get_queryset().all():
                try:
                    self.assertInstanceEqual(instance, (self.bulk_add_data or self.bulk_create_data))
                    matching_count += 1
                except AssertionError:
                    pass
            self.assertEqual(matching_count, self.bulk_create_count)

        @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
        def test_bulk_rename(self):
            obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
            obj_perm.save()
            obj_perm.users.add(self.user)
            obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

            objects = self.selected_objects
            pk_list = [obj.pk for obj in objects]
            # Apply button not yet clicked
            data = {"pk": pk_list}
            data.update(self.rename_data)
            verbose_name_plural = self.model._meta.verbose_name_plural

            with self.subTest("Assert device name in HTML"):
                response = self.client.post(self._get_url("bulk_rename"), data)
                message = (
                    f"Renaming {len(objects)} {helpers.bettertitle(verbose_name_plural)} "
                    f"on {self.selected_objects_parent_name}"
                )
                self.assertInHTML(message, response.content.decode(response.charset))

            with self.subTest("Assert update successfully"):
                data["_apply"] = True  # Form Apply button
                response = self.client.post(self._get_url("bulk_rename"), data)
                self.assertHttpStatus(response, 302)
                queryset = self._get_queryset().filter(pk__in=pk_list)
                for instance in objects:
                    self.assertEqual(queryset.get(pk=instance.pk).name, f"{instance.name}X")

            with self.subTest("Assert if no valid objects selected return with error"):
                for values in ([], [str(uuid.uuid4())]):
                    data["pk"] = values
                    response = self.client.post(self._get_url("bulk_rename"), data, follow=True)
                    expected_message = f"No valid {verbose_name_plural} were selected."
                    self.assertIn(expected_message, response.content.decode(response.charset))

BulkDeleteObjectsViewTestCase

Bases: ModelViewTestCase

Delete multiple instances.

Source code in nautobot/core/testing/views.py

class BulkDeleteObjectsViewTestCase(ModelViewTestCase):
    """
    Delete multiple instances.
    """

    def get_deletable_object_pks(self):
        """
        Get a list of PKs corresponding to objects that can be safely bulk-deleted.

        For some models this may just be any random objects, but when we have FKs with `on_delete=models.PROTECT`
        (as is often the case) we need to find or create an instance that doesn't have such entanglements.
        """
        return testing.get_deletable_objects(self.model, self._get_queryset()).values_list("pk", flat=True)[:3]

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_bulk_delete_objects_without_permission(self):
        pk_list = self.get_deletable_object_pks()
        data = {
            "pk": pk_list,
            "confirm": True,
            "_confirm": True,  # Form button
        }

        # Try POST without permission
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.post(self._get_url("bulk_delete"), data), 403)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_bulk_delete_objects_with_permission(self):
        pk_list = self.get_deletable_object_pks()
        initial_count = self._get_queryset().count()
        data = {
            "pk": pk_list,
            "confirm": True,
            "_confirm": True,  # Form button
        }

        # Assign unconstrained permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try POST with model-level permission
        self.assertHttpStatus(self.client.post(self._get_url("bulk_delete"), data), 302)
        self.assertEqual(self._get_queryset().count(), initial_count - len(pk_list))

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_bulk_delete_form_contains_all_pks(self):
        # We are testing the intermediary step of bulk_delete with pagination applied.
        # i.e. "_all" passed in the form.
        pk_list = self._get_queryset().values_list("pk", flat=True)
        # We only pass in one pk to test the functionality of "_all"
        # which should grab all instance pks regardless of "pks".
        selected_data = {
            "pk": pk_list[:1],
            "confirm": True,
            "_all": "on",
        }

        # Assign unconstrained permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try POST with the selected data first. Emulating selecting all -> pressing Delete Selected button.
        response = self.client.post(self._get_url("bulk_delete"), selected_data)
        self.assertHttpStatus(response, 200)
        response_body = testing.extract_page_body(response.content.decode(response.charset))
        # Check if all the pks are passed into the BulkDeleteForm/BulkDestroyForm
        for pk in pk_list:
            self.assertIn(f'<input type="hidden" name="pk" value="{pk}"', response_body)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_bulk_delete_objects_with_constrained_permission(self):
        pk_list = self.get_deletable_object_pks()
        initial_count = self._get_queryset().count()
        data = {
            "pk": pk_list,
            "confirm": True,
            "_confirm": True,  # Form button
        }

        # Assign constrained permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={"pk": str(uuid.uuid4())},  # Match a non-existent pk (i.e., deny all)
            actions=["delete"],
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Attempt to bulk delete non-permitted objects
        self.assertHttpStatus(self.client.post(self._get_url("bulk_delete"), data), 302)
        self.assertEqual(self._get_queryset().count(), initial_count)

        # Update permission constraints
        obj_perm.constraints = {"pk__isnull": False}  # Match a non-existent pk (i.e., allow all)
        obj_perm.save()

        # Bulk delete permitted objects
        self.assertHttpStatus(self.client.post(self._get_url("bulk_delete"), data), 302)
        self.assertEqual(self._get_queryset().count(), initial_count - len(pk_list))

get_deletable_object_pks()

Get a list of PKs corresponding to objects that can be safely bulk-deleted.

For some models this may just be any random objects, but when we have FKs with on_delete=models.PROTECT (as is often the case) we need to find or create an instance that doesn't have such entanglements.

Source code in nautobot/core/testing/views.py

def get_deletable_object_pks(self):
    """
    Get a list of PKs corresponding to objects that can be safely bulk-deleted.

    For some models this may just be any random objects, but when we have FKs with `on_delete=models.PROTECT`
    (as is often the case) we need to find or create an instance that doesn't have such entanglements.
    """
    return testing.get_deletable_objects(self.model, self._get_queryset()).values_list("pk", flat=True)[:3]

BulkEditObjectsViewTestCase

Bases: ModelViewTestCase

Edit multiple instances.

:bulk_edit_data: A dictionary of data to be used when bulk editing a set of objects. This data should differ from that used for initial object creation within setUpTestData().

Source code in nautobot/core/testing/views.py

class BulkEditObjectsViewTestCase(ModelViewTestCase):
    """
    Edit multiple instances.

    :bulk_edit_data: A dictionary of data to be used when bulk editing a set of objects. This data should differ
                     from that used for initial object creation within setUpTestData().
    """

    bulk_edit_data = {}

    def test_bulk_edit_objects_without_permission(self):
        pk_list = list(self._get_queryset().values_list("pk", flat=True)[:3])
        data = {
            "pk": pk_list,
            "_apply": True,  # Form button
        }

        # Try POST without permission
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.post(self._get_url("bulk_edit"), data), 403)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_bulk_edit_objects_with_permission(self):
        pk_list = list(self._get_queryset().values_list("pk", flat=True)[:3])
        data = {
            "pk": pk_list,
            "_apply": True,  # Form button
        }

        # Append the form data to the request
        data.update(testing.post_data(self.bulk_edit_data))

        # Assign model-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try POST with model-level permission
        self.assertHttpStatus(self.client.post(self._get_url("bulk_edit"), data), 302)
        for instance in self._get_queryset().filter(pk__in=pk_list):
            self.assertInstanceEqual(instance, self.bulk_edit_data)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_bulk_edit_form_contains_all_pks(self):
        # We are testing the intermediary step of bulk_edit with pagination applied.
        # i.e. "_all" passed in the form.
        pk_list = self._get_queryset().values_list("pk", flat=True)
        # We only pass in one pk to test the functionality of "_all"
        # which should grab all instance pks regardless of "pk"
        selected_data = {
            "pk": pk_list[:1],
            "_all": "on",
        }
        # Assign model-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try POST with model-level permission
        response = self.client.post(self._get_url("bulk_edit"), selected_data)
        # Expect a 200 status cause we are only rendering the bulk edit table.
        # after pressing Edit Selected button.
        self.assertHttpStatus(response, 200)
        response_body = testing.extract_page_body(response.content.decode(response.charset))
        # Check if all the pks are passed into the BulkEditForm/BulkUpdateForm
        for pk in pk_list:
            self.assertIn(f'<input type="hidden" name="pk" value="{pk}"', response_body)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_bulk_edit_objects_with_constrained_permission(self):
        # Select some objects that are *not* already set to match the first value in self.bulk_edit_data or null.
        # We have to exclude null cases because Django filter()/exclude() doesn't like `__in=[None]` as a case.
        attr_name = list(self.bulk_edit_data.keys())[0]
        objects = (
            self._get_queryset()
            .exclude(**{attr_name: self.bulk_edit_data[attr_name]})
            .exclude(**{f"{attr_name}__isnull": True})
        )[:3]
        self.assertEqual(objects.count(), 3)
        pk_list = list(objects.values_list("pk", flat=True))

        # Define a permission that permits the above objects, but will not permit them after updating them.
        field = self.model._meta.get_field(attr_name)
        values = [field.value_from_object(obj) for obj in objects]

        # Assign constrained permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={f"{attr_name}__in": values},
            actions=["change"],
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Build form data
        data = {
            "pk": pk_list,
            "_apply": True,  # Form button
        }
        data.update(testing.post_data(self.bulk_edit_data))

        # Attempt to bulk edit permitted objects into a non-permitted state
        response = self.client.post(self._get_url("bulk_edit"), data)
        # 200 because we're sent back to the edit form to try again; if the update were successful it'd be a 302
        self.assertHttpStatus(response, 200)
        # Assert that the objects are NOT updated
        for instance in self._get_queryset().filter(pk__in=pk_list):
            self.assertIn(field.value_from_object(instance), values)
            self.assertNotEqual(field.value_from_object(instance), self.bulk_edit_data[attr_name])

        # Update permission constraints to permit all objects
        obj_perm.constraints = {"pk__gt": 0}
        obj_perm.save()

        # Bulk edit permitted objects and expect a redirect back to the list view
        self.assertHttpStatus(self.client.post(self._get_url("bulk_edit"), data), 302)
        # Assert that the objects were all updated correctly
        for instance in self._get_queryset().filter(pk__in=pk_list):
            self.assertInstanceEqual(instance, self.bulk_edit_data)

BulkImportObjectsViewTestCase

Bases: ModelViewTestCase

Create multiple instances from imported data.

Note that CSV import, since it's now implemented via the REST API, is also exercised by APIViewTestCases.CreateObjectViewTestCase.test_recreate_object_csv().

:csv_data: A list of CSV-formatted lines (starting with the headers) to be used for bulk object import.

Source code in nautobot/core/testing/views.py

class BulkImportObjectsViewTestCase(ModelViewTestCase):
    """
    Create multiple instances from imported data.

    Note that CSV import, since it's now implemented via the REST API,
    is also exercised by APIViewTestCases.CreateObjectViewTestCase.test_recreate_object_csv().

    :csv_data: A list of CSV-formatted lines (starting with the headers) to be used for bulk object import.
    """

    csv_data = ()

    def _get_csv_data(self):
        return "\n".join(self.csv_data)

    def test_bulk_import_objects_without_permission(self):
        data = {
            "csv_data": self._get_csv_data(),
        }

        # Test GET without permission
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.get(self._get_url("import")), 403)

        # Try POST without permission
        response = self.client.post(self._get_url("import"), data)
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(response, 403)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_bulk_import_objects_with_permission(self):
        initial_count = self._get_queryset().count()
        data = {
            "csv_data": self._get_csv_data(),
        }

        # Assign model-level permission
        obj_perm = users_models.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(self._get_queryset().count(), initial_count + len(self.csv_data) - 1)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_bulk_import_objects_with_permission_csv_file(self):
        initial_count = self._get_queryset().count()
        self.file_contents = bytes(self._get_csv_data(), "utf-8")
        self.bulk_import_file = SimpleUploadedFile(name="bulk_import_data.csv", content=self.file_contents)
        data = {
            "csv_file": self.bulk_import_file,
        }

        # Assign model-level permission
        obj_perm = users_models.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
        response = self.client.post(self._get_url("import"), data)
        self.assertHttpStatus(response, 200)
        self.assertEqual(
            self._get_queryset().count(),
            initial_count + len(self.csv_data) - 1,
            testing.extract_page_body(response.content.decode(response.charset)),
        )

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_bulk_import_objects_with_constrained_permission(self):
        initial_count = self._get_queryset().count()
        data = {
            "csv_data": self._get_csv_data(),
        }

        # Assign constrained permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={"pk": str(uuid.uuid4())},  # Match a non-existent pk (i.e., deny all)
            actions=["add"],
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Attempt to import non-permitted objects
        self.assertHttpStatus(self.client.post(self._get_url("import"), data), 200)
        self.assertEqual(self._get_queryset().count(), initial_count)

        # Update permission constraints
        obj_perm.constraints = {"pk__isnull": False}  # Set permission to allow all
        obj_perm.save()

        # Import permitted objects
        self.assertHttpStatus(self.client.post(self._get_url("import"), data), 200)
        self.assertEqual(self._get_queryset().count(), initial_count + len(self.csv_data) - 1)

BulkRenameObjectsViewTestCase

Bases: ModelViewTestCase

Rename multiple instances.

Source code in nautobot/core/testing/views.py

class BulkRenameObjectsViewTestCase(ModelViewTestCase):
    """
    Rename multiple instances.
    """

    rename_data = {
        "find": "^(.*)$",
        "replace": "\\1X",  # Append an X to the original value
        "use_regex": True,
    }

    def test_bulk_rename_objects_without_permission(self):
        pk_list = list(self._get_queryset().values_list("pk", flat=True)[:3])
        data = {
            "pk": pk_list,
            "_apply": True,  # Form button
        }
        data.update(self.rename_data)

        # Test GET without permission
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.get(self._get_url("bulk_rename")), 403)

        # Try POST without permission
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.post(self._get_url("bulk_rename"), data), 403)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_bulk_rename_objects_with_permission(self):
        objects = list(self._get_queryset().all()[:3])
        pk_list = [obj.pk for obj in objects]
        data = {
            "pk": pk_list,
            "_apply": True,  # Form button
        }
        data.update(self.rename_data)

        # Assign model-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try POST with model-level permission
        self.assertHttpStatus(self.client.post(self._get_url("bulk_rename"), data), 302)
        for i, instance in enumerate(self._get_queryset().filter(pk__in=pk_list)):
            self.assertEqual(instance.name, f"{objects[i].name}X")

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_bulk_rename_objects_with_constrained_permission(self):
        objects = list(self._get_queryset().all()[:3])
        pk_list = [obj.pk for obj in objects]
        data = {
            "pk": pk_list,
            "_apply": True,  # Form button
        }
        data.update(self.rename_data)

        # Assign constrained permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={"name__regex": "[^X]$"},
            actions=["change"],
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Attempt to bulk edit permitted objects into a non-permitted state
        response = self.client.post(self._get_url("bulk_rename"), data)
        self.assertHttpStatus(response, 200)

        # Update permission constraints
        obj_perm.constraints = {"pk__gt": 0}
        obj_perm.save()

        # Bulk rename permitted objects
        self.assertHttpStatus(self.client.post(self._get_url("bulk_rename"), data), 302)
        for i, instance in enumerate(self._get_queryset().filter(pk__in=pk_list)):
            self.assertEqual(instance.name, f"{objects[i].name}X")

CreateMultipleObjectsViewTestCase

Bases: ModelViewTestCase

Create multiple instances using a single form. Expects the creation of three new instances by default.

:bulk_create_count: The number of objects expected to be created (default: 3). :bulk_create_data: A dictionary of data to be used for bulk object creation.

Source code in nautobot/core/testing/views.py

class CreateMultipleObjectsViewTestCase(ModelViewTestCase):
    """
    Create multiple instances using a single form. Expects the creation of three new instances by default.

    :bulk_create_count: The number of objects expected to be created (default: 3).
    :bulk_create_data: A dictionary of data to be used for bulk object creation.
    """

    bulk_create_count = 3
    bulk_create_data = {}

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_create_multiple_objects_without_permission(self):
        request = {
            "path": self._get_url("add"),
            "data": testing.post_data(self.bulk_create_data),
        }

        # Try POST without permission
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.post(**request), 403)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_create_multiple_objects_with_permission(self):
        initial_count = self._get_queryset().count()
        request = {
            "path": self._get_url("add"),
            "data": testing.post_data(self.bulk_create_data),
        }

        # Assign non-constrained permission
        obj_perm = users_models.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))

        # Bulk create objects
        response = self.client.post(**request)
        self.assertHttpStatus(response, 302)
        self.assertEqual(initial_count + self.bulk_create_count, self._get_queryset().count())
        matching_count = 0
        for instance in self._get_queryset().all():
            try:
                self.assertInstanceEqual(instance, self.bulk_create_data)
                matching_count += 1
            except AssertionError:
                pass
        self.assertEqual(matching_count, self.bulk_create_count)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_create_multiple_objects_with_constrained_permission(self):
        initial_count = self._get_queryset().count()
        request = {
            "path": self._get_url("add"),
            "data": testing.post_data(self.bulk_create_data),
        }

        # Assign constrained permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            actions=["add"],
            constraints={"pk": uuid.uuid4()},  # Match a non-existent pk (i.e., deny all)
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Attempt to make the request with unmet constraints
        self.assertHttpStatus(self.client.post(**request), 200)
        self.assertEqual(self._get_queryset().count(), initial_count)

        # Update the ObjectPermission to allow creation
        obj_perm.constraints = {"pk__isnull": False}  # Set constraint to allow all
        obj_perm.save()

        response = self.client.post(**request)
        self.assertHttpStatus(response, 302)
        self.assertEqual(initial_count + self.bulk_create_count, self._get_queryset().count())

        matching_count = 0
        for instance in self._get_queryset().all():
            try:
                self.assertInstanceEqual(instance, self.bulk_create_data)
                matching_count += 1
            except AssertionError:
                pass
        self.assertEqual(matching_count, self.bulk_create_count)

CreateObjectViewTestCase

Bases: ModelViewTestCase

Create a single new instance.

:form_data: Data to be used when creating a new object.

Source code in nautobot/core/testing/views.py

class CreateObjectViewTestCase(ModelViewTestCase):
    """
    Create a single new instance.

    :form_data: Data to be used when creating a new object.
    """

    form_data = {}
    slug_source = None
    slugify_function = staticmethod(slugify)
    slug_test_object = ""

    def test_create_object_without_permission(self):
        # Try GET without permission
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.get(self._get_url("add")), 403)

        # Try POST without permission
        request = {
            "path": self._get_url("add"),
            "data": testing.post_data(self.form_data),
        }
        response = self.client.post(**request)
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(response, 403)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_create_object_with_permission(self):
        initial_count = self._get_queryset().count()

        # Assign unconstrained permission
        obj_perm = users_models.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("add")), 200)

        # Try POST with model-level permission
        request = {
            "path": self._get_url("add"),
            "data": testing.post_data(self.form_data),
        }
        self.assertHttpStatus(self.client.post(**request), 302)
        self.assertEqual(initial_count + 1, self._get_queryset().count())
        # order_by() is no supported by django TreeNode,
        # So we directly retrieve the instance by "slug" or "name".
        if isinstance(self._get_queryset().first(), TreeNode):
            filter_by = self.slug_source if getattr(self, "slug_source", None) else "name"
            instance = self._get_queryset().get(**{filter_by: self.form_data.get(filter_by)})
            self.assertInstanceEqual(instance, self.form_data)
        else:
            if hasattr(self.model, "last_updated"):
                instance = self._get_queryset().order_by("last_updated").last()
                self.assertInstanceEqual(instance, self.form_data)
            else:
                instance = self._get_queryset().last()
                self.assertInstanceEqual(instance, self.form_data)

        if hasattr(self.model, "to_objectchange"):
            # Verify ObjectChange creation
            objectchanges = lookup.get_changes_for_model(instance)
            self.assertEqual(len(objectchanges), 1)
            self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_CREATE)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_create_object_with_constrained_permission(self):
        initial_count = self._get_queryset().count()

        # Assign constrained permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={"pk": str(uuid.uuid4())},  # Match a non-existent pk (i.e., deny all)
            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 object-level permission
        self.assertHttpStatus(self.client.get(self._get_url("add")), 200)

        # Try to create an object (not permitted)
        request = {
            "path": self._get_url("add"),
            "data": testing.post_data(self.form_data),
        }
        self.assertHttpStatus(self.client.post(**request), 200)
        self.assertEqual(initial_count, self._get_queryset().count())  # Check that no object was created

        # Update the ObjectPermission to allow creation
        obj_perm.constraints = {"pk__isnull": False}
        obj_perm.save()

        # Try to create an object (permitted)
        request = {
            "path": self._get_url("add"),
            "data": testing.post_data(self.form_data),
        }
        self.assertHttpStatus(self.client.post(**request), 302)
        self.assertEqual(initial_count + 1, self._get_queryset().count())
        # order_by() is no supported by django TreeNode,
        # So we directly retrieve the instance by "slug".
        if isinstance(self._get_queryset().first(), TreeNode):
            filter_by = self.slug_source if getattr(self, "slug_source", None) else "name"
            instance = self._get_queryset().get(**{filter_by: self.form_data.get(filter_by)})
            self.assertInstanceEqual(instance, self.form_data)
        else:
            if hasattr(self.model, "last_updated"):
                self.assertInstanceEqual(self._get_queryset().order_by("last_updated").last(), self.form_data)
            else:
                self.assertInstanceEqual(self._get_queryset().last(), self.form_data)

    def test_slug_autocreation(self):
        """Test that slug is autocreated through ORM."""
        # This really should go on a models test page, but we don't have test structures for models.
        if getattr(self.model, "slug_source", None) is not None:
            obj = self.model.objects.get(**{self.slug_source: self.slug_test_object})
            expected_slug = self.slugify_function(getattr(obj, self.slug_source))
            self.assertEqual(obj.slug, expected_slug)

    def test_slug_not_modified(self):
        """Ensure save method does not modify slug that is passed in."""
        # This really should go on a models test page, but we don't have test structures for models.
        if getattr(self.model, "slug_source", None) is not None:
            new_slug_source_value = "kwyjibo"

            obj = self.model.objects.get(**{self.slug_source: self.slug_test_object})
            expected_slug = self.slugify_function(getattr(obj, self.slug_source))
            # Update slug source field str
            filter_ = self.slug_source + "__exact"
            self.model.objects.filter(**{filter_: self.slug_test_object}).update(
                **{self.slug_source: new_slug_source_value}
            )

            obj.refresh_from_db()
            self.assertEqual(getattr(obj, self.slug_source), new_slug_source_value)
            self.assertEqual(obj.slug, expected_slug)

test_slug_autocreation()

Test that slug is autocreated through ORM.

Source code in nautobot/core/testing/views.py

def test_slug_autocreation(self):
    """Test that slug is autocreated through ORM."""
    # This really should go on a models test page, but we don't have test structures for models.
    if getattr(self.model, "slug_source", None) is not None:
        obj = self.model.objects.get(**{self.slug_source: self.slug_test_object})
        expected_slug = self.slugify_function(getattr(obj, self.slug_source))
        self.assertEqual(obj.slug, expected_slug)

test_slug_not_modified()

Ensure save method does not modify slug that is passed in.

Source code in nautobot/core/testing/views.py

def test_slug_not_modified(self):
    """Ensure save method does not modify slug that is passed in."""
    # This really should go on a models test page, but we don't have test structures for models.
    if getattr(self.model, "slug_source", None) is not None:
        new_slug_source_value = "kwyjibo"

        obj = self.model.objects.get(**{self.slug_source: self.slug_test_object})
        expected_slug = self.slugify_function(getattr(obj, self.slug_source))
        # Update slug source field str
        filter_ = self.slug_source + "__exact"
        self.model.objects.filter(**{filter_: self.slug_test_object}).update(
            **{self.slug_source: new_slug_source_value}
        )

        obj.refresh_from_db()
        self.assertEqual(getattr(obj, self.slug_source), new_slug_source_value)
        self.assertEqual(obj.slug, expected_slug)

DeleteObjectViewTestCase

Bases: ModelViewTestCase

Delete a single instance.

Source code in nautobot/core/testing/views.py

class DeleteObjectViewTestCase(ModelViewTestCase):
    """
    Delete a single instance.
    """

    def get_deletable_object(self):
        """
        Get an instance that can be deleted.

        For some models this may just be any random object, but when we have FKs with `on_delete=models.PROTECT`
        (as is often the case) we need to find or create an instance that doesn't have such entanglements.
        """
        instance = testing.get_deletable_objects(self.model, self._get_queryset()).first()
        if instance is None:
            self.fail("Couldn't find a single deletable object!")
        return instance

    def test_delete_object_without_permission(self):
        instance = self.get_deletable_object()

        # Try GET without permission
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.get(self._get_url("delete", instance)), [403, 404])

        # Try POST without permission
        request = {
            "path": self._get_url("delete", instance),
            "data": testing.post_data({"confirm": True}),
        }
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.post(**request), [403, 404])

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_delete_object_with_permission(self):
        instance = self.get_deletable_object()

        # Assign model-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
        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("delete", instance)), 200)

        # Try POST with model-level permission
        request = {
            "path": self._get_url("delete", instance),
            "data": testing.post_data({"confirm": True}),
        }
        self.assertHttpStatus(self.client.post(**request), 302)
        with self.assertRaises(ObjectDoesNotExist):
            self._get_queryset().get(pk=instance.pk)

        if hasattr(self.model, "to_objectchange"):
            # Verify ObjectChange creation
            objectchanges = lookup.get_changes_for_model(instance)
            self.assertEqual(len(objectchanges), 1)
            self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_DELETE)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_delete_object_with_permission_and_xwwwformurlencoded(self):
        instance = self.get_deletable_object()

        # Assign model-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["delete"])
        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("delete", instance)), 200)

        # Try POST with model-level permission
        request = {
            "path": self._get_url("delete", instance),
            "data": urlencode({"confirm": True}),
            "content_type": "application/x-www-form-urlencoded",
        }
        self.assertHttpStatus(self.client.post(**request), 302)
        with self.assertRaises(ObjectDoesNotExist):
            self._get_queryset().get(pk=instance.pk)

        if hasattr(self.model, "to_objectchange"):
            # Verify ObjectChange creation
            objectchanges = lookup.get_changes_for_model(instance)
            self.assertEqual(len(objectchanges), 1)
            self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_DELETE)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_delete_object_with_constrained_permission(self):
        instance1 = self.get_deletable_object()
        instance2 = self._get_queryset().exclude(pk=instance1.pk)[0]

        # Assign object-level permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={"pk": instance1.pk},
            actions=["delete"],
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try GET with a permitted object
        self.assertHttpStatus(self.client.get(self._get_url("delete", instance1)), 200)

        # Try GET with a non-permitted object
        self.assertHttpStatus(self.client.get(self._get_url("delete", instance2)), 404)

        # Try to delete a permitted object
        request = {
            "path": self._get_url("delete", instance1),
            "data": testing.post_data({"confirm": True}),
        }
        self.assertHttpStatus(self.client.post(**request), 302)
        with self.assertRaises(ObjectDoesNotExist):
            self._get_queryset().get(pk=instance1.pk)

        # Try to delete a non-permitted object
        # Note that in the case of tree models, deleting instance1 above may have cascade-deleted to instance2,
        # so to be safe, we need to get another object instance that definitely exists:
        instance3 = self._get_queryset().first()
        request = {
            "path": self._get_url("delete", instance3),
            "data": testing.post_data({"confirm": True}),
        }
        self.assertHttpStatus(self.client.post(**request), 404)
        self.assertTrue(self._get_queryset().filter(pk=instance3.pk).exists())

get_deletable_object()

Get an instance that can be deleted.

For some models this may just be any random object, but when we have FKs with on_delete=models.PROTECT (as is often the case) we need to find or create an instance that doesn't have such entanglements.

Source code in nautobot/core/testing/views.py

def get_deletable_object(self):
    """
    Get an instance that can be deleted.

    For some models this may just be any random object, but when we have FKs with `on_delete=models.PROTECT`
    (as is often the case) we need to find or create an instance that doesn't have such entanglements.
    """
    instance = testing.get_deletable_objects(self.model, self._get_queryset()).first()
    if instance is None:
        self.fail("Couldn't find a single deletable object!")
    return instance

DeviceComponentTemplateViewTestCase

Bases: EditObjectViewTestCase, DeleteObjectViewTestCase, CreateMultipleObjectsViewTestCase, BulkEditObjectsViewTestCase, BulkRenameObjectsViewTestCase, BulkDeleteObjectsViewTestCase

TestCase suitable for testing device component template models (ConsolePortTemplates, InterfaceTemplates, etc.)

Source code in nautobot/core/testing/views.py

class DeviceComponentTemplateViewTestCase(
    EditObjectViewTestCase,
    DeleteObjectViewTestCase,
    CreateMultipleObjectsViewTestCase,
    BulkEditObjectsViewTestCase,
    BulkRenameObjectsViewTestCase,
    BulkDeleteObjectsViewTestCase,
):
    """
    TestCase suitable for testing device component template models (ConsolePortTemplates, InterfaceTemplates, etc.)
    """

    maxDiff = None

DeviceComponentViewTestCase

Bases: GetObjectViewTestCase, GetObjectChangelogViewTestCase, GetObjectNotesViewTestCase, EditObjectViewTestCase, DeleteObjectViewTestCase, ListObjectsViewTestCase, CreateMultipleObjectsViewTestCase, BulkImportObjectsViewTestCase, BulkEditObjectsViewTestCase, BulkRenameObjectsViewTestCase, BulkDeleteObjectsViewTestCase

TestCase suitable for testing device component models (ConsolePorts, Interfaces, etc.)

Source code in nautobot/core/testing/views.py

class DeviceComponentViewTestCase(
    GetObjectViewTestCase,
    GetObjectChangelogViewTestCase,
    GetObjectNotesViewTestCase,
    EditObjectViewTestCase,
    DeleteObjectViewTestCase,
    ListObjectsViewTestCase,
    CreateMultipleObjectsViewTestCase,
    BulkImportObjectsViewTestCase,
    BulkEditObjectsViewTestCase,
    BulkRenameObjectsViewTestCase,
    BulkDeleteObjectsViewTestCase,
):
    """
    TestCase suitable for testing device component models (ConsolePorts, Interfaces, etc.)
    """

    maxDiff = None
    bulk_add_data = None
    """Used for bulk-add (distinct from bulk-create) view testing; self.bulk_create_data will be used if unset."""

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_bulk_add_component(self):
        """Test bulk-adding this component to devices/virtual-machines."""
        obj_perm = users_models.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))

        initial_count = self._get_queryset().count()

        data = (self.bulk_add_data or self.bulk_create_data).copy()

        # Load the device-bulk-add or virtualmachine-bulk-add form
        if "device" in data:
            url = reverse(f"dcim:device_bulk_add_{self.model._meta.model_name}")
            request = {
                "path": url,
                "data": testing.post_data({"pk": data["device"]}),
            }
        else:
            url = reverse(f"virtualization:virtualmachine_bulk_add_{self.model._meta.model_name}")
            request = {
                "path": url,
                "data": testing.post_data({"pk": data["virtual_machine"]}),
            }
        self.assertHttpStatus(self.client.post(**request), 200)

        # Post to the device-bulk-add or virtualmachine-bulk-add form to create records
        if "device" in data:
            data["pk"] = data.pop("device")
        else:
            data["pk"] = data.pop("virtual_machine")
        data["_create"] = ""
        request["data"] = testing.post_data(data)
        self.assertHttpStatus(self.client.post(**request), 302)

        updated_count = self._get_queryset().count()
        self.assertEqual(updated_count, initial_count + self.bulk_create_count)

        matching_count = 0
        for instance in self._get_queryset().all():
            try:
                self.assertInstanceEqual(instance, (self.bulk_add_data or self.bulk_create_data))
                matching_count += 1
            except AssertionError:
                pass
        self.assertEqual(matching_count, self.bulk_create_count)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_bulk_rename(self):
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        objects = self.selected_objects
        pk_list = [obj.pk for obj in objects]
        # Apply button not yet clicked
        data = {"pk": pk_list}
        data.update(self.rename_data)
        verbose_name_plural = self.model._meta.verbose_name_plural

        with self.subTest("Assert device name in HTML"):
            response = self.client.post(self._get_url("bulk_rename"), data)
            message = (
                f"Renaming {len(objects)} {helpers.bettertitle(verbose_name_plural)} "
                f"on {self.selected_objects_parent_name}"
            )
            self.assertInHTML(message, response.content.decode(response.charset))

        with self.subTest("Assert update successfully"):
            data["_apply"] = True  # Form Apply button
            response = self.client.post(self._get_url("bulk_rename"), data)
            self.assertHttpStatus(response, 302)
            queryset = self._get_queryset().filter(pk__in=pk_list)
            for instance in objects:
                self.assertEqual(queryset.get(pk=instance.pk).name, f"{instance.name}X")

        with self.subTest("Assert if no valid objects selected return with error"):
            for values in ([], [str(uuid.uuid4())]):
                data["pk"] = values
                response = self.client.post(self._get_url("bulk_rename"), data, follow=True)
                expected_message = f"No valid {verbose_name_plural} were selected."
                self.assertIn(expected_message, response.content.decode(response.charset))

bulk_add_data = None class-attribute instance-attribute

Used for bulk-add (distinct from bulk-create) view testing; self.bulk_create_data will be used if unset.

test_bulk_add_component()

Test bulk-adding this component to devices/virtual-machines.

Source code in nautobot/core/testing/views.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
def test_bulk_add_component(self):
    """Test bulk-adding this component to devices/virtual-machines."""
    obj_perm = users_models.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))

    initial_count = self._get_queryset().count()

    data = (self.bulk_add_data or self.bulk_create_data).copy()

    # Load the device-bulk-add or virtualmachine-bulk-add form
    if "device" in data:
        url = reverse(f"dcim:device_bulk_add_{self.model._meta.model_name}")
        request = {
            "path": url,
            "data": testing.post_data({"pk": data["device"]}),
        }
    else:
        url = reverse(f"virtualization:virtualmachine_bulk_add_{self.model._meta.model_name}")
        request = {
            "path": url,
            "data": testing.post_data({"pk": data["virtual_machine"]}),
        }
    self.assertHttpStatus(self.client.post(**request), 200)

    # Post to the device-bulk-add or virtualmachine-bulk-add form to create records
    if "device" in data:
        data["pk"] = data.pop("device")
    else:
        data["pk"] = data.pop("virtual_machine")
    data["_create"] = ""
    request["data"] = testing.post_data(data)
    self.assertHttpStatus(self.client.post(**request), 302)

    updated_count = self._get_queryset().count()
    self.assertEqual(updated_count, initial_count + self.bulk_create_count)

    matching_count = 0
    for instance in self._get_queryset().all():
        try:
            self.assertInstanceEqual(instance, (self.bulk_add_data or self.bulk_create_data))
            matching_count += 1
        except AssertionError:
            pass
    self.assertEqual(matching_count, self.bulk_create_count)

EditObjectViewTestCase

Bases: ModelViewTestCase

Edit a single existing instance.

:form_data: Data to be used when updating the first existing object.

Source code in nautobot/core/testing/views.py

class EditObjectViewTestCase(ModelViewTestCase):
    """
    Edit a single existing instance.

    :form_data: Data to be used when updating the first existing object.
    """

    form_data = {}

    def test_edit_object_without_permission(self):
        instance = self._get_queryset().first()

        # Try GET without permission
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.get(self._get_url("edit", instance)), [403, 404])

        # Try POST without permission
        request = {
            "path": self._get_url("edit", instance),
            "data": testing.post_data(self.form_data),
        }
        with testing.disable_warnings("django.request"):
            self.assertHttpStatus(self.client.post(**request), [403, 404])

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_edit_object_with_permission(self):
        instance = self._get_queryset().first()

        # Assign model-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["change"])
        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("edit", instance)), 200)

        # Try POST with model-level permission
        request = {
            "path": self._get_url("edit", instance),
            "data": testing.post_data(self.form_data),
        }
        self.assertHttpStatus(self.client.post(**request), 302)
        self.assertInstanceEqual(self._get_queryset().get(pk=instance.pk), self.form_data)

        if hasattr(self.model, "to_objectchange"):
            # Verify ObjectChange creation
            objectchanges = lookup.get_changes_for_model(instance)
            self.assertEqual(len(objectchanges), 1)
            self.assertEqual(objectchanges[0].action, extras_choices.ObjectChangeActionChoices.ACTION_UPDATE)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_edit_object_with_constrained_permission(self):
        instance1, instance2 = self._get_queryset().all()[:2]

        # Assign constrained permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={"pk": instance1.pk},
            actions=["change"],
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try GET with a permitted object
        self.assertHttpStatus(self.client.get(self._get_url("edit", instance1)), 200)

        # Try GET with a non-permitted object
        self.assertHttpStatus(self.client.get(self._get_url("edit", instance2)), 404)

        # Try to edit a permitted object
        request = {
            "path": self._get_url("edit", instance1),
            "data": testing.post_data(self.form_data),
        }
        self.assertHttpStatus(self.client.post(**request), 302)
        self.assertInstanceEqual(self._get_queryset().get(pk=instance1.pk), self.form_data)

        # Try to edit a non-permitted object
        request = {
            "path": self._get_url("edit", instance2),
            "data": testing.post_data(self.form_data),
        }
        self.assertHttpStatus(self.client.post(**request), 404)

GetObjectChangelogViewTestCase

Bases: ModelViewTestCase

View the changelog for an instance.

Source code in nautobot/core/testing/views.py

class GetObjectChangelogViewTestCase(ModelViewTestCase):
    """
    View the changelog for an instance.
    """

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_get_object_changelog(self):
        url = self._get_url("changelog", self._get_queryset().first())
        response = self.client.get(url)
        self.assertHttpStatus(response, 200)

GetObjectNotesViewTestCase

Bases: ModelViewTestCase

View the notes for an instance.

Source code in nautobot/core/testing/views.py

class GetObjectNotesViewTestCase(ModelViewTestCase):
    """
    View the notes for an instance.
    """

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_get_object_notes(self):
        if hasattr(self.model, "notes"):
            url = self._get_url("notes", self._get_queryset().first())
            response = self.client.get(url)
            self.assertHttpStatus(response, 200)

GetObjectViewTestCase

Bases: ModelViewTestCase

Retrieve a single instance.

Source code in nautobot/core/testing/views.py

class GetObjectViewTestCase(ModelViewTestCase):
    """
    Retrieve a single instance.
    """

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_get_object_anonymous(self):
        # Make the request as an unauthenticated user
        self.client.logout()
        response = self.client.get(self._get_queryset().first().get_absolute_url())
        self.assertHttpStatus(response, 200)
        response_body = response.content.decode(response.charset)
        self.assertIn(
            "/login/?next=" + self._get_queryset().first().get_absolute_url(), response_body, msg=response_body
        )

        # The "Change Log" tab should appear in the response since we have all exempt permissions
        if issubclass(self.model, extras_models.ChangeLoggedModel):
            response_body = testing.extract_page_body(response.content.decode(response.charset))
            self.assertIn("Change Log", response_body, msg=response_body)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_get_object_without_permission(self):
        instance = self._get_queryset().first()

        # Try GET without permission
        with testing.disable_warnings("django.request"):
            response = self.client.get(instance.get_absolute_url())
            self.assertHttpStatus(response, [403, 404])
            response_body = response.content.decode(response.charset)
            self.assertNotIn("/login/", response_body, msg=response_body)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_get_object_with_permission(self):
        instance = self._get_queryset().first()

        # Add model-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view"])
        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
        response = self.client.get(instance.get_absolute_url())
        self.assertHttpStatus(response, 200)

        response_body = testing.extract_page_body(response.content.decode(response.charset))

        # The object's display name or string representation should appear in the response
        self.assertIn(escape(getattr(instance, "display", str(instance))), response_body, msg=response_body)

        # If any Relationships are defined, they should appear in the response
        if self.relationships is not None:
            for relationship in self.relationships:  # false positive pylint: disable=not-an-iterable
                content_type = ContentType.objects.get_for_model(instance)
                if content_type == relationship.source_type:
                    self.assertIn(
                        escape(relationship.get_label(extras_choices.RelationshipSideChoices.SIDE_SOURCE)),
                        response_body,
                        msg=response_body,
                    )
                if content_type == relationship.destination_type:
                    self.assertIn(
                        escape(relationship.get_label(extras_choices.RelationshipSideChoices.SIDE_DESTINATION)),
                        response_body,
                        msg=response_body,
                    )

        # If any Custom Fields are defined, they should appear in the response
        if self.custom_fields is not None:
            for custom_field in self.custom_fields:  # false positive pylint: disable=not-an-iterable
                self.assertIn(escape(str(custom_field)), response_body, msg=response_body)
                if custom_field.type == extras_choices.CustomFieldTypeChoices.TYPE_MULTISELECT:
                    for value in instance.cf.get(custom_field.key):
                        self.assertIn(escape(str(value)), response_body, msg=response_body)
                else:
                    self.assertIn(
                        escape(str(instance.cf.get(custom_field.key) or "")), response_body, msg=response_body
                    )

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_get_object_with_constrained_permission(self):
        instance1, instance2 = self._get_queryset().all()[:2]

        # Add object-level permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={"pk": instance1.pk},
            # To get a different rendering flow than the `test_get_object_with_permission` test above,
            # enable additional permissions for this object so that add/edit/delete buttons are rendered.
            actions=["view", "add", "change", "delete"],
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try GET to permitted object
        self.assertHttpStatus(self.client.get(instance1.get_absolute_url()), 200)

        # Try GET to non-permitted object
        self.assertHttpStatus(self.client.get(instance2.get_absolute_url()), 404)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_has_advanced_tab(self):
        instance = self._get_queryset().first()

        # Add model-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view"])
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        response = self.client.get(instance.get_absolute_url())
        response_body = testing.extract_page_body(response.content.decode(response.charset))
        advanced_tab_href = f"{instance.get_absolute_url()}#advanced"

        self.assertIn(advanced_tab_href, response_body)
        self.assertIn("Advanced", response_body)

ListObjectsViewTestCase

Bases: ModelViewTestCase

Retrieve multiple instances.

Source code in nautobot/core/testing/views.py

class ListObjectsViewTestCase(ModelViewTestCase):
    """
    Retrieve multiple instances.
    """

    filterset = None

    def get_filterset(self):
        return self.filterset or lookup.get_filterset_for_model(self.model)

    # Helper methods to be overriden by special cases.
    # See ConsoleConnectionsTestCase, InterfaceConnectionsTestCase and PowerConnectionsTestCase
    def get_list_url(self):
        return reverse(helpers.validated_viewname(self.model, "list"))

    def get_title(self):
        return helpers.bettertitle(self.model._meta.verbose_name_plural)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_list_objects_anonymous(self):
        # Make the request as an unauthenticated user
        self.client.logout()
        response = self.client.get(self._get_url("list"))
        self.assertHttpStatus(response, 200)
        response_body = response.content.decode(response.charset)
        self.assertIn("/login/?next=" + self._get_url("list"), response_body, msg=response_body)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"])
    def test_list_objects_filtered(self):
        instance1, instance2 = self._get_queryset().all()[:2]
        response = self.client.get(f"{self._get_url('list')}?id={instance1.pk}")
        self.assertHttpStatus(response, 200)
        content = testing.extract_page_body(response.content.decode(response.charset))
        # TODO: it'd make test failures more readable if we strip the page headers/footers from the content
        if hasattr(self.model, "name"):
            self.assertRegex(content, r">\s*" + re.escape(escape(instance1.name)) + r"\s*<", msg=content)
            self.assertNotRegex(content, r">\s*" + re.escape(escape(instance2.name)) + r"\s*<", msg=content)
        if instance1.get_absolute_url() in content:
            self.assertNotIn(instance2.get_absolute_url(), content, msg=content)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"], STRICT_FILTERING=True)
    def test_list_objects_unknown_filter_strict_filtering(self):
        """Verify that with STRICT_FILTERING, an unknown filter results in an error message and no matches."""
        response = self.client.get(f"{self._get_url('list')}?ice_cream_flavor=chocolate")
        self.assertHttpStatus(response, 200)
        content = testing.extract_page_body(response.content.decode(response.charset))
        # TODO: it'd make test failures more readable if we strip the page headers/footers from the content
        self.assertIn("Unknown filter field", content, msg=content)
        # There should be no table rows displayed except for the empty results row
        self.assertIn(f"No {self.model._meta.verbose_name_plural} found", content, msg=content)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=["*"], STRICT_FILTERING=False)
    def test_list_objects_unknown_filter_no_strict_filtering(self):
        """Verify that without STRICT_FILTERING, an unknown filter is ignored."""
        instance1, instance2 = self._get_queryset().all()[:2]
        with self.assertLogs("nautobot.core.filters") as cm:
            response = self.client.get(f"{self._get_url('list')}?ice_cream_flavor=chocolate")
        filterset = self.get_filterset()
        if not filterset:
            self.fail(
                f"Couldn't find filterset for model {self.model}. The FilterSet class is expected to be in the "
                "filters module within the application associated with the model and its name is expected to be "
                f"{self.model.__name__}FilterSet."
            )
        self.assertEqual(
            cm.output,
            [
                f'WARNING:nautobot.core.filters:{filterset.__name__}: Unknown filter field "ice_cream_flavor"',
            ],
        )
        self.assertHttpStatus(response, 200)
        content = testing.extract_page_body(response.content.decode(response.charset))
        # TODO: it'd make test failures more readable if we strip the page headers/footers from the content
        self.assertNotIn("Unknown filter field", content, msg=content)
        self.assertNotIn(f"No {self.model._meta.verbose_name_plural} found", content, msg=content)
        if hasattr(self.model, "name"):
            self.assertRegex(content, r">\s*" + re.escape(instance1.name) + r"\s*<", msg=content)
            self.assertRegex(content, r">\s*" + re.escape(instance2.name) + r"\s*<", msg=content)
        if instance1.get_absolute_url() in content:
            self.assertIn(instance2.get_absolute_url(), content, msg=content)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects_without_permission(self):
        # Try GET without permission
        with testing.disable_warnings("django.request"):
            response = self.client.get(self._get_url("list"))
            self.assertHttpStatus(response, 403)
            response_body = response.content.decode(response.charset)
            self.assertNotIn("/login/", response_body, msg=response_body)

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects_with_permission(self):
        # Add model-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view"])
        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
        response = self.client.get(self._get_url("list"))
        self.assertHttpStatus(response, 200)
        response_body = response.content.decode(response.charset)

        list_url = self.get_list_url()
        title = self.get_title()

        # Check if breadcrumb is rendered correctly
        self.assertIn(
            f'<a href="{list_url}">{title}</a>',
            response_body,
        )

    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_objects_with_constrained_permission(self):
        instance1, instance2 = self._get_queryset().all()[:2]

        # Add object-level permission
        obj_perm = users_models.ObjectPermission(
            name="Test permission",
            constraints={"pk": instance1.pk},
            actions=["view"],
        )
        obj_perm.save()
        obj_perm.users.add(self.user)
        obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))

        # Try GET with object-level permission
        response = self.client.get(self._get_url("list"))
        self.assertHttpStatus(response, 200)
        content = testing.extract_page_body(response.content.decode(response.charset))
        # TODO: it'd make test failures more readable if we strip the page headers/footers from the content
        if hasattr(self.model, "name"):
            self.assertRegex(content, r">\s*" + re.escape(instance1.name) + r"\s*<", msg=content)
            self.assertNotRegex(content, r">\s*" + re.escape(instance2.name) + r"\s*<", msg=content)
        elif hasattr(self.model, "get_absolute_url"):
            self.assertIn(instance1.get_absolute_url(), content, msg=content)
            self.assertNotIn(instance2.get_absolute_url(), content, msg=content)

    @skipIf(
        "example_plugin" not in settings.PLUGINS,
        "example_plugin not in settings.PLUGINS",
    )
    @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
    def test_list_view_plugin_banner(self):
        """
        If example plugin is installed, check if the plugin banner is rendered correctly in ObjectListView.
        """
        # Add model-level permission
        obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view"])
        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
        response = self.client.get(self._get_url("list"))
        self.assertHttpStatus(response, 200)
        response_body = response.content.decode(response.charset)

        # Check plugin banner is rendered correctly
        self.assertIn(
            f"<div>You are viewing a table of {self.model._meta.verbose_name_plural}</div>", response_body
        )

test_list_objects_unknown_filter_no_strict_filtering()

Verify that without STRICT_FILTERING, an unknown filter is ignored.

Source code in nautobot/core/testing/views.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=["*"], STRICT_FILTERING=False)
def test_list_objects_unknown_filter_no_strict_filtering(self):
    """Verify that without STRICT_FILTERING, an unknown filter is ignored."""
    instance1, instance2 = self._get_queryset().all()[:2]
    with self.assertLogs("nautobot.core.filters") as cm:
        response = self.client.get(f"{self._get_url('list')}?ice_cream_flavor=chocolate")
    filterset = self.get_filterset()
    if not filterset:
        self.fail(
            f"Couldn't find filterset for model {self.model}. The FilterSet class is expected to be in the "
            "filters module within the application associated with the model and its name is expected to be "
            f"{self.model.__name__}FilterSet."
        )
    self.assertEqual(
        cm.output,
        [
            f'WARNING:nautobot.core.filters:{filterset.__name__}: Unknown filter field "ice_cream_flavor"',
        ],
    )
    self.assertHttpStatus(response, 200)
    content = testing.extract_page_body(response.content.decode(response.charset))
    # TODO: it'd make test failures more readable if we strip the page headers/footers from the content
    self.assertNotIn("Unknown filter field", content, msg=content)
    self.assertNotIn(f"No {self.model._meta.verbose_name_plural} found", content, msg=content)
    if hasattr(self.model, "name"):
        self.assertRegex(content, r">\s*" + re.escape(instance1.name) + r"\s*<", msg=content)
        self.assertRegex(content, r">\s*" + re.escape(instance2.name) + r"\s*<", msg=content)
    if instance1.get_absolute_url() in content:
        self.assertIn(instance2.get_absolute_url(), content, msg=content)

test_list_objects_unknown_filter_strict_filtering()

Verify that with STRICT_FILTERING, an unknown filter results in an error message and no matches.

Source code in nautobot/core/testing/views.py

@override_settings(EXEMPT_VIEW_PERMISSIONS=["*"], STRICT_FILTERING=True)
def test_list_objects_unknown_filter_strict_filtering(self):
    """Verify that with STRICT_FILTERING, an unknown filter results in an error message and no matches."""
    response = self.client.get(f"{self._get_url('list')}?ice_cream_flavor=chocolate")
    self.assertHttpStatus(response, 200)
    content = testing.extract_page_body(response.content.decode(response.charset))
    # TODO: it'd make test failures more readable if we strip the page headers/footers from the content
    self.assertIn("Unknown filter field", content, msg=content)
    # There should be no table rows displayed except for the empty results row
    self.assertIn(f"No {self.model._meta.verbose_name_plural} found", content, msg=content)

test_list_view_plugin_banner()

If example plugin is installed, check if the plugin banner is rendered correctly in ObjectListView.

Source code in nautobot/core/testing/views.py

@skipIf(
    "example_plugin" not in settings.PLUGINS,
    "example_plugin not in settings.PLUGINS",
)
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_list_view_plugin_banner(self):
    """
    If example plugin is installed, check if the plugin banner is rendered correctly in ObjectListView.
    """
    # Add model-level permission
    obj_perm = users_models.ObjectPermission(name="Test permission", actions=["view"])
    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
    response = self.client.get(self._get_url("list"))
    self.assertHttpStatus(response, 200)
    response_body = response.content.decode(response.charset)

    # Check plugin banner is rendered correctly
    self.assertIn(
        f"<div>You are viewing a table of {self.model._meta.verbose_name_plural}</div>", response_body
    )

OrganizationalObjectViewTestCase

Bases: GetObjectViewTestCase, GetObjectChangelogViewTestCase, GetObjectNotesViewTestCase, CreateObjectViewTestCase, EditObjectViewTestCase, DeleteObjectViewTestCase, ListObjectsViewTestCase, BulkImportObjectsViewTestCase, BulkDeleteObjectsViewTestCase

TestCase suitable for all organizational objects

Source code in nautobot/core/testing/views.py

class OrganizationalObjectViewTestCase(
    GetObjectViewTestCase,
    GetObjectChangelogViewTestCase,
    GetObjectNotesViewTestCase,
    CreateObjectViewTestCase,
    EditObjectViewTestCase,
    DeleteObjectViewTestCase,
    ListObjectsViewTestCase,
    BulkImportObjectsViewTestCase,
    BulkDeleteObjectsViewTestCase,
):
    """
    TestCase suitable for all organizational objects
    """

    maxDiff = None

PrimaryObjectViewTestCase

Bases: GetObjectViewTestCase, GetObjectChangelogViewTestCase, GetObjectNotesViewTestCase, CreateObjectViewTestCase, EditObjectViewTestCase, DeleteObjectViewTestCase, ListObjectsViewTestCase, BulkImportObjectsViewTestCase, BulkEditObjectsViewTestCase, BulkDeleteObjectsViewTestCase

TestCase suitable for testing all standard View functions for primary objects

Source code in nautobot/core/testing/views.py

class PrimaryObjectViewTestCase(
    GetObjectViewTestCase,
    GetObjectChangelogViewTestCase,
    GetObjectNotesViewTestCase,
    CreateObjectViewTestCase,
    EditObjectViewTestCase,
    DeleteObjectViewTestCase,
    ListObjectsViewTestCase,
    BulkImportObjectsViewTestCase,
    BulkEditObjectsViewTestCase,
    BulkDeleteObjectsViewTestCase,
):
    """
    TestCase suitable for testing all standard View functions for primary objects
    """

    maxDiff = None