netbox/netbox/extras/tests/test_api.py

import datetime
import hashlib
import io
import json
from contextlib import contextmanager
from unittest.mock import MagicMock, patch

from django.contrib.contenttypes.models import ContentType
from django.core.files.uploadedfile import SimpleUploadedFile
from django.db import IntegrityError
from django.urls import reverse
from django.utils.timezone import make_aware, now
from rest_framework import status

from core.choices import ManagedFileRootPathChoices
from core.events import *
from core.models import DataFile, DataSource, ObjectType
from dcim.models import Device, DeviceRole, DeviceType, Location, Manufacturer, Rack, RackRole, Site
from extras.choices import *
from extras.models import *
from extras.scripts import BooleanVar, IntegerVar, StringVar
from extras.scripts import Script as PythonClass
from users.constants import TOKEN_PREFIX
from users.models import Group, ObjectPermission, Token, User
from utilities.tables import get_table_for_model
from utilities.testing import APITestCase, APIViewTestCases


class AppTestCase(APITestCase):

    def test_root(self):

        url = reverse('extras-api:api-root')
        response = self.client.get('{}?format=api'.format(url), **self.header)

        self.assertEqual(response.status_code, 200)


class WebhookTestCase(APIViewTestCases.APIViewTestCase):
    model = Webhook
    brief_fields = ['description', 'display', 'id', 'name', 'url']
    create_data = [
        {
            'name': 'Webhook 4',
            'payload_url': 'http://example.com/?4',
        },
        {
            'name': 'Webhook 5',
            'payload_url': 'http://example.com/?5',
        },
        {
            'name': 'Webhook 6',
            'payload_url': 'http://example.com/?6',
        },
    ]
    bulk_update_data = {
        'description': 'New description',
        'ssl_verification': False,
    }

    @classmethod
    def setUpTestData(cls):

        webhooks = (
            Webhook(
                name='Webhook 1',
                payload_url='http://example.com/?1',
            ),
            Webhook(
                name='Webhook 2',
                payload_url='http://example.com/?1',
            ),
            Webhook(
                name='Webhook 3',
                payload_url='http://example.com/?1',
            ),
        )
        Webhook.objects.bulk_create(webhooks)


class EventRuleTestCase(APIViewTestCases.APIViewTestCase):
    model = EventRule
    brief_fields = ['description', 'display', 'id', 'name', 'url']
    bulk_update_data = {
        'enabled': False,
        'description': 'New description',
    }
    update_data = {
        'name': 'Event Rule X',
        'enabled': False,
        'description': 'New description',
    }

    @classmethod
    def setUpTestData(cls):
        webhooks = (
            Webhook(
                name='Webhook 1',
                payload_url='http://example.com/?1',
            ),
            Webhook(
                name='Webhook 2',
                payload_url='http://example.com/?1',
            ),
            Webhook(
                name='Webhook 3',
                payload_url='http://example.com/?1',
            ),
            Webhook(
                name='Webhook 4',
                payload_url='http://example.com/?1',
            ),
            Webhook(
                name='Webhook 5',
                payload_url='http://example.com/?1',
            ),
            Webhook(
                name='Webhook 6',
                payload_url='http://example.com/?1',
            ),
        )
        Webhook.objects.bulk_create(webhooks)

        event_rules = (
            EventRule(name='EventRule 1', event_types=[OBJECT_CREATED], action_object=webhooks[0]),
            EventRule(name='EventRule 2', event_types=[OBJECT_CREATED], action_object=webhooks[1]),
            EventRule(name='EventRule 3', event_types=[OBJECT_CREATED], action_object=webhooks[2]),
        )
        EventRule.objects.bulk_create(event_rules)

        cls.create_data = [
            {
                'name': 'EventRule 4',
                'object_types': ['dcim.device', 'dcim.devicetype'],
                'event_types': [OBJECT_CREATED],
                'action_type': EventRuleActionChoices.WEBHOOK,
                'action_object_type': 'extras.webhook',
                'action_object_id': webhooks[3].pk,
            },
            {
                'name': 'EventRule 5',
                'object_types': ['dcim.device', 'dcim.devicetype'],
                'event_types': [OBJECT_CREATED],
                'action_type': EventRuleActionChoices.WEBHOOK,
                'action_object_type': 'extras.webhook',
                'action_object_id': webhooks[4].pk,
            },
            {
                'name': 'EventRule 6',
                'object_types': ['dcim.device', 'dcim.devicetype'],
                'event_types': [OBJECT_CREATED],
                'action_type': EventRuleActionChoices.WEBHOOK,
                'action_object_type': 'extras.webhook',
                'action_object_id': webhooks[5].pk,
            },
        ]


class CustomFieldTestCase(APIViewTestCases.APIViewTestCase):
    model = CustomField
    brief_fields = ['description', 'display', 'id', 'name', 'url']
    create_data = [
        {
            'object_types': ['dcim.site'],
            'name': 'cf4',
            'type': 'date',
        },
        {
            'object_types': ['dcim.site'],
            'name': 'cf5',
            'type': 'url',
        },
        {
            'object_types': ['dcim.site'],
            'name': 'cf6',
            'type': 'text',
        },
    ]
    bulk_update_data = {
        'description': 'New description',
    }
    update_data = {
        'object_types': ['dcim.device'],
        'name': 'New_Name',
        'description': 'New description',
    }

    @classmethod
    def setUpTestData(cls):
        site_ct = ObjectType.objects.get_for_model(Site)

        custom_fields = (
            CustomField(
                name='cf1',
                type='text'
            ),
            CustomField(
                name='cf2',
                type='integer'
            ),
            CustomField(
                name='cf3',
                type='boolean'
            ),
        )
        CustomField.objects.bulk_create(custom_fields)
        for cf in custom_fields:
            cf.object_types.add(site_ct)


