fix(api): Add tags field to DataSource serializer

Adds tags field to DataSourceSerializer fields list and updates query
count baseline. Includes comprehensive test coverage for tag
assignment, updates, and clearing operations.

Fixes #23120

netbox/core/api/serializers_/data.py

@@ -26,7 +26,7 @@ class DataSourceSerializer(PrimaryModelSerializer):
         model = DataSource
         fields = [
             'id', 'url', 'display_url', 'display', 'name', 'type', 'source_url', 'enabled', 'status', 'description',
-            'sync_interval', 'parameters', 'ignore_rules', 'owner', 'comments', 'custom_fields', 'created',
+            'sync_interval', 'parameters', 'ignore_rules', 'owner', 'comments', 'tags', 'custom_fields', 'created',
             'last_updated', 'last_synced', 'file_count',
         ]
         brief_fields = ('id', 'url', 'display', 'name', 'description')

netbox/core/tests/query_counts.json

@@ -1,7 +1,7 @@
 {
   "datafile:api_list_objects": 10,
   "datafile:list_objects_with_permission": 17,
-  "datasource:api_list_objects": 11,
+  "datasource:api_list_objects": 12,
   "datasource:list_objects_with_permission": 17,
   "job:api_list_objects": 12,
   "job:list_objects_with_permission": 19

netbox/core/tests/test_api.py

@@ -12,7 +12,7 @@ from rq.registry import FailedJobRegistry, StartedJobRegistry
 
 from users.constants import TOKEN_PREFIX
 from users.models import Token
-from utilities.testing import APITestCase, APIViewTestCases, GraphQLQueryTest, TestCase
+from utilities.testing import APITestCase, APIViewTestCases, GraphQLQueryTest, TestCase, create_tags
 from utilities.testing.mixins import RQQueueTestMixin
 from utilities.testing.utils import disable_logging
 
@@ -100,6 +100,57 @@ class DataSourceTestCase(APIViewTestCases.APIViewTestCase):
             },
         ]
 
+    def test_tags_in_representation(self):
+        """Assigned tags are rendered in the detail representation."""
+        data_source = DataSource.objects.first()
+        data_source.tags.set(create_tags('Alpha'))
+        self.add_permissions('core.view_datasource')
+
+        response = self.client.get(self._get_detail_url(data_source), **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertIn('tags', response.data)
+        self.assertEqual([tag['slug'] for tag in response.data['tags']], ['alpha'])
+
+    def test_create_with_tags(self):
+        """Tags supplied on creation are assigned to the new data source."""
+        create_tags('Alpha')
+        self.add_permissions('core.add_datasource', 'extras.view_tag')
+
+        data = {
+            'name': 'Data Source 7',
+            'type': 'git',
+            'source_url': 'https://example.com/git/source7',
+            'tags': [{'slug': 'alpha'}],
+        }
+        response = self.client.post(self._get_list_url(), data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+
+        data_source = DataSource.objects.get(pk=response.data['id'])
+        self.assertEqual(list(data_source.tags.values_list('slug', flat=True)), ['alpha'])
+
+    def test_update_tags(self):
+        """Tags supplied on update replace the existing assignment."""
+        data_source = DataSource.objects.first()
+        tags = create_tags('Alpha', 'Bravo')
+        data_source.tags.set([tags[0]])
+        self.add_permissions('core.change_datasource', 'extras.view_tag')
+
+        data = {'tags': [{'slug': 'bravo'}]}
+        response = self.client.patch(self._get_detail_url(data_source), data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(list(data_source.tags.values_list('slug', flat=True)), ['bravo'])
+
+    def test_clear_tags(self):
+        """An empty tag list clears the existing assignment."""
+        data_source = DataSource.objects.first()
+        data_source.tags.set(create_tags('Alpha'))
+        self.add_permissions('core.change_datasource')
+
+        data = {'tags': []}
+        response = self.client.patch(self._get_detail_url(data_source), data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(list(data_source.tags.values_list('slug', flat=True)), [])
+
     def assert_only_source_1(self, data):
         """The JSON lookup returns exactly the source carrying the matching value."""
         ids = sorted(result['id'] for result in data['data_source_list'])