From 5016ed61728a60df6d64e1a461dca1255dc66b59 Mon Sep 17 00:00:00 2001 From: Jason Novinger Date: Fri, 14 Nov 2025 15:09:59 -0600 Subject: [PATCH] Fixes #353: Add dedicated archive endpoint to REST API - Add POST /api/plugins/branching/branches/{id}/archive/ endpoint - Make status field read-only in serializer to prevent unsafe PATCH operations - Add comprehensive test coverage for archive endpoint and status field protection --- netbox_branching/api/serializers.py | 2 +- netbox_branching/api/views.py | 23 ++++++ netbox_branching/tests/test_api.py | 106 ++++++++++++++++++++++++++++ 3 files changed, 130 insertions(+), 1 deletion(-) diff --git a/netbox_branching/api/serializers.py b/netbox_branching/api/serializers.py index a53cd97..8a4391a 100644 --- a/netbox_branching/api/serializers.py +++ b/netbox_branching/api/serializers.py @@ -32,7 +32,7 @@ class BranchSerializer(NetBoxModelSerializer): ) status = ChoiceField( choices=BranchStatusChoices, - required=False + read_only=True ) class Meta: diff --git a/netbox_branching/api/views.py b/netbox_branching/api/views.py index 8d115b3..aac93ce 100644 --- a/netbox_branching/api/views.py +++ b/netbox_branching/api/views.py @@ -112,6 +112,29 @@ def revert(self, request, pk): return Response(JobSerializer(job, context={'request': request}).data) + @extend_schema( + methods=['post'], + responses={200: serializers.BranchSerializer()},) + @action(detail=True, methods=['post']) + def archive(self, request, pk): + """ + Archive a merged branch, deprovisioning its schema. + """ + if not request.user.has_perm('netbox_branching.archive_branch'): + raise PermissionDenied("This user does not have permission to archive branches.") + + branch = self.get_object() + if not branch.merged: + return HttpResponseBadRequest("Only merged branches can be archived.") + if not branch.can_archive: + return HttpResponseBadRequest("Archiving this branch is not permitted.") + + branch.archive(user=request.user) + branch.refresh_from_db() + + serializer = self.get_serializer(branch) + return Response(serializer.data) + class BranchEventViewSet(ListModelMixin, RetrieveModelMixin, BaseViewSet): queryset = BranchEvent.objects.all() diff --git a/netbox_branching/tests/test_api.py b/netbox_branching/tests/test_api.py index af3852c..2a355a8 100644 --- a/netbox_branching/tests/test_api.py +++ b/netbox_branching/tests/test_api.py @@ -99,3 +99,109 @@ def test_with_branch_cookie(self): results = self.get_results(response) self.assertEqual(len(results), 1) self.assertEqual(results[0]['name'], 'Site 2') + + +class BranchArchiveAPITestCase(TransactionTestCase): + serialized_rollback = True + + def setUp(self): + self.client = Client() + self.user = get_user_model().objects.create_user(username='testuser', is_superuser=True) + token = Token(user=self.user) + token.save() + self.header = { + 'HTTP_AUTHORIZATION': f'Token {token.key}', + 'HTTP_ACCEPT': 'application/json', + 'HTTP_CONTENT_TYPE': 'application/json', + } + + ContentType.objects.get_for_model(Branch) + + def test_archive_endpoint_success(self): + branch = Branch(name='Test Branch') + branch.save(provision=False) + branch.provision(self.user) + branch.refresh_from_db() + + from netbox_branching.choices import BranchStatusChoices + Branch.objects.filter(pk=branch.pk).update(status=BranchStatusChoices.MERGED) + branch.refresh_from_db() + self.assertEqual(branch.status, 'merged') + + url = reverse('plugins-api:netbox_branching-api:branch-archive', kwargs={'pk': branch.pk}) + response = self.client.post(url, **self.header) + + self.assertEqual(response.status_code, 200) + data = json.loads(response.content) + self.assertEqual(data['status']['value'], 'archived') + + branch.refresh_from_db() + self.assertEqual(branch.status, 'archived') + + from django.db import connection + with connection.cursor() as cursor: + cursor.execute( + "SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s", + [branch.schema_name] + ) + self.assertIsNone(cursor.fetchone()) + + def test_archive_endpoint_permission_denied(self): + user = get_user_model().objects.create_user(username='limited_user') + token = Token(user=user) + token.save() + header = { + 'HTTP_AUTHORIZATION': f'Token {token.key}', + 'HTTP_ACCEPT': 'application/json', + 'HTTP_CONTENT_TYPE': 'application/json', + } + + branch = Branch(name='Test Branch') + branch.save(provision=False) + branch.provision(self.user) + branch.refresh_from_db() + + from netbox_branching.choices import BranchStatusChoices + Branch.objects.filter(pk=branch.pk).update(status=BranchStatusChoices.MERGED) + branch.refresh_from_db() + + url = reverse('plugins-api:netbox_branching-api:branch-archive', kwargs={'pk': branch.pk}) + response = self.client.post(url, **header) + + self.assertEqual(response.status_code, 403) + + def test_archive_endpoint_not_mergeable(self): + branch = Branch(name='Test Branch') + branch.save(provision=False) + branch.provision(self.user) + + branch.refresh_from_db() + self.assertEqual(branch.status, 'ready') + + url = reverse('plugins-api:netbox_branching-api:branch-archive', kwargs={'pk': branch.pk}) + response = self.client.post(url, **self.header) + + self.assertEqual(response.status_code, 400) + + def test_patch_status_archived_blocked(self): + branch = Branch(name='Test Branch') + branch.save(provision=False) + branch.provision(self.user) + branch.refresh_from_db() + + from netbox_branching.choices import BranchStatusChoices + Branch.objects.filter(pk=branch.pk).update(status=BranchStatusChoices.MERGED) + branch.refresh_from_db() + + url = reverse('plugins-api:netbox_branching-api:branch-detail', kwargs={'pk': branch.pk}) + response = self.client.patch( + url, + data=json.dumps({'status': 'archived'}), + content_type='application/json', + **self.header + ) + + self.assertEqual(response.status_code, 200) + + branch.refresh_from_db() + self.assertEqual(branch.status, 'merged')