diff --git a/netbox/netbox/middleware.py b/netbox/netbox/middleware.py index a9ee93fc6..28167683b 100644 --- a/netbox/netbox/middleware.py +++ b/netbox/netbox/middleware.py @@ -105,8 +105,8 @@ class CoreMiddleware: if settings.DEBUG: return None - # Cleanly handle exceptions that occur from REST API requests - if is_api_request(request): + # Cleanly handle exceptions that occur from REST or GraphQL API requests + if is_api_request(request) or is_graphql_request(request): # Fire Django's got_request_exception signal so error-tracking # integrations (e.g. Sentry) capture the exception. got_request_exception.send(sender=self.__class__, request=request) @@ -283,7 +283,7 @@ class MaintenanceModeMiddleware: error_message = 'NetBox is currently operating in maintenance mode and is unable to perform write ' \ 'operations. Please try again later.' - if is_api_request(request): + if is_api_request(request) or is_graphql_request(request): return handle_rest_api_exception(request, error=error_message) messages.error(request, error_message) diff --git a/netbox/netbox/tests/test_middleware.py b/netbox/netbox/tests/test_middleware.py new file mode 100644 index 000000000..748150c2a --- /dev/null +++ b/netbox/netbox/tests/test_middleware.py @@ -0,0 +1,158 @@ +import json +from contextlib import contextmanager +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from django.core.signals import got_request_exception +from django.db.utils import InternalError +from django.test import RequestFactory, override_settings +from django.urls import reverse +from rest_framework import status + +from netbox.middleware import CoreMiddleware, MaintenanceModeMiddleware +from utilities.testing import TestCase + + +class CoreMiddlewareTestCase(TestCase): + + def setUp(self): + super().setUp() + + self.factory = RequestFactory() + self.middleware = CoreMiddleware(lambda request: None) + self.maintenance_mode_middleware = MaintenanceModeMiddleware(lambda request: None) + + @contextmanager + def capture_request_exception_signal(self): + captured_requests = [] + + def receiver(sender, request, **kwargs): + captured_requests.append(request) + + got_request_exception.connect(receiver, sender=CoreMiddleware, weak=False) + + try: + yield captured_requests + finally: + got_request_exception.disconnect(receiver, sender=CoreMiddleware) + + def process_runtime_error(self, request, message='Test exception'): + """ + Call CoreMiddleware.process_exception() from inside an active exception + handler. handle_rest_api_exception() uses sys.exc_info(), so calling this + inside an except block is important for the JSON response body. + """ + try: + raise RuntimeError(message) + except RuntimeError as exc: + return self.middleware.process_exception(request, exc) + + def process_internal_error(self, request, message='Test database error'): + """ + Call MaintenanceModeMiddleware.process_exception() from inside an active + exception handler with an InternalError (the maintenance-mode trigger). + """ + try: + raise InternalError(message) + except InternalError as exc: + return self.maintenance_mode_middleware.process_exception(request, exc) + + def assert_json_500_response(self, response, *, error=None, exception=None): + self.assertIsNotNone(response) + self.assertHttpStatus(response, status.HTTP_500_INTERNAL_SERVER_ERROR) + self.assertEqual(response.headers['Content-Type'], 'application/json') + + data = json.loads(response.content) + + self.assertIn('error', data) + self.assertIn('exception', data) + self.assertIn('netbox_version', data) + self.assertIn('python_version', data) + + if error is not None: + self.assertEqual(data['error'], error) + + if exception is not None: + self.assertEqual(data['exception'], exception) + + @override_settings(DEBUG=False) + def test_process_exception_handles_rest_api_request(self): + request = self.factory.get(reverse('api-root')) + + with self.capture_request_exception_signal() as captured_requests: + response = self.process_runtime_error(request, 'Simulated REST API error') + + self.assert_json_500_response(response, error='Simulated REST API error', exception='RuntimeError') + self.assertEqual(captured_requests, [request]) + + @override_settings(DEBUG=False) + def test_process_exception_handles_graphql_json_request(self): + request = self.factory.post( + reverse('graphql'), + data='{"query": "{ __typename }"}', + content_type='application/json', + ) + + with self.capture_request_exception_signal() as captured_requests: + response = self.process_runtime_error(request, 'Simulated GraphQL error') + + self.assert_json_500_response(response, error='Simulated GraphQL error', exception='RuntimeError') + self.assertEqual(captured_requests, [request]) + + @override_settings(DEBUG=False) + def test_process_exception_does_not_handle_graphql_request_without_json_content_type(self): + request = self.factory.get(reverse('graphql')) + + response = self.process_runtime_error(request, 'Simulated GraphiQL error') + + self.assertIsNone(response) + + @override_settings(DEBUG=False) + def test_process_exception_does_not_handle_non_api_request(self): + request = self.factory.get('/login/') + + response = self.process_runtime_error(request, 'Simulated UI error') + + self.assertIsNone(response) + + @override_settings(DEBUG=True) + def test_process_exception_does_not_handle_api_requests_in_debug_mode(self): + requests = ( + self.factory.get(reverse('api-root')), + self.factory.post( + reverse('graphql'), + data='{"query": "{ __typename }"}', + content_type='application/json', + ), + ) + + for request in requests: + with self.subTest(path=request.path_info): + response = self.process_runtime_error(request, 'Debug exception') + + self.assertIsNone(response) + + def test_maintenance_mode_handles_rest_api_request(self): + request = self.factory.get(reverse('api-root')) + + with patch('netbox.middleware.get_config', return_value=SimpleNamespace(MAINTENANCE_MODE=True)): + response = self.process_internal_error(request, 'Simulated maintenance mode REST API error') + + self.assert_json_500_response(response) + + def test_maintenance_mode_handles_graphql_json_request(self): + request = self.factory.post( + reverse('graphql'), + data='{"query": "{ __typename }"}', + content_type='application/json', + ) + + # With the fix, is_graphql_request short-circuits to the JSON handler before the + # messages/redirect path. Mock message storage so that if the fix regresses, the + # test fails on response shape instead of erroring on absent message middleware. + request._messages = Mock() + + with patch('netbox.middleware.get_config', return_value=SimpleNamespace(MAINTENANCE_MODE=True)): + response = self.process_internal_error(request, 'Simulated maintenance mode GraphQL error') + + self.assert_json_500_response(response)