"""Integration tests for POST /vendors/company_brands/dataloader endpoint.""" import pytest import requests from tests.integration.conftest import ProfileInfo from tests.integration.consts import api vendor_data_single = [ { 'vendor_id': 66975, 'uuid': '79a391d5-85d2-416a-9551-b26faaf09311', 'company_brand_id': 1, 'company_brand_uuid': '77c6c150-19f3-451f-a692-1a5d7ef30e85', } ] vendor_data_multiple = [ { 'vendor_id': 66975, 'uuid': '79a391d5-85d2-416a-9551-b26faaf09311', 'company_brand_id': 1, 'company_brand_uuid': '77c6c150-19f3-451f-a692-1a5d7ef30e85', }, { 'vendor_id': 1, 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand_id': 2, 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', }, { 'vendor_id': 34659, 'uuid': 'bd6d540a-db73-46a6-b3dd-fddc6d6ac38a', 'company_brand_id': 3, 'company_brand_uuid': 'ae886607-ae74-42f4-b7a7-d33f6a6288f4', }, { 'vendor_id': 9162, 'uuid': 'ef1c0372-aa6f-466d-845a-1e2b864dd48f', 'company_brand_id': 4, 'company_brand_uuid': '9b36a627-4be9-4762-b790-5628676f5f90', }, { 'vendor_id': 22699, 'uuid': 'bccfb1a5-eca4-4a92-ba1b-7e91d426c91d', 'company_brand_id': 5, 'company_brand_uuid': '31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588', }, { 'vendor_id': 17330, 'uuid': 'e720eab3-4d94-4691-8065-4300d593be3a', 'company_brand_id': 6, 'company_brand_uuid': '581557e3-94a4-40a1-8f85-1b4328170144', }, { 'vendor_id': 74330, 'uuid': '5fe7fd8b-816e-4d3b-b337-8fb021b7d044', 'company_brand_id': 7, 'company_brand_uuid': '7c8b382c-fc37-4179-9115-2165b1a93bed', }, { 'vendor_id': 781769, 'uuid': 'e2b38818-5b1a-11ef-abbe-120c1dfc8269', 'company_brand_id': 8, 'company_brand_uuid': '9c270cea-cde6-4240-81cb-945ec2bac811', }, { 'vendor_id': 8882, 'uuid': 'a466f36d-d135-44b8-8562-59acca667746', 'company_brand_id': 9, 'company_brand_uuid': '567e4261-ad8a-4e89-8875-d44c26b96f2c', }, { 'vendor_id': 76355, 'uuid': '8cf72161-165b-40e4-be89-b5d415b7c467', 'company_brand_id': 10, 'company_brand_uuid': '2ed1077d-bcd3-4db6-9a47-c37da82aab18', }, None, # for non-existent vendor ] @pytest.mark.parametrize( 'vendor_uuids,expected_data', [ ( ['79a391d5-85d2-416a-9551-b26faaf09311'], vendor_data_single, ), ( [ '79a391d5-85d2-416a-9551-b26faaf09311', '87682992-bff2-40ff-aa75-18eb6214679e', 'bd6d540a-db73-46a6-b3dd-fddc6d6ac38a', 'ef1c0372-aa6f-466d-845a-1e2b864dd48f', 'bccfb1a5-eca4-4a92-ba1b-7e91d426c91d', 'e720eab3-4d94-4691-8065-4300d593be3a', '5fe7fd8b-816e-4d3b-b337-8fb021b7d044', 'e2b38818-5b1a-11ef-abbe-120c1dfc8269', 'a466f36d-d135-44b8-8562-59acca667746', '8cf72161-165b-40e4-be89-b5d415b7c467', 'f29dd336-d39e-4163-b0b8-cf2d0bbf3910', # non-existent ], vendor_data_multiple, ), ], ) def test_vendors_company_brands_dataloader_with_skip_access_check( vendor_uuids: list[str], expected_data: list ) -> None: """Test dataloader endpoint with skip_access_check=True.""" response = requests.post( api.VENDORS_COMPANY_BRANDS_DATALOADER, json={'vendor_uuids': vendor_uuids, 'skip_access_check': True}, ) assert response.status_code == 200 response_json = response.json() assert 'vendors' in response_json assert response_json['vendors'] == expected_data def test_vendors_company_brands_dataloader_with_access_granted( ows_account_test_user_settings_profile: ProfileInfo, ) -> None: """Test dataloader endpoint with access check passing.""" response = requests.post( api.VENDORS_COMPANY_BRANDS_DATALOADER, json={'vendor_uuids': ['79a391d5-85d2-416a-9551-b26faaf09311'], 'skip_access_check': False}, headers={ 'Orchard-Profile-Type': ows_account_test_user_settings_profile.profile_type, 'Orchard-Profile-Id': str(ows_account_test_user_settings_profile.profile_id), 'Orchard-Identity-Id': ows_account_test_user_settings_profile.identity_id, }, ) assert response.status_code == 200 response_json = response.json() assert 'vendors' in response_json assert response_json['vendors'] == vendor_data_single @pytest.mark.skip(reason='Access check no longer returns 403, just logs and allows access') def test_vendors_company_brands_dataloader_with_access_denied( bearer_token_without_vendor_star: str, ows_permissions_test_user_label_profile: ProfileInfo, ) -> None: """Test dataloader endpoint when user has no access to requested vendors.""" response = requests.post( api.VENDORS_COMPANY_BRANDS_DATALOADER, json={'vendor_uuids': ['79a391d5-85d2-416a-9551-b26faaf09311'], 'skip_access_check': False}, headers={ 'Authorization': f'Bearer {bearer_token_without_vendor_star}', 'Orchard-Profile-Type': ows_permissions_test_user_label_profile.profile_type, 'Orchard-Profile-Id': str(ows_permissions_test_user_label_profile.profile_id), 'Orchard-Identity-Id': ows_permissions_test_user_label_profile.identity_id, }, ) assert response.status_code == 403 response_json = response.json() assert response_json['code'] == 'authorization_error' assert 'Access denied' in response_json['message'] def test_vendors_company_brands_dataloader_moneyhub_profile_access( moneyhub_test_user_profile: ProfileInfo, ) -> None: """Test dataloader endpoint with MoneyhubProfile that only has access to the subaccount.""" from tests.integration import config # Look up the vendor UUID for vendor_id 26760 lookup_response = requests.post( f'{config.QA_BASE_URL}/lookup/vendors/vendor-ids/', json={'vendor_ids': [26760]}, ) assert lookup_response.status_code == 200 vendor_uuid = lookup_response.json()['vendors'][0]['uuid'] # Call company_brands dataloader with MoneyhubProfile headers response = requests.post( api.VENDORS_COMPANY_BRANDS_DATALOADER, json={'vendor_uuids': [vendor_uuid], 'skip_access_check': False}, headers={ 'Orchard-Identity-Id': moneyhub_test_user_profile.identity_id, 'Orchard-Identity-Uuid': moneyhub_test_user_profile.identity_id, 'Orchard-Profile-Id': str(moneyhub_test_user_profile.profile_id), 'Orchard-Profile-Type': moneyhub_test_user_profile.profile_type, 'Orchard-Roles': 'accounting', }, ) assert response.status_code == 200 response_json = response.json() assert 'vendors' in response_json assert len(response_json['vendors']) == 1 vendor_entry = response_json['vendors'][0] assert vendor_entry['vendor_id'] == 26760 assert vendor_entry['uuid'] == vendor_uuid assert 'company_brand_id' in vendor_entry assert 'company_brand_uuid' in vendor_entry @pytest.mark.parametrize( 'vendor_uuids,expected_status', [ ([], 200), (['invalid-uuid'], 400), ], ) def test_vendors_company_brands_dataloader_edge_cases( vendor_uuids: list[str], expected_status: int ) -> None: """Test dataloader endpoint with edge cases.""" response = requests.post( api.VENDORS_COMPANY_BRANDS_DATALOADER, json={'vendor_uuids': vendor_uuids, 'skip_access_check': True}, ) assert response.status_code == expected_status