"""Test for utils/handler_util.py::check_admin_access.""" from unittest import TestCase from unittest.mock import MagicMock, patch from flask import Flask, g from requests import Response from account.constants import error, tenants from account.utils import handler_util def dummy_function(*args, **kwargs): return 'Function Executed' class TestCheckAdminAccess(TestCase): """Tests for check_admin_access decorator.""" @classmethod def setUpClass(cls) -> None: cls.app = Flask(__name__) def setUp(self) -> None: # Decorate the dummy function with the check_admin_access decorator self.decorated_function = handler_util.check_admin_access( tenant_type=tenants.TenantType.ACCOUNT, tenant_uuid_key='vendor_uuid' )(dummy_function) self.app_context = self.app.app_context() self.app_context.push() self.test_uuid = 'bca2352e-bbbc-4414-839b-24ed9a0a1bdd' def tearDown(self) -> None: self.app_context.pop() def test_missing_vendor_uuid(self) -> None: """Test scenario where 'vendor_uuid' is missing.""" resp = self.decorated_function(vendor_uuid=None) self.assertEqual(resp.errors['code'], error.ERROR_CODE_INVALID_INPUT) self.assertEqual(resp.errors['message'], error.ERROR_MESSAGE_MISSING_VENDOR_ID) @patch('owsrequest.request.process') @patch('owsrequest.flask_request.next_correlation_id') def test_check_with_access( self, mock_next_correlation_id: MagicMock, mock_request_process: MagicMock ) -> None: """Test scenario where request is successful.""" g.request_context = MagicMock(authorization='Bearer token') mock_next_correlation_id.return_value = 'correlation_id' mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = { 'tenants': [{'tenant_uuid': self.test_uuid, 'access': True}] } mock_request_process.return_value = mock_response # decorator resp = self.decorated_function(vendor_uuid=self.test_uuid) self.assertEqual(resp, 'Function Executed') @patch('owsrequest.request.process') @patch('owsrequest.flask_request.next_correlation_id') def test_check_without_access( self, mock_next_correlation_id: MagicMock, mock_request_process: MagicMock ) -> None: """Test scenario where request is successful.""" g.request_context = MagicMock(authorization='Bearer token') mock_next_correlation_id.return_value = 'correlation_id' mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = { 'tenants': [{'tenant_uuid': self.test_uuid, 'access': False}] } mock_request_process.return_value = mock_response resp = self.decorated_function(vendor_uuid=self.test_uuid) # decorator intercepts the request before it reaches the handler. self.assertEqual(resp.status_code, 403) self.assertEqual( resp.json, {'code': 'authorization_error', 'message': 'Unauthorized to access tenants.'} ) @patch('owsrequest.request.process') @patch('owsrequest.flask_request.next_correlation_id') def test_unauthorized_request( self, mock_next_correlation_id: MagicMock, mock_request_process: MagicMock ) -> None: """Test scenario where request returns unauthorized status.""" g.request_context = MagicMock(authorization='Bearer token') mock_next_correlation_id.return_value = 'correlation_id' mock_response = MagicMock() mock_response.status_code = 401 mock_request_process.return_value = mock_response resp = self.decorated_function(vendor_uuid=self.test_uuid) self.assertEqual(resp.json['code'], error.ERROR_CODE_AUTHORIZATION) self.assertEqual(resp.json['message'], error.ERROR_MESSAGE_UNAUTHORIZED_TENANT_ACCESS) self.assertEqual(resp.status_code, 401) @patch('owsrequest.request.process') @patch('owsrequest.flask_request.next_correlation_id') def test_other_error_status( self, mock_next_correlation_id: MagicMock, mock_request_process: MagicMock ) -> None: """Test scenario where request returns other error status.""" g.request_context = MagicMock(authorization='Bearer token') mock_next_correlation_id.return_value = 'correlation_id' mock_response = Response() mock_response.status_code = 500 mock_response._content = b'Internal Server Error' mock_request_process.return_value = mock_response resp = self.decorated_function(vendor_uuid=self.test_uuid) self.assertEqual(resp.json['code'], error.ERROR_CODE_OWS_PERMISSIONS_REQUEST) self.assertEqual(resp.json['message'], 'Internal Server Error') self.assertEqual(resp.status_code, 500)