"""Test for utils/handler_util.py::check_bulk_admin_access.""" from unittest import TestCase from unittest.mock import MagicMock, patch from flask import Flask, g from account.constants import error, tenants from account.utils import handler_util def dummy_function(*args, **kwargs): return 'Function Executed' class TestCheckBulkAdminAccessAccount(TestCase): """Tests for check_bulk_admin_access decorator.""" @classmethod def setUpClass(cls) -> None: cls.app = Flask(__name__) def setUp(self) -> None: # Decorate the dummy function with the check_bulk_admin_access decorator self.decorated_function = handler_util.check_bulk_admin_access( tenant_type=tenants.TenantType.ACCOUNT )(dummy_function) # Set up Flask application context self.app_context = self.app.app_context() self.app_context.push() # Set up Flask request context (required for flask.request.get_json()) self.request_context = self.app.test_request_context() self.request_context.push() self.test_uuids = [ 'bca2352e-bbbc-4414-839b-24ed9a0a1bdd', 'dca2352e-bbbc-4414-839b-24ed9a0a1bdd', ] def tearDown(self) -> None: self.request_context.pop() self.app_context.pop() # Note: Input validation tests (test_invalid_json, test_non_list_payload, test_empty_list) # removed since validation is now handled by separate marshmallow schema decorator @patch('owsrequest.request.process') @patch('owsrequest.flask_request.next_correlation_id') @patch('flask.request.get_json') def test_bulk_check_with_access( self, mock_get_json: MagicMock, mock_next_correlation_id: MagicMock, mock_request_process: MagicMock, ) -> None: """Test scenario where all tenants have access.""" g.request_context = MagicMock(authorization='Bearer token') mock_next_correlation_id.return_value = 'correlation_id' mock_get_json.return_value = {'vendor_uuids': self.test_uuids} mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = { 'tenants': [ {'tenant_uuid': self.test_uuids[0], 'access': True}, {'tenant_uuid': self.test_uuids[1], 'access': True}, ] } mock_request_process.return_value = mock_response resp = self.decorated_function() self.assertEqual(resp, 'Function Executed') @patch('owsrequest.request.process') @patch('owsrequest.flask_request.next_correlation_id') @patch('flask.request.get_json') def test_bulk_check_without_access( self, mock_get_json: MagicMock, mock_next_correlation_id: MagicMock, mock_request_process: MagicMock, ) -> None: """Test scenario where one tenant doesn't have access.""" g.request_context = MagicMock(authorization='Bearer token') mock_next_correlation_id.return_value = 'correlation_id' mock_get_json.return_value = {'vendor_uuids': self.test_uuids} mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = { 'tenants': [ {'tenant_uuid': self.test_uuids[0], 'access': True}, {'tenant_uuid': self.test_uuids[1], 'access': False}, ] } mock_request_process.return_value = mock_response resp = self.decorated_function() self.assertEqual(resp.status_code, 403) self.assertEqual( resp.json, { 'code': 'authorization_error', 'message': 'Unauthorized to access one or more tenants.', }, ) @patch('owsrequest.request.process') @patch('owsrequest.flask_request.next_correlation_id') @patch('flask.request.get_json') def test_bulk_check_unauthorized_request( self, mock_get_json: MagicMock, mock_next_correlation_id: MagicMock, mock_request_process: MagicMock, ) -> None: """Test scenario where bulk request returns unauthorized status.""" g.request_context = MagicMock(authorization='Bearer token') mock_next_correlation_id.return_value = 'correlation_id' mock_get_json.return_value = {'vendor_uuids': self.test_uuids} mock_response = MagicMock() mock_response.status_code = 403 mock_request_process.return_value = mock_response resp = self.decorated_function() 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, 403) class TestCheckBulkAdminAccessSubaccount(TestCase): """Tests for check_bulk_admin_access decorator with TenantType.SUBACCOUNT.""" @classmethod def setUpClass(cls) -> None: cls.app = Flask(__name__) def setUp(self) -> None: # Decorate the dummy function with the check_bulk_admin_access decorator self.decorated_function = handler_util.check_bulk_admin_access( tenant_type=tenants.TenantType.SUBACCOUNT )(dummy_function) # Set up Flask application context self.app_context = self.app.app_context() self.app_context.push() # Set up Flask request context (required for flask.request.get_json()) self.request_context = self.app.test_request_context() self.request_context.push() self.test_uuids = [ 'bca2352e-bbbc-4414-839b-24ed9a0a1bdd', 'dca2352e-bbbc-4414-839b-24ed9a0a1bdd', ] def tearDown(self) -> None: self.request_context.pop() self.app_context.pop() @patch('owsrequest.request.process') @patch('owsrequest.flask_request.next_correlation_id') @patch('flask.request.get_json') def test_bulk_check_with_access( self, mock_get_json: MagicMock, mock_next_correlation_id: MagicMock, mock_request_process: MagicMock, ) -> None: """Test scenario where all tenants have access.""" g.request_context = MagicMock(authorization='Bearer token') mock_next_correlation_id.return_value = 'correlation_id' mock_get_json.return_value = {'vendor_uuids': self.test_uuids} mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = { 'tenants': [ {'tenant_uuid': self.test_uuids[0], 'access': True}, {'tenant_uuid': self.test_uuids[1], 'access': True}, ] } mock_request_process.return_value = mock_response resp = self.decorated_function() self.assertEqual(resp, 'Function Executed') @patch('owsrequest.request.process') @patch('owsrequest.flask_request.next_correlation_id') @patch('flask.request.get_json') def test_bulk_check_without_access( self, mock_get_json: MagicMock, mock_next_correlation_id: MagicMock, mock_request_process: MagicMock, ) -> None: """Test scenario where one tenant doesn't have access.""" g.request_context = MagicMock(authorization='Bearer token') mock_next_correlation_id.return_value = 'correlation_id' mock_get_json.return_value = {'vendor_uuids': self.test_uuids} mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = { 'tenants': [ {'tenant_uuid': self.test_uuids[0], 'access': True}, {'tenant_uuid': self.test_uuids[1], 'access': False}, ] } mock_request_process.return_value = mock_response resp = self.decorated_function() self.assertEqual(resp.status_code, 403) self.assertEqual( resp.json, { 'code': 'authorization_error', 'message': 'Unauthorized to access one or more tenants.', }, ) @patch('owsrequest.request.process') @patch('owsrequest.flask_request.next_correlation_id') @patch('flask.request.get_json') def test_bulk_check_unauthorized_request( self, mock_get_json: MagicMock, mock_next_correlation_id: MagicMock, mock_request_process: MagicMock, ) -> None: """Test scenario where bulk request returns unauthorized status.""" g.request_context = MagicMock(authorization='Bearer token') mock_next_correlation_id.return_value = 'correlation_id' mock_get_json.return_value = {'vendor_uuids': self.test_uuids} mock_response = MagicMock() mock_response.status_code = 403 mock_request_process.return_value = mock_response resp = self.decorated_function() 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, 403)