"""Test for the subaccount model.""" from typing import Any from unittest.mock import MagicMock, patch import pytest from owsresponse import response from pythonfeatures import pythonfeatures from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm.session import Session from account.connectors import mysql from account.constants import constants, error from account.models import subaccount from account.models.sql.subaccount_document import ( SUBACCOUNT_DOCUMENT_SQL, SUBACCOUNT_DOCUMENT_WITH_TENANT_UUIDS_SQL, ) from tests.unit import db_operations @pytest.fixture def db_fixture(): """Set up the subaccount table.""" db_operations.create_tables() db_operations.seed_vendor_table() db_operations.seed_vend_contact_table() db_operations.seed_subaccount_table() db_operations.seed_parent_company_table() db_operations.seed_company_brand_table() def test_get_active_subaccounts_count(db_fixture): """Test getting non-zero number of subaccounts.""" response = subaccount.get_subaccount_count(vendor_id=1) assert response.status == 200 assert response.message == 5 def test_get_zero_active_subaccounts(db_fixture): """Test getting zero number of subaccounts.""" response = subaccount.get_subaccount_count(vendor_id=100) assert response.status == 200 assert response.message == 0 def test_get_active_subaccounts(db_fixture): """Test only active subaccounts are returned.""" response = subaccount.get_subaccounts(vendor_id=1, status=None, page_offset=0, page_limit=50) assert response.status == 200 assert len(response.message.get('items')) == 5 assert response.message.get('pagination').get('page_offset') == 0 assert response.message.get('pagination').get('page_limit') == 50 def test_get_subaccounts_with_contact_ids(db_fixture): """Test subaccounts with active contacts return contact ids.""" response = subaccount.get_subaccounts(vendor_id=1, status=None, page_offset=0, page_limit=50) assert response.status == 200 subaccounts_with_contact_ids = list( filter( lambda x: x.get('vend_contact_id') is not None, response.message.get('items'), ) ) assert len(subaccounts_with_contact_ids) == 2 assert response.message.get('items')[0].get('vend_contact_id') == 2 assert response.message.get('items')[1].get('vend_contact_id') == 3 or 4 def test_get_deactive_subaccounts(db_fixture): """Test only inactive subaccounts are returned.""" response = subaccount.get_subaccounts( vendor_id=1, status='deactivated', page_offset=0, page_limit=50 ) assert response.status == 200 assert len(response.message.get('items')) == 1 assert response.message.get('pagination').get('page_offset') == 0 assert response.message.get('pagination').get('page_limit') == 50 def test_get_subaccounts_page_limit_offset(db_fixture): """Test page limit and offsets are respected.""" offset0 = subaccount.get_subaccounts(vendor_id=1, status=None, page_offset=0, page_limit=2) assert offset0.status == 200 assert len(offset0.message.get('items')) == 2 assert offset0.message.get('pagination').get('page_offset') == 0 assert offset0.message.get('pagination').get('page_limit') == 2 offset1 = subaccount.get_subaccounts(vendor_id=1, status=None, page_offset=2, page_limit=2) assert offset1.status == 200 assert len(offset1.message.get('items')) == 2 assert offset1.message.get('pagination').get('page_offset') == 2 assert offset1.message.get('pagination').get('page_limit') == 2 offset2 = subaccount.get_subaccounts(vendor_id=1, status=None, page_offset=0, page_limit=0) assert offset2.status == 200 assert len(offset2.message.get('items')) == 5 assert offset2.message.get('pagination').get('page_offset') == 0 assert offset2.message.get('pagination').get('page_limit') == 0 assert offset0.message != offset1.message def test_get_subaccounts_not_found(db_fixture): """Test error is returned when no subaccounts are found.""" response = subaccount.get_subaccounts(vendor_id=100, status=None, page_offset=0, page_limit=50) assert response.status == 404 def test_get_subaccount(db_fixture): """Test specified subaccount is returned.""" expected_response = { 'subaccount_id': 1, 'vendor_id': 1, 'subaccount_name': 'Subaccount 1', 'description': 'Subaccount Description 1', 'country_id': 1, 'commission_override': 0.5, 'subaccount_split_type': 'Gross', 'subaccount_uuid': '11387154-0267-11ef-82c8-4a2888760684', } response = subaccount.get_subaccount(subaccount_id=1) assert response.status == 200 assert response.message == expected_response def test_get_subaccount_not_found(db_fixture): """Test error is returned when specified subaccount is not found.""" response = subaccount.get_subaccount(subaccount_id=100) assert response.status == 404 def test_get_subaccount_for_vendor(db_fixture): """Test a specified subaccount for vendor is returned.""" expected_response = { 'subaccount_id': 1, 'vendor_id': 1, 'subaccount_name': 'Subaccount 1', 'description': 'Subaccount Description 1', 'country_id': 1, 'commission_override': 0.5, 'subaccount_split_type': 'Gross', 'subaccount_uuid': '11387154-0267-11ef-82c8-4a2888760684', } response = subaccount.get_subaccount_for_vendor(subaccount_id=1, vendor_id=1) assert response.status == 200 assert response.message == expected_response def test_get_subaccount_for_vendor_not_found(db_fixture): """Test error is returned when subaccount is not found for vendor.""" response = subaccount.get_subaccount_for_vendor(subaccount_id=1, vendor_id=100) assert response.status == 404 def test_get_subaccount_document_by_id(monkeypatch, fixture_subaccount_document): """Test get subaccount document by id.""" result_message = MagicMock(fetchone=lambda: fixture_subaccount_document) monkeypatch.setattr(Session, 'execute', value=MagicMock(return_value=result_message)) response = subaccount.get_subaccount_document_by_id(2) assert response.status == 200 assert response.message == fixture_subaccount_document Session.execute.assert_called_once_with(SUBACCOUNT_DOCUMENT_SQL, {'subaccount_id': 2}) def test_get_subaccount_document_by_id_with_tenant(monkeypatch, fixture_subaccount_document): """Test get subaccount document by id.""" result_message = MagicMock(fetchone=lambda: fixture_subaccount_document) monkeypatch.setattr(Session, 'execute', value=MagicMock(return_value=result_message)) response = subaccount.get_subaccount_document_by_id(2, True) assert response.status == 200 assert response.message == fixture_subaccount_document Session.execute.assert_called_once_with( SUBACCOUNT_DOCUMENT_WITH_TENANT_UUIDS_SQL, {'subaccount_id': 2} ) def test_get_subaccount_document_by_id_for_no_results(monkeypatch): """Test get subaccount document by id for no results.""" result = None result_message = MagicMock(fetchone=lambda: result) monkeypatch.setattr(Session, 'execute', value=MagicMock(return_value=result_message)) response = subaccount.get_subaccount_document_by_id(2) assert response.status == 404 @patch('account.connectors.mysql._account_session', side_effect=SQLAlchemyError()) def test_get_subaccount_document_by_id_failure(session_scope, monkeypatch): """Test get subaccount document by id when failed.""" monkeypatch.setattr(session_scope, 'execute', value=MagicMock(return_value=False)) response = subaccount.get_subaccount_document_by_id(2) assert response.status == 500 def test_update_subaccount_success_activate( fixture_subaccount_id, fixture_subaccount_data_to_activate ): """Test to activate subaccount status successfully.""" response = subaccount.update_subaccount_status( fixture_subaccount_id, fixture_subaccount_data_to_activate['active'] ) assert response.status == 200 assert 'active' in response.message assert response.message['active'] is True def test_update_subaccount_success_deactivate( fixture_subaccount_id, fixture_subaccount_data_to_deactivate ): """Test to deactivate subaccount status successfully.""" response = subaccount.update_subaccount_status( fixture_subaccount_id, fixture_subaccount_data_to_deactivate['active'] ) assert response.status == 200 assert 'active' in response.message assert response.message['active'] is False def test_update_subaccount_status_failure(fixture_subaccount_data_to_activate): """Test when failed to activate/deactivate subaccount status.""" subaccount_id = 1234 response = subaccount.update_subaccount_status( subaccount_id, fixture_subaccount_data_to_activate ) assert response.status == 404 assert response.errors['message'] == error.ERROR_MESSAGE_SUBACCOUNT_NOT_FOUND @patch('account.connectors.mysql._account_session', side_effect=SQLAlchemyError()) @patch('account.models.subaccount.g') def test_update_subaccount_status_db_error( subaccount_id, fixture_subaccount_data_to_activate, app_context ): """Test get subaccount document by id when failed.""" response = subaccount.update_subaccount_status( subaccount_id, fixture_subaccount_data_to_activate ) assert response.status == 500 def test_get_subaccount_by_vendor_id_success(db_fixture): """Test a specified subaccount is exists or not.""" expected_response = { 'subaccount_id': 1, 'vendor_id': 1, 'subaccount_name': 'Subaccount 1', 'description': 'Subaccount Description 1', 'country_id': 1, 'commission_override': 0.5, 'subaccount_split_type': 'Gross', 'subaccount_uuid': '11387154-0267-11ef-82c8-4a2888760684', } response = subaccount.get_subaccount_by_vendor_id(subaccount_id=1, vendor_id=1) assert response.status == 200 assert response.message == expected_response def test_get_subaccount_by_vendor_id_not_found(db_fixture): """Test error is returned when subaccount is not found for vendor.""" response = subaccount.get_subaccount_by_vendor_id(subaccount_id=1, vendor_id=100) assert response.status == 404 def test_create_subaccount(db_fixture): """Test create subaccount.""" details = { 'vendor_id': 1, 'subaccount_name': 'Subaccount 1', } response = subaccount.create_subaccount(details) assert response.status == 200 @pytest.mark.parametrize('fetch_flags', [([]), (['IGNORED'])]) @pytest.mark.parametrize( 'uuids, expected, description', [ ([], [], 'empty list returns empty list'), ( [ '11387154-0267-11ef-82c8-4a2888760684', '77975ab6-e4bd-4564-afcb-09cc8618d359', '071aa349-f35b-47ee-96a3-35878bb6713e', ], [ { 'subaccount_id': 1, 'vendor_id': 1, 'uuid': '11387154-0267-11ef-82c8-4a2888760684', }, { 'subaccount_id': 4, 'vendor_id': 1, 'uuid': '77975ab6-e4bd-4564-afcb-09cc8618d359', }, ], 'only two items are returned; 071aa349-f35b-47ee-96a3-35878bb6713e is not in db', ), ], ) @patch('account.models.subaccount.g', spec=['request_context']) def test_lookup_subaccounts_by_uuids( mock_g: MagicMock, monkeypatch, db_fixture, fetch_flags: list[str], uuids: list[str], expected: list[dict[str, Any]], description: str, ) -> None: """Test lookup_subaccounts_by_uuids.""" monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response(message='enabled')), ) result = subaccount.lookup_subaccounts_by_uuids(uuids, fetch_flags) assert result.message == expected, description @pytest.mark.parametrize('fetch_flags', [([]), (['IGNORED'])]) @pytest.mark.parametrize( 'ids, expected, description', [ ([], [], 'empty list returns empty list'), ( [ '1', '4', 'not-in-there', ], [ { 'subaccount_id': 1, 'vendor_id': 1, 'uuid': '11387154-0267-11ef-82c8-4a2888760684', }, { 'subaccount_id': 4, 'vendor_id': 1, 'uuid': '77975ab6-e4bd-4564-afcb-09cc8618d359', }, ], 'only two items are returned; not-in-there is not in db', ), ], ) @patch('account.models.subaccount.g', spec=['request_context']) def test_lookup_subaccounts_by_subaccount_ids( mock_g: MagicMock, monkeypatch, db_fixture, fetch_flags: list[str], ids: list[str], expected: dict[str, Any], description: str, ) -> None: """Test lookup subaccounts by ids.""" monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response(message='enabled')), ) lookup_response = subaccount.lookup_subaccounts_by_subaccount_ids(ids, fetch_flags) assert lookup_response.status == 200 assert lookup_response.message == expected, description @pytest.mark.parametrize( 'fetch_flags', [ ([constants.FETCH_TENANT_HIERARCHY]), ([constants.FETCH_TENANT_HIERARCHY, 'IGNORED']), ], ) @pytest.mark.parametrize( 'uuids, expected, description', [ ([], [], 'empty list returns empty list'), ( [ '11387154-0267-11ef-82c8-4a2888760684', '77975ab6-e4bd-4564-afcb-09cc8618d359', '071aa349-f35b-47ee-96a3-35878bb6713e', ], [ { 'subaccount_id': 1, 'vendor_id': 1, 'uuid': '11387154-0267-11ef-82c8-4a2888760684', 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', }, { 'subaccount_id': 4, 'vendor_id': 1, 'uuid': '77975ab6-e4bd-4564-afcb-09cc8618d359', 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', }, ], 'only two items are returned; 071aa349-f35b-47ee-96a3-35878bb6713e is not in db', ), ], ) @patch('account.models.subaccount.g', spec=['request_context']) def test_lookup_subaccounts_and_hierarchy_by_uuids( mock_g: MagicMock, monkeypatch, db_fixture, fetch_flags: list[str], uuids: list[str], expected: list[dict[str, Any]], description: str, ) -> None: """Test lookup_subaccounts_by_uuids.""" monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response(message='enabled')), ) result = subaccount.lookup_subaccounts_by_uuids(uuids, fetch_flags) assert result.message == expected, description @pytest.mark.parametrize( 'fetch_flags', [ ([constants.FETCH_TENANT_HIERARCHY]), ([constants.FETCH_TENANT_HIERARCHY, 'IGNORED']), ], ) @pytest.mark.parametrize( 'ids, expected, description', [ ([], [], 'empty list returns empty list'), ( [ '1', '4', 'not-in-there', ], [ { 'subaccount_id': 1, 'vendor_id': 1, 'uuid': '11387154-0267-11ef-82c8-4a2888760684', 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', }, { 'subaccount_id': 4, 'vendor_id': 1, 'uuid': '77975ab6-e4bd-4564-afcb-09cc8618d359', 'vendor_uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'company_brand_uuid': 'd25a4cd1-e820-45f2-be5c-56edcfeb8298', 'parent_company_uuid': '955a1bbd-b623-4ea1-ab5f-8d6620c442fb', }, ], 'only two items are returned; not-in-there is not in db', ), ], ) @patch('account.models.subaccount.g', spec=['request_context']) def test_lookup_subaccounts_and_hierarchy_by_ids( mock_g: MagicMock, monkeypatch, db_fixture, fetch_flags: list[str], ids: list[str], expected: dict[str, Any], description: str, ) -> None: """Test lookup_subaccounts_by_ids.""" monkeypatch.setattr( pythonfeatures, 'get_single_feature', MagicMock(return_value=response.Response(message='enabled')), ) result = subaccount.lookup_subaccounts_by_subaccount_ids(ids, fetch_flags) assert result.message == expected, description def test_get_subaccounts_names_valid_uuids(db_fixture, app_context) -> None: """Test get_subaccount_names to return valid records.""" response = subaccount.get_subaccount_names( ['11387154-0267-11ef-82c8-4a2888760684', '3167fe36-0267-11ef-8440-4a2888760684'] ) assert len(response) == 2 assert response[0] == { 'subaccount_name': 'Subaccount 1', 'subaccount_id': 1, 'subaccount_uuid': '11387154-0267-11ef-82c8-4a2888760684', } assert response[1] == { 'subaccount_name': 'Subaccount 2', 'subaccount_id': 2, 'subaccount_uuid': '3167fe36-0267-11ef-8440-4a2888760684', } def test_get_subaccount_names_empty_list(db_fixture, app_context) -> None: """Test get_subaccount_names to return empty list""" response = subaccount.get_subaccount_names([]) assert len(response) == 0 def test_get_subaccount_names_invalid_uuids(db_fixture, app_context) -> None: """Test get_subaccount_names to return for invalid uuid""" response = subaccount.get_subaccount_names(['invalid_uuid']) assert len(response) == 0 def test_get_vendor_names_valid_and_not_valid_uuids(db_fixture, app_context) -> None: """Test get_subaccount_names to return a records for valid uuids.""" response = subaccount.get_subaccount_names( ['11387154-0267-11ef-82c8-4a2888760684', 'invalid_uuid'] ) assert len(response) == 1 assert response[0] == { 'subaccount_name': 'Subaccount 1', 'subaccount_id': 1, 'subaccount_uuid': '11387154-0267-11ef-82c8-4a2888760684', } @pytest.mark.parametrize( 'uuid, expected', [ pytest.param( '11387154-0267-11ef-82c8-4a2888760684', ( { 'subaccount_id': 1, 'uuid': '11387154-0267-11ef-82c8-4a2888760684', }, { 'vendor_id': 1, 'uuid': '87682992-bff2-40ff-aa75-18eb6214679e', 'migrated_to_abacus': False, }, {'id': 1, 'name': 'test_theorchard'}, ), id='Valid UUID', ), pytest.param('not-a-uuid', None, id='Invalid UUID'), ], ) def test_lookup_subaccount_and_vendor_and_company_brand_by_uuid( db_fixture, uuid: str, expected: dict[str, Any] ) -> None: """Test lookup_subaccount_and_vendor_and_company_brand_by_uuid.""" with mysql.session_scope(read_only=True) as session: result = subaccount.lookup_subaccount_and_vendor_and_company_brand_by_uuid(uuid, session) assert result == expected @pytest.mark.parametrize( 'subaccount_uuids, expected_length, expected_keys', [ pytest.param( [], 0, [], id='empty_list', ), pytest.param( ['12345678-1234-4567-8901-123456789012', '87654321-4321-4567-8901-210987654321'], 0, [], id='not_found_uuids', ), pytest.param( [ '11387154-0267-11ef-82c8-4a2888760684', '3167fe36-0267-11ef-8440-4a2888760684', ], 2, [ '11387154-0267-11ef-82c8-4a2888760684', '3167fe36-0267-11ef-8440-4a2888760684', ], id='valid_uuids', ), pytest.param( ['11387154-0267-11ef-82c8-4a2888760684', '12345678-1234-4567-8901-123456789012'], 1, ['11387154-0267-11ef-82c8-4a2888760684'], id='mixed_uuids', ), ], ) def test_lookup_subaccounts_and_vendors_and_company_brands_by_uuids( db_fixture, app_context: MagicMock, subaccount_uuids: list[str], expected_length: int, expected_keys: list[str], ) -> None: """Test lookup_subaccounts_and_vendors_and_company_brands_by_uuids.""" with mysql.session_scope(read_only=True) as session: result = subaccount.lookup_subaccounts_and_vendors_and_company_brands_by_uuids( subaccount_uuids, session ) assert len(result) == expected_length for key in expected_keys: assert key in result # Validate structure for valid entries if '11387154-0267-11ef-82c8-4a2888760684' in result: subaccount_1 = result['11387154-0267-11ef-82c8-4a2888760684'] assert subaccount_1[0]['subaccount_id'] == 1 assert subaccount_1[0]['uuid'] == '11387154-0267-11ef-82c8-4a2888760684' assert subaccount_1[1]['vendor_id'] == 1 assert subaccount_1[1]['uuid'] == '87682992-bff2-40ff-aa75-18eb6214679e' assert subaccount_1[1]['migrated_to_abacus'] is False assert subaccount_1[2]['id'] == 1 assert subaccount_1[2]['name'] == 'test_theorchard' if '3167fe36-0267-11ef-8440-4a2888760684' in result: subaccount_2 = result['3167fe36-0267-11ef-8440-4a2888760684'] assert subaccount_2[0]['subaccount_id'] == 2 assert subaccount_2[0]['uuid'] == '3167fe36-0267-11ef-8440-4a2888760684' assert subaccount_2[1]['vendor_id'] == 1 assert subaccount_2[1]['uuid'] == '87682992-bff2-40ff-aa75-18eb6214679e' assert subaccount_2[1]['migrated_to_abacus'] is False assert subaccount_2[2]['id'] == 1 assert subaccount_2[2]['name'] == 'test_theorchard' def test_delete_subaccount_by_uuid_active(db_fixture): """Test deleting an active subaccount sets date_deleted to the current time.""" import datetime uuid = '11387154-0267-11ef-82c8-4a2888760684' frozen_now = datetime.datetime(2024, 6, 15, 12, 0, 0) with patch('account.models.subaccount.datetime') as mock_dt: mock_dt.datetime.now.return_value = frozen_now result = subaccount.delete_subaccount_by_uuid(uuid) assert result == {'subaccount_uuid': uuid, 'date_deleted': frozen_now} session = mysql._account_session() row = session.query(subaccount.Subaccount).filter_by(subaccount_uuid=uuid).first() assert row.date_deleted == frozen_now session.close() def test_delete_subaccount_by_uuid_already_deleted(db_fixture): """Test deleting an already-deleted subaccount is idempotent, date_deleted unchanged.""" uuid = 'dc7afecc-4ae8-4b29-8949-bbdc58d6634a' session = mysql._account_session() original_date = ( session.query(subaccount.Subaccount).filter_by(subaccount_uuid=uuid).first().date_deleted ) session.close() result = subaccount.delete_subaccount_by_uuid(uuid) assert result == {'subaccount_uuid': uuid, 'date_deleted': None} session = mysql._account_session() row = session.query(subaccount.Subaccount).filter_by(subaccount_uuid=uuid).first() assert row.date_deleted == original_date session.close()