Skip to content

Commit 5016ed6

Browse files
committed
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
1 parent 7dfa0bb commit 5016ed6

File tree

3 files changed

+130
-1
lines changed

3 files changed

+130
-1
lines changed

netbox_branching/api/serializers.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ class BranchSerializer(NetBoxModelSerializer):
3232
)
3333
status = ChoiceField(
3434
choices=BranchStatusChoices,
35-
required=False
35+
read_only=True
3636
)
3737

3838
class Meta:

netbox_branching/api/views.py

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,29 @@ def revert(self, request, pk):
112112

113113
return Response(JobSerializer(job, context={'request': request}).data)
114114

115+
@extend_schema(
116+
methods=['post'],
117+
responses={200: serializers.BranchSerializer()},)
118+
@action(detail=True, methods=['post'])
119+
def archive(self, request, pk):
120+
"""
121+
Archive a merged branch, deprovisioning its schema.
122+
"""
123+
if not request.user.has_perm('netbox_branching.archive_branch'):
124+
raise PermissionDenied("This user does not have permission to archive branches.")
125+
126+
branch = self.get_object()
127+
if not branch.merged:
128+
return HttpResponseBadRequest("Only merged branches can be archived.")
129+
if not branch.can_archive:
130+
return HttpResponseBadRequest("Archiving this branch is not permitted.")
131+
132+
branch.archive(user=request.user)
133+
branch.refresh_from_db()
134+
135+
serializer = self.get_serializer(branch)
136+
return Response(serializer.data)
137+
115138

116139
class BranchEventViewSet(ListModelMixin, RetrieveModelMixin, BaseViewSet):
117140
queryset = BranchEvent.objects.all()

netbox_branching/tests/test_api.py

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,3 +99,109 @@ def test_with_branch_cookie(self):
9999
results = self.get_results(response)
100100
self.assertEqual(len(results), 1)
101101
self.assertEqual(results[0]['name'], 'Site 2')
102+
103+
104+
class BranchArchiveAPITestCase(TransactionTestCase):
105+
serialized_rollback = True
106+
107+
def setUp(self):
108+
self.client = Client()
109+
self.user = get_user_model().objects.create_user(username='testuser', is_superuser=True)
110+
token = Token(user=self.user)
111+
token.save()
112+
self.header = {
113+
'HTTP_AUTHORIZATION': f'Token {token.key}',
114+
'HTTP_ACCEPT': 'application/json',
115+
'HTTP_CONTENT_TYPE': 'application/json',
116+
}
117+
118+
ContentType.objects.get_for_model(Branch)
119+
120+
def test_archive_endpoint_success(self):
121+
branch = Branch(name='Test Branch')
122+
branch.save(provision=False)
123+
branch.provision(self.user)
124+
branch.refresh_from_db()
125+
126+
from netbox_branching.choices import BranchStatusChoices
127+
Branch.objects.filter(pk=branch.pk).update(status=BranchStatusChoices.MERGED)
128+
branch.refresh_from_db()
129+
self.assertEqual(branch.status, 'merged')
130+
131+
url = reverse('plugins-api:netbox_branching-api:branch-archive', kwargs={'pk': branch.pk})
132+
response = self.client.post(url, **self.header)
133+
134+
self.assertEqual(response.status_code, 200)
135+
data = json.loads(response.content)
136+
self.assertEqual(data['status']['value'], 'archived')
137+
138+
branch.refresh_from_db()
139+
self.assertEqual(branch.status, 'archived')
140+
141+
from django.db import connection
142+
with connection.cursor() as cursor:
143+
cursor.execute(
144+
"SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s",
145+
[branch.schema_name]
146+
)
147+
self.assertIsNone(cursor.fetchone())
148+
149+
def test_archive_endpoint_permission_denied(self):
150+
user = get_user_model().objects.create_user(username='limited_user')
151+
token = Token(user=user)
152+
token.save()
153+
header = {
154+
'HTTP_AUTHORIZATION': f'Token {token.key}',
155+
'HTTP_ACCEPT': 'application/json',
156+
'HTTP_CONTENT_TYPE': 'application/json',
157+
}
158+
159+
branch = Branch(name='Test Branch')
160+
branch.save(provision=False)
161+
branch.provision(self.user)
162+
branch.refresh_from_db()
163+
164+
from netbox_branching.choices import BranchStatusChoices
165+
Branch.objects.filter(pk=branch.pk).update(status=BranchStatusChoices.MERGED)
166+
branch.refresh_from_db()
167+
168+
url = reverse('plugins-api:netbox_branching-api:branch-archive', kwargs={'pk': branch.pk})
169+
response = self.client.post(url, **header)
170+
171+
self.assertEqual(response.status_code, 403)
172+
173+
def test_archive_endpoint_not_mergeable(self):
174+
branch = Branch(name='Test Branch')
175+
branch.save(provision=False)
176+
branch.provision(self.user)
177+
178+
branch.refresh_from_db()
179+
self.assertEqual(branch.status, 'ready')
180+
181+
url = reverse('plugins-api:netbox_branching-api:branch-archive', kwargs={'pk': branch.pk})
182+
response = self.client.post(url, **self.header)
183+
184+
self.assertEqual(response.status_code, 400)
185+
186+
def test_patch_status_archived_blocked(self):
187+
branch = Branch(name='Test Branch')
188+
branch.save(provision=False)
189+
branch.provision(self.user)
190+
branch.refresh_from_db()
191+
192+
from netbox_branching.choices import BranchStatusChoices
193+
Branch.objects.filter(pk=branch.pk).update(status=BranchStatusChoices.MERGED)
194+
branch.refresh_from_db()
195+
196+
url = reverse('plugins-api:netbox_branching-api:branch-detail', kwargs={'pk': branch.pk})
197+
response = self.client.patch(
198+
url,
199+
data=json.dumps({'status': 'archived'}),
200+
content_type='application/json',
201+
**self.header
202+
)
203+
204+
self.assertEqual(response.status_code, 200)
205+
206+
branch.refresh_from_db()
207+
self.assertEqual(branch.status, 'merged')

0 commit comments

Comments
 (0)