Skip to content

Commit 487bfcb

Browse files
committed
feat: add GetDefaultBranch API view and corresponding tests
Signed-off-by: Michal Fiedorowicz <mfiedorowicz@netboxlabs.com>
1 parent 4d45f4e commit 487bfcb

File tree

3 files changed

+193
-1
lines changed

3 files changed

+193
-1
lines changed

netbox_diode_plugin/api/urls.py

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,12 +5,13 @@
55
from django.urls import include, path
66
from netbox.api.routers import NetBoxRouter
77

8-
from .views import ApplyChangeSetView, GenerateDiffView
8+
from .views import ApplyChangeSetView, GenerateDiffView, GetDefaultBranchView
99

1010
router = NetBoxRouter()
1111

1212
urlpatterns = [
1313
path("apply-change-set/", ApplyChangeSetView.as_view()),
1414
path("generate-diff/", GenerateDiffView.as_view()),
15+
path("default-branch/", GetDefaultBranchView.as_view()),
1516
path("", include(router.urls)),
1617
]

netbox_diode_plugin/api/views.py

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -211,3 +211,32 @@ def _post(self, request, *args, **kwargs):
211211
)
212212

213213
return Response(result.to_dict(), status=result.get_status_code())
214+
215+
216+
class GetDefaultBranchView(views.APIView):
217+
"""GetDefaultBranch view."""
218+
219+
authentication_classes = [DiodeOAuth2Authentication]
220+
permission_classes = [IsAuthenticated, require_scopes(SCOPE_NETBOX_READ)]
221+
222+
def get(self, request, *args, **kwargs):
223+
"""Get default branch from settings."""
224+
branch_data = None
225+
226+
# Check for default branch in settings
227+
if Branch is not None:
228+
try:
229+
from netbox_diode_plugin.models import Setting
230+
settings = Setting.objects.first()
231+
if settings and settings.branch:
232+
branch_data = {
233+
"id": settings.branch.schema_id,
234+
"name": settings.branch.name
235+
}
236+
logger.debug(
237+
f"Default branch from settings: {settings.branch.name} ({settings.branch.schema_id})"
238+
)
239+
except Exception as e:
240+
logger.warning(f"Could not retrieve default branch from settings: {e}")
241+
242+
return Response({"branch": branch_data})
Lines changed: 162 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,162 @@
1+
#!/usr/bin/env python
2+
# Copyright 2025 NetBox Labs, Inc.
3+
"""Diode NetBox Plugin - GetDefaultBranch API Tests."""
4+
5+
import logging
6+
from types import SimpleNamespace
7+
from unittest import mock
8+
9+
from rest_framework import status
10+
from utilities.testing import APITestCase
11+
12+
from netbox_diode_plugin.api.authentication import DiodeOAuth2Authentication
13+
from netbox_diode_plugin.models import Setting
14+
from netbox_diode_plugin.plugin_config import get_diode_user
15+
16+
logger = logging.getLogger(__name__)
17+
18+
19+
class GetDefaultBranchViewTestCase(APITestCase):
20+
"""Test cases for GetDefaultBranchView."""
21+
22+
def setUp(self):
23+
"""Set up the test case."""
24+
self.url = "/netbox/api/plugins/diode/default-branch/"
25+
26+
self.authorization_header = {"HTTP_AUTHORIZATION": "Bearer mocked_oauth_token"}
27+
self.diode_user = SimpleNamespace(
28+
user=get_diode_user(),
29+
token_scopes=["netbox:read", "netbox:write"],
30+
token_data={"scope": "netbox:read netbox:write"}
31+
)
32+
33+
self.introspect_patcher = mock.patch.object(
34+
DiodeOAuth2Authentication,
35+
'_introspect_token',
36+
return_value=self.diode_user
37+
)
38+
self.introspect_patcher.start()
39+
40+
def tearDown(self):
41+
"""Clean up after tests."""
42+
self.introspect_patcher.stop()
43+
super().tearDown()
44+
45+
def test_get_default_branch_unauthenticated(self):
46+
"""Test that unauthenticated requests are rejected."""
47+
response = self.client.get(self.url)
48+
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
49+
50+
def test_get_default_branch_without_read_scope(self):
51+
"""Test that requests without netbox:read scope are rejected."""
52+
# Mock user with only write scope
53+
user_without_read = SimpleNamespace(
54+
user=get_diode_user(),
55+
token_scopes=["netbox:write"],
56+
token_data={"scope": "netbox:write"}
57+
)
58+
59+
with mock.patch.object(
60+
DiodeOAuth2Authentication,
61+
'_introspect_token',
62+
return_value=user_without_read
63+
):
64+
response = self.client.get(self.url, **self.authorization_header)
65+
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
66+
67+
def test_get_default_branch_no_branching_plugin(self):
68+
"""Test response when branching plugin is not installed."""
69+
# Create a setting without branch
70+
Setting.objects.create(diode_target="grpc://localhost:8080/diode")
71+
72+
# Mock Branch as None (simulating plugin not installed)
73+
with mock.patch('netbox_diode_plugin.api.views.Branch', None):
74+
response = self.client.get(self.url, **self.authorization_header)
75+
76+
self.assertEqual(response.status_code, status.HTTP_200_OK)
77+
self.assertIn("branch", response.json())
78+
self.assertIsNone(response.json()["branch"])
79+
80+
def test_get_default_branch_no_settings(self):
81+
"""Test response when no settings exist."""
82+
# Ensure no settings exist
83+
Setting.objects.all().delete()
84+
85+
response = self.client.get(self.url, **self.authorization_header)
86+
87+
self.assertEqual(response.status_code, status.HTTP_200_OK)
88+
self.assertIn("branch", response.json())
89+
self.assertIsNone(response.json()["branch"])
90+
91+
def test_get_default_branch_settings_without_branch(self):
92+
"""Test response when settings exist but branch is not set."""
93+
# Create a setting without branch
94+
Setting.objects.create(diode_target="grpc://localhost:8080/diode", branch_id=None)
95+
96+
response = self.client.get(self.url, **self.authorization_header)
97+
98+
self.assertEqual(response.status_code, status.HTTP_200_OK)
99+
self.assertIn("branch", response.json())
100+
self.assertIsNone(response.json()["branch"])
101+
102+
def test_get_default_branch_with_branching_plugin_and_branch_set(self):
103+
"""Test response when branching plugin is installed and branch is set."""
104+
# Create a mock Branch object
105+
mock_branch = mock.Mock()
106+
mock_branch.schema_id = "branch-123"
107+
mock_branch.name = "main"
108+
mock_branch.id = 1
109+
110+
# Create a setting with branch_id
111+
setting = Setting.objects.create(
112+
diode_target="grpc://localhost:8080/diode",
113+
branch_id=1
114+
)
115+
116+
# Mock the Branch model and query
117+
mock_branch_model = mock.Mock()
118+
mock_branch_model.objects.get.return_value = mock_branch
119+
120+
with mock.patch('netbox_diode_plugin.api.views.Branch', mock_branch_model):
121+
with mock.patch.object(Setting, 'branch', new_callable=mock.PropertyMock) as mock_branch_property:
122+
mock_branch_property.return_value = mock_branch
123+
124+
response = self.client.get(self.url, **self.authorization_header)
125+
126+
self.assertEqual(response.status_code, status.HTTP_200_OK)
127+
self.assertIn("branch", response.json())
128+
self.assertIsNotNone(response.json()["branch"])
129+
self.assertEqual(response.json()["branch"]["id"], "branch-123")
130+
self.assertEqual(response.json()["branch"]["name"], "main")
131+
132+
def test_get_default_branch_exception_handling(self):
133+
"""Test that exceptions during branch retrieval are handled gracefully."""
134+
# Create a setting with branch_id
135+
setting = Setting.objects.create(
136+
diode_target="grpc://localhost:8080/diode",
137+
branch_id=1
138+
)
139+
140+
# Mock Branch model to exist but raise exception on query
141+
mock_branch_model = mock.Mock()
142+
143+
with mock.patch('netbox_diode_plugin.api.views.Branch', mock_branch_model):
144+
with mock.patch.object(Setting, 'branch', new_callable=mock.PropertyMock) as mock_branch_property:
145+
# Simulate an exception when accessing the branch property
146+
mock_branch_property.side_effect = Exception("Database error")
147+
148+
response = self.client.get(self.url, **self.authorization_header)
149+
150+
# Should return 200 with null branch due to exception handling
151+
self.assertEqual(response.status_code, status.HTTP_200_OK)
152+
self.assertIn("branch", response.json())
153+
self.assertIsNone(response.json()["branch"])
154+
155+
def test_get_default_branch_with_valid_authentication(self):
156+
"""Test that authenticated requests with proper scope are successful."""
157+
response = self.client.get(self.url, **self.authorization_header)
158+
159+
self.assertEqual(response.status_code, status.HTTP_200_OK)
160+
self.assertIn("branch", response.json())
161+
# Response structure is correct even if branch is None
162+
self.assertIsInstance(response.json(), dict)

0 commit comments

Comments
 (0)