netbox/netbox/extras/tests/test_api.py

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

from django.contrib.contenttypes.models import ContentType
from django.core.files.uploadedfile import SimpleUploadedFile
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):
        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',
            },
        ]


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; mimics allow_overwrite=True."""

    def __init__(self):
        self.files = {}

    def save(self, name, content):
        content.seek(0)
        self.files[name] = content.read()
        return name

    def open(self, name, mode='rb'):
        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/

    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)