class CustomFieldChoiceSetTestCase(APIViewTestCases.APIViewTestCase):
    model = CustomFieldChoiceSet
    brief_fields = ['choices_count', 'description', 'display', 'id', 'name', 'url']
    create_data = [
        {
            'name': 'Choice Set 4',
            'extra_choices': [
                ['4A', 'Choice 1'],
                ['4B', 'Choice 2'],
                ['4C', 'Choice 3'],
            ],
            'choice_colors': {
                '4A': 'red',
                '4B': 'green',
            },
        },
        {
            'name': 'Choice Set 5',
            'extra_choices': [
                ['5A', 'Choice 1'],
                ['5B', 'Choice 2'],
                ['5C', 'Choice 3'],
            ],
            'choice_colors': {
                '5C': 'blue',
            },
        },
        {
            'name': 'Choice Set 6',
            'extra_choices': [
                ['6A', 'Choice 1'],
                ['6B', 'Choice 2'],
                ['6C', 'Choice 3'],
            ],
        },
    ]
    bulk_update_data = {
        'description': 'New description',
    }
    update_data = {
        'name': 'Choice Set X',
        'extra_choices': [
            ['X1', 'Choice 1'],
            ['X2', 'Choice 2'],
            ['X3', 'Choice 3'],
        ],
        'choice_colors': {
            'X1': 'red',
            'X3': 'green',
        },
        'description': 'New description',
    }

    @classmethod
    def setUpTestData(cls):
        choice_sets = (
            CustomFieldChoiceSet(
                name='Choice Set 1',
                extra_choices=[['1A', '1A'], ['1B', '1B'], ['1C', '1C'], ['1D', '1D'], ['1E', '1E']],
            ),
            CustomFieldChoiceSet(
                name='Choice Set 2',
                extra_choices=[['2A', '2A'], ['2B', '2B'], ['2C', '2C'], ['2D', '2D'], ['2E', '2E']],
            ),
            CustomFieldChoiceSet(
                name='Choice Set 3',
                extra_choices=[['3A', '3A'], ['3B', '3B'], ['3C', '3C'], ['3D', '3D'], ['3E', '3E']],
            ),
        )
        CustomFieldChoiceSet.objects.bulk_create(choice_sets)

    def test_invalid_choice_items(self):
        """
        Attempting to define each choice as a single-item list should return a 400 error.
        """
        self.add_permissions('extras.add_customfieldchoiceset')
        data = {
            "name": "test",
            "extra_choices": [
                ["choice1"],
                ["choice2"],
                ["choice3"],
            ]
        }

        response = self.client.post(self._get_list_url(), data, format='json', **self.header)
        self.assertEqual(response.status_code, 400)

    def test_invalid_choice_color(self):
        self.add_permissions('extras.add_customfieldchoiceset')
        data = {
            'name': 'test',
            'extra_choices': [
                ['choice1', 'Choice 1'],
                ['choice2', 'Choice 2'],
            ],
            'choice_colors': {
                'choice1': 'magenta',
            },
        }

        response = self.client.post(self._get_list_url(), data, format='json', **self.header)
        self.assertEqual(response.status_code, 400)

    def test_invalid_choice_color_reference(self):
        self.add_permissions('extras.add_customfieldchoiceset')
        data = {
            'name': 'test',
            'extra_choices': [
                ['choice1', 'Choice 1'],
                ['choice2', 'Choice 2'],
            ],
            'choice_colors': {
                'choice3': 'red',
            },
        }

        response = self.client.post(self._get_list_url(), data, format='json', **self.header)
        self.assertEqual(response.status_code, 400)

    def test_graphql_filter_extra_choices(self):
        """Filter choice sets by choice value and by number of choices."""
        self.add_permissions('extras.view_customfieldchoiceset')

        # '1A' appears here only as a label, so it must not match contains
        CustomFieldChoiceSet.objects.create(
            name='Choice Set Labels',
            extra_choices=[['sel1', 'Selection 1'], ['other', '1A']],
        )

        def run(lookup):
            query = '{ custom_field_choice_set_list(filters: {extra_choices: ' + lookup + '}) { name } }'
            response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            data = response.json()
            self.assertNotIn('errors', data)
            return sorted(row['name'] for row in data['data']['custom_field_choice_set_list'])

        # contains matches choice values only, never labels
        self.assertEqual(run('{contains: "1A"}'), ['Choice Set 1'])
        self.assertEqual(run('{contains: "sel1"}'), ['Choice Set Labels'])
        self.assertEqual(run('{contains: "Selection 1"}'), [])
        # length is the number of [value, label] pairs
        self.assertEqual(run('{length: 2}'), ['Choice Set Labels'])
        self.assertEqual(run('{length: 1}'), [])

    def test_graphql_filter_extra_choices_rejects_array_operands(self):
        """The legacy flat and nested array operand shapes fail schema validation."""
        self.add_permissions('extras.view_customfieldchoiceset')

        def run_invalid(lookup):
            query = '{ custom_field_choice_set_list(filters: {extra_choices: ' + lookup + '}) { name } }'
            response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
            self.assertHttpStatus(response, status.HTTP_200_OK)
            self.assertIn('errors', response.json())

        # shapes advertised or attempted before #22324
        run_invalid('{contains: ["1A"]}')
        run_invalid('{contains: [["1A", "Choice 1A"]]}')

    def test_graphql_filter_extra_choices_via_relation(self):
        """The extra_choices lookup composes through the choice_set relation prefix."""
        self.add_permissions('extras.view_customfield')

        for choice_set in CustomFieldChoiceSet.objects.filter(name__in=['Choice Set 1', 'Choice Set 2']):
            CustomField.objects.create(
                name=f'cf_{choice_set.name[-1]}',
                type=CustomFieldTypeChoices.TYPE_SELECT,
                choice_set=choice_set,
            )

        query = '{ custom_field_list(filters: {choice_set: {extra_choices: {contains: "1A"}}}) { name } }'
        response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        data = response.json()
        self.assertNotIn('errors', data)
        self.assertEqual([row['name'] for row in data['data']['custom_field_list']], ['cf_1'])


class CustomLinkTestCase(APIViewTestCases.APIViewTestCase):
    model = CustomLink
    brief_fields = ['display', 'id', 'name', 'url']
    create_data = [
        {
            'object_types': ['dcim.site'],
            'name': 'Custom Link 4',
            'enabled': True,
            'link_text': 'Link 4',
            'link_url': 'http://example.com/?4',
        },
        {
            'object_types': ['dcim.site'],
            'name': 'Custom Link 5',
            'enabled': True,
            'link_text': 'Link 5',
            'link_url': 'http://example.com/?5',
        },
        {
            'object_types': ['dcim.site'],
            'name': 'Custom Link 6',
            'enabled': False,
            'link_text': 'Link 6',
            'link_url': 'http://example.com/?6',
        },
    ]
    bulk_update_data = {
        'new_window': True,
        'enabled': False,
    }

    @classmethod
    def setUpTestData(cls):
        site_type = ObjectType.objects.get_for_model(Site)

        custom_links = (
            CustomLink(
                name='Custom Link 1',
                enabled=True,
                link_text='Link 1',
                link_url='http://example.com/?1',
            ),
            CustomLink(
                name='Custom Link 2',
                enabled=True,
                link_text='Link 2',
                link_url='http://example.com/?2',
            ),
            CustomLink(
                name='Custom Link 3',
                enabled=False,
                link_text='Link 3',
                link_url='http://example.com/?3',
            ),
        )
        CustomLink.objects.bulk_create(custom_links)
        for i, custom_link in enumerate(custom_links):
            custom_link.object_types.set([site_type])


class SharedObjectAPITestMixin:
    """
    Helpers for testing the shared/owner visibility enforced on SavedFilter and TableConfig.
    """
    def _grant_view_permission_and_authenticate(self, user, model):
        """
        Grant `user` an unconstrained view permission on `model`, create an API token, and return the
        corresponding authentication header.
        """
        obj_perm = ObjectPermission(name=f'{model._meta.model_name} view', actions=['view'])
        obj_perm.save()
        obj_perm.users.add(user)
        obj_perm.object_types.add(ObjectType.objects.get_for_model(model))
        token = Token.objects.create(user=user)
        return {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{token.key}.{token.token}'}


class SavedFilterTestCase(SharedObjectAPITestMixin, APIViewTestCases.APIViewTestCase):
    model = SavedFilter
    brief_fields = ['description', 'display', 'id', 'name', 'slug', 'url']
    create_data = [
        {
            'object_types': ['dcim.site'],
            'name': 'Saved Filter 4',
            'slug': 'saved-filter-4',
            'weight': 100,
            'enabled': True,
            'shared': True,
            'parameters': {'status': ['active']},
        },
        {
            'object_types': ['dcim.site'],
            'name': 'Saved Filter 5',
            'slug': 'saved-filter-5',
            'weight': 200,
            'enabled': True,
            'shared': True,
            'parameters': {'status': ['planned']},
        },
        {
            'object_types': ['dcim.site'],
            'name': 'Saved Filter 6',
            'slug': 'saved-filter-6',
            'weight': 300,
            'enabled': True,
            'shared': True,
            'parameters': {'status': ['retired']},
        },
    ]
    bulk_update_data = {
        'weight': 1000,
        'enabled': False,
        'shared': False,
    }

    @classmethod
    def setUpTestData(cls):
        site_type = ObjectType.objects.get_for_model(Site)

        saved_filters = (
            SavedFilter(
                name='Saved Filter 1',
                slug='saved-filter-1',
                weight=100,
                enabled=True,
                shared=True,
                parameters={'status': ['active']}
            ),
            SavedFilter(
                name='Saved Filter 2',
                slug='saved-filter-2',
                weight=200,
                enabled=True,
                shared=True,
                parameters={'status': ['planned']}
            ),
            SavedFilter(
                name='Saved Filter 3',
                slug='saved-filter-3',
                weight=300,
                enabled=True,
                shared=True,
                parameters={'status': ['retired']}
            ),
        )
        SavedFilter.objects.bulk_create(saved_filters)
        for i, savedfilter in enumerate(saved_filters):
            savedfilter.object_types.set([site_type])

    def test_private_filter_not_visible_to_other_users(self):
        """
        A private (shared=False) SavedFilter owned by another user must not be exposed via the REST API, even to
        a user holding an unconstrained view permission.
        """
        site_type = ObjectType.objects.get_for_model(Site)
        owner = User.objects.create_user(username='filter-owner')
        private_filter = SavedFilter.objects.create(
            name='Private Filter',
            slug='private-filter',
            user=owner,
            shared=False,
            parameters={'status': ['active']},
        )
        private_filter.object_types.set([site_type])

        # Grant an unconstrained view permission (the common case)
        self.add_permissions('extras.view_savedfilter')

        # The private filter must not appear in the list
        response = self.client.get(self._get_list_url(), **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        returned_ids = [obj['id'] for obj in response.data['results']]
        self.assertNotIn(private_filter.pk, returned_ids)

        # The private filter must not be retrievable directly
        response = self.client.get(self._get_detail_url(private_filter), **self.header)
        self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)

        # The private filter must not be exposed via GraphQL either
        query = '{ saved_filter_list { id } }'
        response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        data = json.loads(response.content)
        returned_ids = [int(obj['id']) for obj in data['data']['saved_filter_list']]
        self.assertNotIn(private_filter.pk, returned_ids)

        # The owner, however, must still be able to access their own private filter
        owner_header = self._grant_view_permission_and_authenticate(owner, SavedFilter)
        response = self.client.get(self._get_detail_url(private_filter), **owner_header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **owner_header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        data = json.loads(response.content)
        returned_ids = [int(obj['id']) for obj in data['data']['saved_filter_list']]
        self.assertIn(private_filter.pk, returned_ids)


class TableConfigTestCase(SharedObjectAPITestMixin, APIViewTestCases.APIViewTestCase):
    model = TableConfig
    brief_fields = ['description', 'display', 'id', 'name', 'object_type', 'table', 'url']
    bulk_update_data = {
        'description': 'New description',
        'weight': 999,
        'enabled': False,
        'shared': False,
    }

    @classmethod
    def setUpTestData(cls):
        site_type = ObjectType.objects.get_for_model(Site)
        site_table_name = get_table_for_model(Site).__name__

        users = (
            User(username='User 1'),
            User(username='User 2'),
            User(username='User 3'),
        )
        User.objects.bulk_create(users)

        table_configs = (
            TableConfig(
                name='Table Config 1',
                object_type=site_type,
                table=site_table_name,
                user=users[0],
                shared=True,
                columns=['name', 'status'],
            ),
            TableConfig(
                name='Table Config 2',
                object_type=site_type,
                table=site_table_name,
                user=users[1],
                shared=True,
                columns=['name', 'region'],
            ),
            TableConfig(
                name='Table Config 3',
                object_type=site_type,
                table=site_table_name,
                user=users[2],
                shared=True,
                columns=['name', 'tenant'],
            ),
        )
        TableConfig.objects.bulk_create(table_configs)

        cls.create_data = [
            {
                'object_type': 'dcim.site',
                'table': site_table_name,
                'name': 'Table Config 4',
                'columns': ['name', 'status'],
                'ordering': ['name'],
            },
            {
                'object_type': 'dcim.site',
                'table': site_table_name,
                'name': 'Table Config 5',
                'columns': ['name', 'region'],
                'ordering': ['-name'],
            },
            {
                'object_type': 'dcim.site',
                'table': site_table_name,
                'name': 'Table Config 6',
                'columns': ['name', 'tenant'],
            },
        ]

    def test_private_table_config_not_visible_to_other_users(self):
        """
        A private (shared=False) TableConfig owned by another user must not be exposed via the REST API, even to
        a user holding an unconstrained view permission.
        """
        site_type = ObjectType.objects.get_for_model(Site)
        site_table_name = get_table_for_model(Site).__name__
        owner = User.objects.create_user(username='tableconfig-owner')
        private_config = TableConfig.objects.create(
            name='Private Table Config',
            object_type=site_type,
            table=site_table_name,
            user=owner,
            shared=False,
            columns=['name', 'status'],
        )

        # Grant an unconstrained view permission (the common case)
        self.add_permissions('extras.view_tableconfig')

        # The private table config must not appear in the list
        response = self.client.get(self._get_list_url(), **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        returned_ids = [obj['id'] for obj in response.data['results']]
        self.assertNotIn(private_config.pk, returned_ids)

        # The private table config must not be retrievable directly
        response = self.client.get(self._get_detail_url(private_config), **self.header)
        self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)

        # The private table config must not be exposed via GraphQL either
        query = '{ table_config_list { id } }'
        response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        data = json.loads(response.content)
        returned_ids = [int(obj['id']) for obj in data['data']['table_config_list']]
        self.assertNotIn(private_config.pk, returned_ids)

        # The owner, however, must still be able to access their own private table config
        owner_header = self._grant_view_permission_and_authenticate(owner, TableConfig)
        response = self.client.get(self._get_detail_url(private_config), **owner_header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **owner_header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        data = json.loads(response.content)
        returned_ids = [int(obj['id']) for obj in data['data']['table_config_list']]
        self.assertIn(private_config.pk, returned_ids)


class BookmarkTestCase(
    APIViewTestCases.GetObjectViewTestCase,
    APIViewTestCases.ListObjectsViewTestCase,
    APIViewTestCases.CreateObjectViewTestCase,
    APIViewTestCases.DeleteObjectViewTestCase
):
    model = Bookmark
    brief_fields = ['display', 'id', 'object_id', 'object_type', 'url']

    @classmethod
    def setUpTestData(cls):
        sites = (
            Site(name='Site 1', slug='site-1'),
            Site(name='Site 2', slug='site-2'),
            Site(name='Site 3', slug='site-3'),
            Site(name='Site 4', slug='site-4'),
            Site(name='Site 5', slug='site-5'),
            Site(name='Site 6', slug='site-6'),
        )
        Site.objects.bulk_create(sites)

    def setUp(self):
        super().setUp()

        sites = Site.objects.all()

        bookmarks = (
            Bookmark(object=sites[0], user=self.user),
            Bookmark(object=sites[1], user=self.user),
            Bookmark(object=sites[2], user=self.user),
        )
        Bookmark.objects.bulk_create(bookmarks)

        self.create_data = [
            {
                'object_type': 'dcim.site',
                'object_id': sites[3].pk,
                'user': self.user.pk,
            },
            {
                'object_type': 'dcim.site',
                'object_id': sites[4].pk,
                'user': self.user.pk,
            },
            {
                'object_type': 'dcim.site',
                'object_id': sites[5].pk,
                'user': self.user.pk,
            },
        ]


class ExportTemplateTestCase(APIViewTestCases.APIViewTestCase):
    model = ExportTemplate
    brief_fields = ['description', 'display', 'id', 'name', 'url']
    create_data = [
        {
            'object_types': ['dcim.device'],
            'name': 'Test Export Template 4',
            'template_code': '{% for obj in queryset %}{{ obj.name }}\n{% endfor %}',
        },
        {
            'object_types': ['dcim.device'],
            'name': 'Test Export Template 5',
            'template_code': '{% for obj in queryset %}{{ obj.name }}\n{% endfor %}',
        },
        {
            'object_types': ['dcim.device'],
            'name': 'Test Export Template 6',
            'template_code': '{% for obj in queryset %}{{ obj.name }}\n{% endfor %}',
            'file_name': 'test_export_template_6',
        },
    ]
    bulk_update_data = {
        'description': 'New description',
    }

    @classmethod
    def setUpTestData(cls):
        export_templates = (
            ExportTemplate(
                name='Export Template 1',
                template_code='{% for obj in queryset %}{{ obj.name }}\n{% endfor %}'
            ),
            ExportTemplate(
                name='Export Template 2',
                template_code='{% for obj in queryset %}{{ obj.name }}\n{% endfor %}',
                file_name='export_template_2',
                file_extension='test',
            ),
            ExportTemplate(
                name='Export Template 3',
                template_code='{% for obj in queryset %}{{ obj.name }}\n{% endfor %}'
            ),
        )
        ExportTemplate.objects.bulk_create(export_templates)

        device_object_type = ObjectType.objects.get_for_model(Device)
        for et in export_templates:
            et.object_types.set([device_object_type])


class TagTestCase(APIViewTestCases.APIViewTestCase):
    model = Tag
    brief_fields = ['color', 'description', 'display', 'id', 'name', 'slug', 'url']
    create_data = [
        {
            'name': 'Tag 4',
            'slug': 'tag-4',
            'weight': 1000,
        },
        {
            'name': 'Tag 5',
            'slug': 'tag-5',
        },
        {
            'name': 'Tag 6',
            'slug': 'tag-6',
        },
    ]
    bulk_update_data = {
        'description': 'New description',
    }

    @classmethod
    def setUpTestData(cls):

        tags = (
            Tag(name='Tag 1', slug='tag-1'),
            Tag(name='Tag 2', slug='tag-2'),
            Tag(name='Tag 3', slug='tag-3', weight=26),
        )
        Tag.objects.bulk_create(tags)


class TaggedItemTestCase(
    APIViewTestCases.GetObjectViewTestCase,
    APIViewTestCases.ListObjectsViewTestCase
):
    model = TaggedItem
    brief_fields = ['display', 'id', 'object', 'object_id', 'object_type', 'tag', 'url']

    @classmethod
    def setUpTestData(cls):

        tags = (
            Tag(name='Tag 1', slug='tag-1'),
            Tag(name='Tag 2', slug='tag-2'),
            Tag(name='Tag 3', slug='tag-3'),
        )
        Tag.objects.bulk_create(tags)

        sites = (
            Site(name='Site 1', slug='site-1'),
            Site(name='Site 2', slug='site-2'),
            Site(name='Site 3', slug='site-3'),
        )
        Site.objects.bulk_create(sites)
        sites[0].tags.set([tags[0], tags[1]])
        sites[1].tags.set([tags[1], tags[2]])
        sites[2].tags.set([tags[2], tags[0]])


# TODO: Standardize to APIViewTestCase (needs create & update tests)
class ImageAttachmentTestCase(
    APIViewTestCases.GetObjectViewTestCase,
    APIViewTestCases.ListObjectsViewTestCase,
    APIViewTestCases.DeleteObjectViewTestCase,
    APIViewTestCases.GraphQLTestCase
):
    model = ImageAttachment
    brief_fields = ['description', 'display', 'id', 'image', 'name', 'url']

    @classmethod
    def setUpTestData(cls):
        ct = ContentType.objects.get_for_model(Site)

        site = Site.objects.create(name='Site 1', slug='site-1')

        image_attachments = (
            ImageAttachment(
                object_type=ct,
                object_id=site.pk,
                name='Image Attachment 1',
                image='http://example.com/image1.png',
                image_height=100,
                image_width=100,
                image_size=1024
            ),
            ImageAttachment(
                object_type=ct,
                object_id=site.pk,
                name='Image Attachment 2',
                image='http://example.com/image2.png',
                image_height=100,
                image_width=100,
                image_size=2048
            ),
            ImageAttachment(
                object_type=ct,
                object_id=site.pk,
                name='Image Attachment 3',
                image='http://example.com/image3.png',
                image_height=100,
                image_width=100,
                image_size=4096
            )
        )
        ImageAttachment.objects.bulk_create(image_attachments)


class JournalEntryTestCase(APIViewTestCases.APIViewTestCase):
    model = JournalEntry
    brief_fields = ['created', 'display', 'id', 'url']
    bulk_update_data = {
        'comments': 'Overwritten',
    }

    @classmethod
    def setUpTestData(cls):
        users = (
            User(username='User 1'),
            User(username='User 2'),
        )
        User.objects.bulk_create(users)

        user = User.objects.first()
        site = Site.objects.create(name='Site 1', slug='site-1')

        journal_entries = (
            JournalEntry(
                created_by=user,
                assigned_object=site,
                comments='Fourth entry',
            ),
            JournalEntry(
                created_by=user,
                assigned_object=site,
                comments='Fifth entry',
            ),
            JournalEntry(
                created_by=user,
                assigned_object=site,
                comments='Sixth entry',
            ),
        )
        JournalEntry.objects.bulk_create(journal_entries)

        cls.create_data = [
            {
                'assigned_object_type': 'dcim.site',
                'assigned_object_id': site.pk,
                'comments': 'First entry',
            },
            {
                'assigned_object_type': 'dcim.site',
                'assigned_object_id': site.pk,
                'comments': 'Second entry',
            },
            {
                'assigned_object_type': 'dcim.site',
                'assigned_object_id': site.pk,
                'comments': 'Third entry',
            },
        ]

    def test_immutable_created_by(self):
        """
        Verify that created_by can't be changed for existing objects
        """
        entry = JournalEntry.objects.first()
        created_by_before = entry.created_by_id
        # select user different from the one currently set
        change_user = User.objects.exclude(pk=created_by_before).only('id').first()

        url = reverse('extras-api:journalentry-detail', kwargs={'pk': entry.pk})
        self.add_permissions('extras.change_journalentry')
        response = self.client.patch(url, {'created_by': change_user.id}, format='json', **self.header)

        self.assertEqual(response.status_code, 200)

        entry.refresh_from_db()
        created_by_after = entry.created_by_id
        self.assertEqual(created_by_before, created_by_after)


class ConfigContextProfileTestCase(APIViewTestCases.APIViewTestCase):
    model = ConfigContextProfile
    brief_fields = ['description', 'display', 'id', 'name', 'url']
    create_data = [
        {
            'name': 'Config Context Profile 4',
        },
        {
            'name': 'Config Context Profile 5',
        },
        {
            'name': 'Config Context Profile 6',
        },
    ]
    bulk_update_data = {
        'description': 'New description',
    }

    @classmethod
    def setUpTestData(cls):
        profiles = (
            ConfigContextProfile(
                name='Config Context Profile 1',
                schema={
                    "properties": {
                        "foo": {
                            "type": "string"
                        }
                    },
                    "required": [
                        "foo"
                    ]
                }
            ),
            ConfigContextProfile(
                name='Config Context Profile 2',
                schema={
                    "properties": {
                        "bar": {
                            "type": "string"
                        }
                    },
                    "required": [
                        "bar"
                    ]
                }
            ),
            ConfigContextProfile(
                name='Config Context Profile 3',
                schema={
                    "properties": {
                        "baz": {
                            "type": "string"
                        }
                    },
                    "required": [
                        "baz"
                    ]
                }
            ),
        )
        ConfigContextProfile.objects.bulk_create(profiles)

    def test_update_data_source_and_data_file(self):
        """
        Regression test: Ensure data_source and data_file can be assigned via the API.

        This specifically covers PATCHing a ConfigContext with integer IDs for both fields.
        """
        self.add_permissions(
            'core.view_datafile',
            'core.view_datasource',
            'extras.view_configcontextprofile',
            'extras.change_configcontextprofile',
        )
        config_context_profile = ConfigContextProfile.objects.first()

        # Create a data source and file
        datasource = DataSource.objects.create(
            name='Data Source 1',
            type='local',
            source_url='file:///tmp/netbox-datasource/',
        )
        # Generate a valid dummy YAML file
        file_data = b'profile: configcontext\n'
        datafile = DataFile.objects.create(
            source=datasource,
            path='dir1/file1.yml',
            last_updated=now(),
            size=len(file_data),
            hash=hashlib.sha256(file_data).hexdigest(),
            data=file_data,
        )

        url = self._get_detail_url(config_context_profile)
        payload = {
            'data_source': datasource.pk,
            'data_file': datafile.pk,
        }
        response = self.client.patch(url, payload, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)

        config_context_profile.refresh_from_db()
        self.assertEqual(config_context_profile.data_source_id, datasource.pk)
        self.assertEqual(config_context_profile.data_file_id, datafile.pk)
        self.assertEqual(response.data['data_source']['id'], datasource.pk)
        self.assertEqual(response.data['data_file']['id'], datafile.pk)


class ConfigContextTestCase(APIViewTestCases.APIViewTestCase):
    model = ConfigContext
    brief_fields = ['description', 'display', 'id', 'name', 'url']
    create_data = [
        {
            'name': 'Config Context 4',
            'data': {'more_foo': True},
        },
        {
            'name': 'Config Context 5',
            'data': {'more_bar': False},
        },
        {
            'name': 'Config Context 6',
            'data': {'more_baz': None},
        },
    ]
    bulk_update_data = {
        'description': 'New description',
    }

    @classmethod
    def setUpTestData(cls):

        config_contexts = (
            ConfigContext(name='Config Context 1', weight=100, data={'foo': 123}),
            ConfigContext(name='Config Context 2', weight=200, data={'bar': 456}),
            ConfigContext(name='Config Context 3', weight=300, data={'baz': 789}),
        )
        ConfigContext.objects.bulk_create(config_contexts)

    def test_render_configcontext_for_object(self):
        """
        Test rendering config context data for a device.
        """
        manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
        devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1', slug='device-type-1')
        role = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1')
        site = Site.objects.create(name='Site-1', slug='site-1')
        device = Device.objects.create(name='Device 1', device_type=devicetype, role=role, site=site)

        # Test default config contexts (created at test setup)
        rendered_context = device.get_config_context()
        self.assertEqual(rendered_context['foo'], 123)
        self.assertEqual(rendered_context['bar'], 456)
        self.assertEqual(rendered_context['baz'], 789)

        # Add another context specific to the site
        configcontext4 = ConfigContext(
            name='Config Context 4',
            data={'site_data': 'ABC'}
        )
        configcontext4.save()
        configcontext4.sites.add(site)
        rendered_context = device.get_config_context()
        self.assertEqual(rendered_context['site_data'], 'ABC')

        # Override one of the default contexts
        configcontext5 = ConfigContext(
            name='Config Context 5',
            weight=2000,
            data={'foo': 999}
        )
        configcontext5.save()
        configcontext5.sites.add(site)
        rendered_context = device.get_config_context()
        self.assertEqual(rendered_context['foo'], 999)

        # Add a context which does NOT match our device and ensure it does not apply
        site2 = Site.objects.create(name='Site 2', slug='site-2')
        configcontext6 = ConfigContext(
            name='Config Context 6',
            weight=2000,
            data={'bar': 999}
        )
        configcontext6.save()
        configcontext6.sites.add(site2)
        rendered_context = device.get_config_context()
        self.assertEqual(rendered_context['bar'], 456)

    def test_update_data_source_and_data_file(self):
        """
        Regression test: Ensure data_source and data_file can be assigned via the API.

        This specifically covers PATCHing a ConfigContext with integer IDs for both fields.
        """
        self.add_permissions(
            'core.view_datafile',
            'core.view_datasource',
            'extras.view_configcontext',
            'extras.change_configcontext',
        )
        config_context = ConfigContext.objects.first()

        # Create a data source and file
        datasource = DataSource.objects.create(
            name='Data Source 1',
            type='local',
            source_url='file:///tmp/netbox-datasource/',
        )
        # Generate a valid dummy YAML file
        file_data = b'context: config\n'
        datafile = DataFile.objects.create(
            source=datasource,
            path='dir1/file1.yml',
            last_updated=now(),
            size=len(file_data),
            hash=hashlib.sha256(file_data).hexdigest(),
            data=file_data,
        )

        url = self._get_detail_url(config_context)
        payload = {
            'data_source': datasource.pk,
            'data_file': datafile.pk,
        }
        response = self.client.patch(url, payload, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)

        config_context.refresh_from_db()
        self.assertEqual(config_context.data_source_id, datasource.pk)
        self.assertEqual(config_context.data_file_id, datafile.pk)
        self.assertEqual(response.data['data_source']['id'], datasource.pk)
        self.assertEqual(response.data['data_file']['id'], datafile.pk)


class ConfigTemplateTestCase(APIViewTestCases.APIViewTestCase):
    model = ConfigTemplate
    brief_fields = ['description', 'display', 'id', 'name', 'url']
    create_data = [
        {
            'name': 'Config Template 4',
            'template_code': 'Foo: {{ foo }}',
            'mime_type': 'text/plain',
            'file_name': 'output4',
            'file_extension': 'txt',
            'as_attachment': True,
        },
        {
            'name': 'Config Template 5',
            'template_code': 'Bar: {{ bar }}',
        },
        {
            'name': 'Config Template 6',
            'template_code': 'Baz: {{ baz }}',
        },
    ]
    bulk_update_data = {
        'description': 'New description',
    }

    @classmethod
    def setUpTestData(cls):
        config_templates = (
            ConfigTemplate(
                name='Config Template 1',
                template_code='Foo: {{ foo }}'
            ),
            ConfigTemplate(
                name='Config Template 2',
                template_code='Bar: {{ bar }}',
            ),
            ConfigTemplate(
                name='Config Template 3',
                template_code='Baz: {{ baz }}'
            ),
        )
        ConfigTemplate.objects.bulk_create(config_templates)

    def test_render(self):
        configtemplate = ConfigTemplate.objects.first()

        self.add_permissions('extras.render_configtemplate', 'extras.view_configtemplate')
        url = reverse('extras-api:configtemplate-render', kwargs={'pk': configtemplate.pk})
        response = self.client.post(url, {'foo': 'bar'}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data['content'], 'Foo: bar')

    def test_render_without_permission(self):
        configtemplate = ConfigTemplate.objects.first()

        # No permissions added - user has no render permission
        url = reverse('extras-api:configtemplate-render', kwargs={'pk': configtemplate.pk})
        response = self.client.post(url, {'foo': 'bar'}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)

    def test_render_token_write_enabled(self):
        configtemplate = ConfigTemplate.objects.first()

        self.add_permissions('extras.render_configtemplate', 'extras.view_configtemplate')
        url = reverse('extras-api:configtemplate-render', kwargs={'pk': configtemplate.pk})

        # Request without token auth should fail with PermissionDenied
        response = self.client.post(url, {'foo': 'bar'}, format='json')
        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

        # Create token with write_enabled=False
        token = Token.objects.create(version=2, user=self.user, write_enabled=False)
        token_header = f'Bearer {TOKEN_PREFIX}{token.key}.{token.token}'

        # Request with write-disabled token should fail
        response = self.client.post(url, {'foo': 'bar'}, format='json', HTTP_AUTHORIZATION=token_header)
        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

        # Enable write and retry
        token.write_enabled = True
        token.save()
        response = self.client.post(url, {'foo': 'bar'}, format='json', HTTP_AUTHORIZATION=token_header)
        self.assertHttpStatus(response, status.HTTP_200_OK)


class ScriptTestCase(APITestCase):

    class TestScriptClass(PythonClass):
        class Meta:
            name = 'Test script'
            commit = True
            scheduling_enabled = True

        var1 = StringVar()
        var2 = IntegerVar()
        var3 = BooleanVar()

        def run(self, data, commit=True):
            self.log_info(data['var1'])
            self.log_success(data['var2'])
            self.log_failure(data['var3'])

            return 'Script complete'

    @classmethod
    def setUpTestData(cls):
        # Avoid trying to import a non-existent on-disk module during setup.
        # This test creates the Script row explicitly and monkey-patches
        # Script.python_class below.
        with patch.object(ScriptModule, 'sync_classes'):
            module = ScriptModule.objects.create(
                file_root=ManagedFileRootPathChoices.SCRIPTS,
                file_path='script.py',
            )
        script = Script.objects.create(
            module=module,
            name='Test script',
            is_executable=True,
        )
        cls.url = reverse('extras-api:script-detail', kwargs={'pk': script.pk})

    @property
    def python_class(self):
        return self.TestScriptClass

    def setUp(self):
        super().setUp()
        self.add_permissions('extras.view_script')

        # Monkey-patch the Script model to return our TestScriptClass above
        Script.python_class = self.python_class

        # The script-run endpoint gates on a live RQ worker. Tests run without
        # one, so bypass the check to exercise validation and the enqueue path.
        worker_patch = patch('extras.api.views.any_workers_for_queue', return_value=True)
        worker_patch.start()
        self.addCleanup(worker_patch.stop)

    def test_get_script(self):
        response = self.client.get(self.url, **self.header)

        self.assertEqual(response.data['name'], self.TestScriptClass.Meta.name)
        self.assertEqual(response.data['vars']['var1'], 'StringVar')
        self.assertEqual(response.data['vars']['var2'], 'IntegerVar')
        self.assertEqual(response.data['vars']['var3'], 'BooleanVar')

    def test_schedule_script_past_time_rejected(self):
        """
        Scheduling with past schedule_at should fail.
        """
        self.add_permissions('extras.run_script')

        payload = {
            'data': {'var1': 'hello', 'var2': 1, 'var3': False},
            'commit': True,
            'schedule_at': now() - datetime.timedelta(hours=1),
        }
        response = self.client.post(self.url, payload, format='json', **self.header)

        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertIn('schedule_at', response.data)
        # Be tolerant of exact wording but ensure we failed on schedule_at being in the past
        self.assertIn('future', str(response.data['schedule_at']).lower())

    def test_schedule_script_interval_only(self):
        """
        Interval without schedule_at should auto-set schedule_at now.
        """
        self.add_permissions('extras.run_script')

        payload = {
            'data': {'var1': 'hello', 'var2': 1, 'var3': False},
            'commit': True,
            'interval': 60,
        }
        response = self.client.post(self.url, payload, format='json', **self.header)

        self.assertHttpStatus(response, status.HTTP_200_OK)
        # The latest job is returned in the script detail serializer under "result"
        self.assertIn('result', response.data)
        self.assertEqual(response.data['result']['interval'], 60)
        # Ensure a start time was autopopulated
        self.assertIsNotNone(response.data['result']['scheduled'])

    def test_schedule_script_when_disabled(self):
        """
        Scheduling should fail when script.scheduling_enabled=False.
        """
        self.add_permissions('extras.run_script')

        # Temporarily disable scheduling on the in-test Python class
        original = getattr(self.TestScriptClass.Meta, 'scheduling_enabled', True)
        self.TestScriptClass.Meta.scheduling_enabled = False
        base = {
            'data': {'var1': 'hello', 'var2': 1, 'var3': False},
            'commit': True,
        }
        # Check both schedule_at and interval paths
        cases = [
            {**base, 'schedule_at': now() + datetime.timedelta(minutes=5)},
            {**base, 'interval': 60},
        ]
        try:
            for case in cases:
                with self.subTest(case=list(case.keys())):
                    response = self.client.post(self.url, case, format='json', **self.header)

                    self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
                    # Error should be attached to whichever field we used
                    key = 'schedule_at' if 'schedule_at' in case else 'interval'
                    self.assertIn(key, response.data)
                    self.assertIn('scheduling is not enabled', str(response.data[key]).lower())
        finally:
            # Restore the original setting for other tests
            self.TestScriptClass.Meta.scheduling_enabled = original


class CreatedUpdatedFilterTestCase(APITestCase):

    @classmethod
    def setUpTestData(cls):
        site1 = Site.objects.create(name='Site 1', slug='site-1')
        location1 = Location.objects.create(site=site1, name='Location 1', slug='location-1')
        rackrole1 = RackRole.objects.create(name='Rack Role 1', slug='rack-role-1', color='ff0000')
        racks = (
            Rack(site=site1, location=location1, role=rackrole1, name='Rack 1', u_height=42),
            Rack(site=site1, location=location1, role=rackrole1, name='Rack 2', u_height=42)
        )
        Rack.objects.bulk_create(racks)

        # Change the created and last_updated of the second rack
        Rack.objects.filter(pk=racks[1].pk).update(
            last_updated=make_aware(datetime.datetime(2001, 2, 3, 1, 2, 3, 4)),
            created=make_aware(datetime.datetime(2001, 2, 3))
        )

    def test_get_rack_created(self):
        rack2 = Rack.objects.get(name='Rack 2')
        self.add_permissions('dcim.view_rack')
        url = reverse('dcim-api:rack-list')
        response = self.client.get('{}?created=2001-02-03'.format(url), **self.header)

        self.assertEqual(response.data['count'], 1)
        self.assertEqual(response.data['results'][0]['id'], rack2.pk)

    def test_get_rack_created_gte(self):
        rack1 = Rack.objects.get(name='Rack 1')
        self.add_permissions('dcim.view_rack')
        url = reverse('dcim-api:rack-list')
        response = self.client.get('{}?created__gte=2001-02-04'.format(url), **self.header)

        self.assertEqual(response.data['count'], 1)
        self.assertEqual(response.data['results'][0]['id'], rack1.pk)

    def test_get_rack_created_lte(self):
        rack2 = Rack.objects.get(name='Rack 2')
        self.add_permissions('dcim.view_rack')
        url = reverse('dcim-api:rack-list')
        response = self.client.get('{}?created__lte=2001-02-04'.format(url), **self.header)

        self.assertEqual(response.data['count'], 1)
        self.assertEqual(response.data['results'][0]['id'], rack2.pk)

    def test_get_rack_last_updated(self):
        rack2 = Rack.objects.get(name='Rack 2')
        self.add_permissions('dcim.view_rack')
        url = reverse('dcim-api:rack-list')
        response = self.client.get('{}?last_updated=2001-02-03%2001:02:03.000004'.format(url), **self.header)

        self.assertEqual(response.data['count'], 1)
        self.assertEqual(response.data['results'][0]['id'], rack2.pk)

    def test_get_rack_last_updated_gte(self):
        rack1 = Rack.objects.get(name='Rack 1')
        self.add_permissions('dcim.view_rack')
        url = reverse('dcim-api:rack-list')
        response = self.client.get('{}?last_updated__gte=2001-02-04%2001:02:03.000004'.format(url), **self.header)

        self.assertEqual(response.data['count'], 1)
        self.assertEqual(response.data['results'][0]['id'], rack1.pk)

    def test_get_rack_last_updated_lte(self):
        rack2 = Rack.objects.get(name='Rack 2')
        self.add_permissions('dcim.view_rack')
        url = reverse('dcim-api:rack-list')
        response = self.client.get('{}?last_updated__lte=2001-02-04%2001:02:03.000004'.format(url), **self.header)

        self.assertEqual(response.data['count'], 1)
        self.assertEqual(response.data['results'][0]['id'], rack2.pk)


class SubscriptionTestCase(APIViewTestCases.APIViewTestCase):
    model = Subscription
    brief_fields = ['display', 'id', 'object_id', 'object_type', 'url', 'user']
    graphql_filter = {
        'id': {'lookup': 'gt', 'value': '0'},
    }

    @classmethod
    def setUpTestData(cls):
        users = (
            User(username='User 1'),
            User(username='User 2'),
            User(username='User 3'),
            User(username='User 4'),
        )
        User.objects.bulk_create(users)
        sites = (
            Site(name='Site 1', slug='site-1'),
            Site(name='Site 2', slug='site-2'),
            Site(name='Site 3', slug='site-3'),
        )
        Site.objects.bulk_create(sites)

        subscriptions = (
            Subscription(
                object=sites[0],
                user=users[0],
            ),
            Subscription(
                object=sites[1],
                user=users[1],
            ),
            Subscription(
                object=sites[2],
                user=users[2],
            ),
        )
        Subscription.objects.bulk_create(subscriptions)

        cls.create_data = [
            {
                'object_type': 'dcim.site',
                'object_id': sites[0].pk,
                'user': users[3].pk,
            },
            {
                'object_type': 'dcim.site',
                'object_id': sites[1].pk,
                'user': users[3].pk,
            },
            {
                'object_type': 'dcim.site',
                'object_id': sites[2].pk,
                'user': users[3].pk,
            },
        ]

        cls.bulk_update_data = {
            'user': users[3].pk,
        }


class NotificationGroupTestCase(APIViewTestCases.APIViewTestCase):
    model = NotificationGroup
    brief_fields = ['description', 'display', 'id', 'name', 'url']
    create_data = [
        {
            'object_types': ['dcim.site'],
            'name': 'Custom Link 4',
            'enabled': True,
            'link_text': 'Link 4',
            'link_url': 'http://example.com/?4',
        },
        {
            'object_types': ['dcim.site'],
            'name': 'Custom Link 5',
            'enabled': True,
            'link_text': 'Link 5',
            'link_url': 'http://example.com/?5',
        },
        {
            'object_types': ['dcim.site'],
            'name': 'Custom Link 6',
            'enabled': False,
            'link_text': 'Link 6',
            'link_url': 'http://example.com/?6',
        },
    ]
    bulk_update_data = {
        'description': 'New description',
    }

    @classmethod
    def setUpTestData(cls):
        users = (
            User(username='User 1'),
            User(username='User 2'),
            User(username='User 3'),
        )
        User.objects.bulk_create(users)
        groups = (
            Group(name='Group 1'),
            Group(name='Group 2'),
            Group(name='Group 3'),
        )
        Group.objects.bulk_create(groups)

        notification_groups = (
            NotificationGroup(name='Notification Group 1'),
            NotificationGroup(name='Notification Group 2'),
            NotificationGroup(name='Notification Group 3'),
        )
        NotificationGroup.objects.bulk_create(notification_groups)
        for i, notification_group in enumerate(notification_groups):
            notification_group.users.add(users[i])
            notification_group.groups.add(groups[i])

        cls.create_data = [
            {
                'name': 'Notification Group 4',
                'description': 'Foo',
                'users': [users[0].pk],
                'groups': [groups[0].pk],
            },
            {
                'name': 'Notification Group 5',
                'description': 'Bar',
                'users': [users[1].pk],
                'groups': [groups[1].pk],
            },
            {
                'name': 'Notification Group 6',
                'description': 'Baz',
                'users': [users[2].pk],
                'groups': [groups[2].pk],
            },
        ]


class NotificationTestCase(APIViewTestCases.APIViewTestCase):
    model = Notification
    brief_fields = ['display', 'event_type', 'id', 'object_id', 'object_type', 'read', 'url', 'user']
    bulk_update_data = {
        'read': now(),
    }
    graphql_filter = {
        'event_type': {'lookup': 'exact', 'value': OBJECT_CREATED},
    }

    @classmethod
    def setUpTestData(cls):
        users = (
            User(username='User 1'),
            User(username='User 2'),
            User(username='User 3'),
            User(username='User 4'),
        )
        User.objects.bulk_create(users)
        sites = (
            Site(name='Site 1', slug='site-1'),
            Site(name='Site 2', slug='site-2'),
            Site(name='Site 3', slug='site-3'),
        )
        Site.objects.bulk_create(sites)

        notifications = (
            Notification(
                object=sites[0],
                event_type=OBJECT_CREATED,
                user=users[0],
            ),
            Notification(
                object=sites[1],
                event_type=OBJECT_UPDATED,
                user=users[1],
            ),
            Notification(
                object=sites[2],
                event_type=OBJECT_DELETED,
                user=users[2],
            ),
        )
        Notification.objects.bulk_create(notifications)

        cls.create_data = [
            {
                'object_type': 'dcim.site',
                'object_id': sites[0].pk,
                'user': users[3].pk,
                'event_type': OBJECT_CREATED,
            },
            {
                'object_type': 'dcim.site',
                'object_id': sites[1].pk,
                'user': users[3].pk,
                'event_type': OBJECT_UPDATED,
            },
            {
                'object_type': 'dcim.site',
                'object_id': sites[2].pk,
                'user': users[3].pk,
                'event_type': OBJECT_DELETED,
            },
        ]


class _InMemoryScriptStorage:
    """Stateful stand-in for the scripts storage backend; mirrors its allow_overwrite option."""

    def __init__(self, allow_overwrite=True):
        self.files = {}
        self.allow_overwrite = allow_overwrite

    def save(self, name, content):
        if not self.allow_overwrite and name in self.files:
            name = f'{name}.1'
        content.seek(0)
        self.files[name] = content.read()
        return name

    def open(self, name, mode='rb'):
        if name not in self.files:
            raise FileNotFoundError(name)
        return io.BytesIO(self.files[name])

    def delete(self, name):
        self.files.pop(name, None)

    def exists(self, name):
        return name in self.files


class ScriptModuleTestCase(APITestCase):
    """
    Tests for the POST /api/extras/scripts/upload/ endpoint.

    ScriptModule is a proxy of core.ManagedFile (a different app) so the standard
    APIViewTestCases mixins cannot be used directly. All tests use add_permissions()
    with explicit Django model-level permissions.
    """

    def setUp(self):
        super().setUp()
        self.url = reverse('extras-api:scriptmodule-list')  # /api/extras/scripts/upload/

    @contextmanager
    def _patched_script_storage(self, fake_storage):
        """Patch both storage entry points (serializer writes, module imports) to the given fake."""
        with (
            patch('extras.api.serializers_.scripts.storages') as serializer_storages,
            patch('extras.models.mixins.storages') as module_storages,
        ):
            serializer_storages.create_storage.return_value = fake_storage
            serializer_storages.backends = {'scripts': {}}
            module_storages.__getitem__.return_value = fake_storage
            yield

    def test_upload_script_module_without_permission(self):
        script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n    pass\n"
        upload_file = SimpleUploadedFile('test_upload.py', script_content, content_type='text/plain')
        response = self.client.post(
            self.url,
            {'file': upload_file},
            format='multipart',
            **self.header,
        )
        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)

    def test_upload_script_module(self):
        # ScriptModule is a proxy of core.ManagedFile; both permissions required.
        self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
        script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n    pass\n"
        upload_file = SimpleUploadedFile('test_upload.py', script_content, content_type='text/plain')
        mock_storage = MagicMock()
        mock_storage.save.return_value = 'test_upload.py'

        # The upload serializer writes the file via storages.create_storage(...).save(),
        # but ScriptModule.sync_classes() later imports it via storages["scripts"].open().
        # Provide both behaviors so the uploaded module can actually be loaded during the test.
        mock_storage.open.side_effect = lambda *args, **kwargs: io.BytesIO(script_content)

        with (
            patch('extras.api.serializers_.scripts.storages') as mock_serializer_storages,
            patch('extras.models.mixins.storages') as mock_module_storages,
        ):
            mock_serializer_storages.create_storage.return_value = mock_storage
            mock_serializer_storages.backends = {'scripts': {}}
            mock_module_storages.__getitem__.return_value = mock_storage

            response = self.client.post(
                self.url,
                {'file': upload_file},
                format='multipart',
                **self.header,
            )
        self.assertHttpStatus(response, status.HTTP_201_CREATED)
        self.assertEqual(response.data['file_path'], 'test_upload.py')
        mock_storage.save.assert_called_once()
        self.assertTrue(ScriptModule.objects.filter(file_path='test_upload.py').exists())
        self.assertTrue(Script.objects.filter(module__file_path='test_upload.py', name='TestScript').exists())

    def test_upload_faulty_script_module(self):
        """Uploading a script with an import error should return 400 and not create a DB record."""
        self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
        # 'extras.script' is invalid; the correct module is 'extras.scripts'
        script_content = b"from extras.script import Script\nclass TestScript(Script):\n    pass\n"
        upload_file = SimpleUploadedFile('test_faulty.py', script_content, content_type='text/plain')
        response = self.client.post(
            self.url,
            {'file': upload_file},
            format='multipart',
            **self.header,
        )
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertFalse(ScriptModule.objects.filter(file_path='test_faulty.py').exists())

    def test_upload_duplicate_script_module_preserves_existing_file(self):
        """A duplicate-filename upload returns 400 and leaves the existing file unchanged."""
        self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
        original_content = (
            b"from extras.scripts import Script\n\n\n"
            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
        )
        updated_content = original_content.replace(b"'v1'", b"'v2'")

        fake_storage = _InMemoryScriptStorage()

        with (
            patch('extras.api.serializers_.scripts.storages') as mock_serializer_storages,
            patch('extras.models.mixins.storages') as mock_module_storages,
        ):
            mock_serializer_storages.create_storage.return_value = fake_storage
            mock_serializer_storages.backends = {'scripts': {}}
            mock_module_storages.__getitem__.return_value = fake_storage

            # First upload succeeds and writes the file
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_probe.py', original_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            self.assertEqual(fake_storage.files['zz_probe.py'], original_content)

            # Re-uploading the same filename with different content must be rejected
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_probe.py', updated_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
            self.assertIn('already exists', str(response.data))

            # Existing file must survive intact: neither deleted nor overwritten with v2
            self.assertTrue(fake_storage.exists('zz_probe.py'))
            self.assertEqual(fake_storage.files['zz_probe.py'], original_content)

        # Exactly one ScriptModule remains, still pointing at the original file
        self.assertEqual(ScriptModule.objects.filter(file_path='zz_probe.py').count(), 1)

    def test_upload_script_module_without_file_fails(self):
        self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
        response = self.client.post(self.url, {}, format='json', **self.header)
        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_update_script_module(self):
        """A PATCH with a new file replaces the stored content and re-syncs the module's scripts."""
        self.add_permissions(
            'extras.add_scriptmodule', 'core.add_managedfile',
            'extras.change_scriptmodule', 'core.change_managedfile',
        )
        original_content = (
            b"from extras.scripts import Script\n\n\n"
            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
        )
        updated_content = (
            b"from extras.scripts import Script\n\n\n"
            b"class ProbeScriptV2(Script):\n    def run(self, data, commit):\n        return 'v2'\n"
        )
        fake_storage = _InMemoryScriptStorage()

        with self._patched_script_storage(fake_storage):
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_update.py', original_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            module_id = response.data['id']
            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': module_id})

            response = self.client.patch(
                detail_url,
                {'file': SimpleUploadedFile('zz_update.py', updated_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_200_OK)

        self.assertEqual(response.data['id'], module_id)
        self.assertEqual(response.data['file_path'], 'zz_update.py')
        self.assertIsNotNone(response.data['last_updated'])
        self.assertEqual(fake_storage.files['zz_update.py'], updated_content)
        # Script classes are re-synced from the new content
        self.assertFalse(Script.objects.filter(module_id=module_id, name='ProbeScript').exists())
        self.assertTrue(Script.objects.filter(module_id=module_id, name='ProbeScriptV2').exists())
        self.assertEqual(ScriptModule.objects.filter(file_path='zz_update.py').count(), 1)

    def test_update_script_module_without_file_fails(self):
        """A PATCH without a file upload is rejected and leaves the stored content unchanged."""
        self.add_permissions(
            'extras.add_scriptmodule', 'core.add_managedfile',
            'extras.change_scriptmodule', 'core.change_managedfile',
        )
        script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n    pass\n"
        fake_storage = _InMemoryScriptStorage()

        with self._patched_script_storage(fake_storage):
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_nofile.py', script_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': response.data['id']})

            response = self.client.patch(detail_url, {}, format='json', **self.header)

        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertIn('file', response.data)
        self.assertEqual(fake_storage.files['zz_nofile.py'], script_content)

    def test_update_script_module_without_permission(self):
        """A PATCH without change permissions returns 403 and leaves the stored content unchanged."""
        self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
        script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n    pass\n"
        fake_storage = _InMemoryScriptStorage()

        with self._patched_script_storage(fake_storage):
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_forbidden.py', script_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': response.data['id']})

            response = self.client.patch(
                detail_url,
                {'file': SimpleUploadedFile('zz_forbidden.py', script_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )

        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
        self.assertEqual(fake_storage.files['zz_forbidden.py'], script_content)

    def test_update_script_module_by_file_name(self):
        """A module can be updated by addressing it with its file name instead of its numeric ID."""
        self.add_permissions(
            'extras.add_scriptmodule', 'core.add_managedfile',
            'extras.change_scriptmodule', 'core.change_managedfile',
        )
        original_content = (
            b"from extras.scripts import Script\n\n\n"
            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
        )
        updated_content = original_content.replace(b"'v1'", b"'v2'")
        fake_storage = _InMemoryScriptStorage()

        with self._patched_script_storage(fake_storage):
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_by_name.py', original_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            module_id = response.data['id']
            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': 'zz_by_name.py'})

            response = self.client.put(
                detail_url,
                {'file': SimpleUploadedFile('zz_by_name.py', updated_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_200_OK)

        self.assertEqual(response.data['id'], module_id)
        self.assertEqual(fake_storage.files['zz_by_name.py'], updated_content)

    def test_update_script_module_rejects_mismatched_file_name(self):
        """An update whose uploaded file name differs from the module's file path is rejected."""
        self.add_permissions(
            'extras.add_scriptmodule', 'core.add_managedfile',
            'extras.change_scriptmodule', 'core.change_managedfile',
        )
        original_content = (
            b"from extras.scripts import Script\n\n\n"
            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
        )
        fake_storage = _InMemoryScriptStorage()

        with self._patched_script_storage(fake_storage):
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_mismatch.py', original_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': response.data['id']})

            response = self.client.patch(
                detail_url,
                {'file': SimpleUploadedFile('zz_other.py', original_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )

        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(fake_storage.files['zz_mismatch.py'], original_content)
        self.assertNotIn('zz_other.py', fake_storage.files)

    def test_update_script_module_storage_name_mismatch_fails(self):
        """If the storage backend saves under an alternate name, the update is rejected and rolled back."""
        self.add_permissions(
            'extras.add_scriptmodule', 'core.add_managedfile',
            'extras.change_scriptmodule', 'core.change_managedfile',
        )
        original_content = (
            b"from extras.scripts import Script\n\n\n"
            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
        )
        updated_content = original_content.replace(b"'v1'", b"'v2'")
        fake_storage = _InMemoryScriptStorage(allow_overwrite=False)

        with self._patched_script_storage(fake_storage):
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_suffix.py', original_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            module = ScriptModule.objects.get(file_path='zz_suffix.py')
            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': module.pk})

            response = self.client.patch(
                detail_url,
                {'file': SimpleUploadedFile('zz_suffix.py', updated_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )

        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
        # Original file intact, alternate-name orphan removed
        self.assertEqual(fake_storage.files['zz_suffix.py'], original_content)
        self.assertNotIn('zz_suffix.py.1', fake_storage.files)
        module.refresh_from_db()
        self.assertEqual(module.file_path, 'zz_suffix.py')

    def test_update_faulty_script_module_preserves_existing_module(self):
        """An update with invalid script content is rejected before storage or Script rows change."""
        self.add_permissions(
            'extras.add_scriptmodule', 'core.add_managedfile',
            'extras.change_scriptmodule', 'core.change_managedfile',
        )
        original_content = (
            b"from extras.scripts import Script\n\n\n"
            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
        )
        # 'extras.script' is invalid; the correct module is 'extras.scripts'
        faulty_content = b"from extras.script import Script\nclass TestScript(Script):\n    pass\n"
        fake_storage = _InMemoryScriptStorage()

        with self._patched_script_storage(fake_storage):
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_faulty_update.py', original_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            module_id = response.data['id']
            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': module_id})

            response = self.client.patch(
                detail_url,
                {'file': SimpleUploadedFile('zz_faulty_update.py', faulty_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )

        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(fake_storage.files['zz_faulty_update.py'], original_content)
        self.assertTrue(Script.objects.filter(module_id=module_id, name='ProbeScript').exists())

    def test_update_script_module_not_found(self):
        """An update addressing a nonexistent module returns 404 for both lookup styles."""
        self.add_permissions('extras.change_scriptmodule', 'core.change_managedfile')
        for lookup in ('999999', 'zz_missing.py', '½'):
            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': lookup})
            response = self.client.patch(detail_url, {}, format='json', **self.header)
            self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)

    def test_update_script_module_via_put_without_file_fails(self):
        """A PUT without a file upload is rejected by the field-level required check."""
        self.add_permissions(
            'extras.add_scriptmodule', 'core.add_managedfile',
            'extras.change_scriptmodule', 'core.change_managedfile',
        )
        script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n    pass\n"
        fake_storage = _InMemoryScriptStorage()

        with self._patched_script_storage(fake_storage):
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_put_nofile.py', script_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': response.data['id']})

            response = self.client.put(detail_url, {}, format='json', **self.header)

        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertIn('file', response.data)
        self.assertEqual(fake_storage.files['zz_put_nofile.py'], script_content)

    def test_update_script_module_rolls_back_scripts_on_save_failure(self):
        """A failed Script sync during an update rolls back all Script row changes."""
        self.add_permissions(
            'extras.add_scriptmodule', 'core.add_managedfile',
            'extras.change_scriptmodule', 'core.change_managedfile',
        )
        original_content = (
            b"from extras.scripts import Script\n\n\n"
            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
        )
        updated_content = (
            b"from extras.scripts import Script\n\n\n"
            b"class ProbeScriptV2(Script):\n    def run(self, data, commit):\n        return 'v2'\n"
        )
        fake_storage = _InMemoryScriptStorage()

        with self._patched_script_storage(fake_storage):
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_rollback.py', original_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            module_id = response.data['id']
            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': module_id})

            # Fail the sync mid-way: the old Script row is already deleted when create() raises
            with patch(
                'extras.models.scripts.Script.objects.create',
                side_effect=IntegrityError('Simulated database error'),
            ):
                with self.assertRaises(IntegrityError):
                    self.client.patch(
                        detail_url,
                        {'file': SimpleUploadedFile('zz_rollback.py', updated_content, content_type='text/plain')},
                        format='multipart',
                        **self.header,
                    )

        # DB is all-or-nothing; storage keeps the new content and a retried update re-syncs the rows
        self.assertTrue(Script.objects.filter(module_id=module_id, name='ProbeScript').exists())
        self.assertFalse(Script.objects.filter(module_id=module_id, name='ProbeScriptV2').exists())
        self.assertEqual(fake_storage.files['zz_rollback.py'], updated_content)

    def test_update_script_module_with_missing_stored_file(self):
        """An update succeeds when the stored file is missing, re-creating it from the upload."""
        self.add_permissions(
            'extras.add_scriptmodule', 'core.add_managedfile',
            'extras.change_scriptmodule', 'core.change_managedfile',
        )
        original_content = (
            b"from extras.scripts import Script\n\n\n"
            b"class ProbeScript(Script):\n    def run(self, data, commit):\n        return 'v1'\n"
        )
        updated_content = original_content.replace(b"'v1'", b"'v2'")
        fake_storage = _InMemoryScriptStorage()

        with self._patched_script_storage(fake_storage):
            response = self.client.post(
                self.url,
                {'file': SimpleUploadedFile('zz_lost_file.py', original_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )
            self.assertHttpStatus(response, status.HTTP_201_CREATED)
            detail_url = reverse('extras-api:scriptmodule-detail', kwargs={'pk': response.data['id']})

            # Simulate a file removed from storage outside NetBox
            del fake_storage.files['zz_lost_file.py']

            response = self.client.patch(
                detail_url,
                {'file': SimpleUploadedFile('zz_lost_file.py', updated_content, content_type='text/plain')},
                format='multipart',
                **self.header,
            )

        self.assertHttpStatus(response, status.HTTP_200_OK)
        self.assertEqual(fake_storage.files['zz_lost_file.py'], updated_content)