"""Test a MySQL connector of the reverse payout flow.""" from copy import deepcopy from unittest import mock import pytest from sqlalchemy import exc from accounting.flows.reserve_payouts.connectors import mysql @pytest.mark.parametrize('db_name, conn_string', [ ('ART_RELATIONS', 'mysql::host/art_rel'), ('ACCOUNTING_FLAT', 'mysql::host/acc_flat'), ]) @mock.patch('accounting.flows.reserve_payouts.connectors.mysql.setting') @mock.patch('accounting.flows.reserve_payouts.connectors.mysql.create_engine') def test_create_engines(mocked_create_engine, settings, db_name, conn_string): """Test _create_engines function.""" settings.DATABASES = [db_name] settings.DB_CONNECTION_STRINGS = {db_name: conn_string} expected_conn_args = {'test': 'test'} settings.DB_CONNECT_ARGS = deepcopy(expected_conn_args) db_engine = mock.MagicMock() mocked_create_engine.return_value = db_engine assert mysql._create_engines() mocked_create_engine.assert_called_with( conn_string, connect_args=expected_conn_args) @pytest.mark.parametrize('db_name, db_engines, expected_return', [ (None, {}, None), ('test_db_name', {}, None), ('test_db_name', {'another': 'session_scope'}, None), ('test_db_name2', {'test_db_name2': 'expctd_fnct'}, 'expctd_fnct') ]) @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql._create_engines') def test_get_db_engine( mock_create_engines, db_name, db_engines, expected_return): """Test get_db_engine helper function.""" mysql._db_engines = db_engines assert mysql.get_db_engine(db_name) == expected_return @pytest.mark.parametrize('exception_class, expected_result', [ (exc.SQLAlchemyError, False), (exc.ProgrammingError, False), (None, True) ]) @pytest.mark.parametrize('db_name', ['art_relations', 'accounting_flat']) @mock.patch('accounting.flows.reserve_payouts.connectors.mysql.setting') @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql.get_db_engine') def test_health_check( mock_get_db_engine, mock_setting, db_name, exception_class, expected_result): """Test health_check mysql connectors function.""" health_check_query = 'SELECT 1;' mock_setting.DB_HEALTH_CHECK_QUERY = health_check_query mock_engine = mock.MagicMock() mock_query_result = mock.MagicMock() if not exception_class: mock_engine.execute.return_value = mock_query_result mock_get_db_engine.return_value = mock_engine if exception_class: mock_engine.execute.side_effect = exception_class(1, 1, 1) result = mysql.health_check(db_name) assert result.bool == expected_result mock_get_db_engine.assert_called_with(db_name) mock_engine.execute.assert_called_with(health_check_query) if not exception_class: mock_query_result.close.assert_has_calls([mock.call()]) @pytest.mark.parametrize('query_params', [ {'a': 'a'}, {'test_param': 'test_value', 'test_param_2': 'test_value_2'} ]) @pytest.mark.parametrize('sql', ['TEST QUERY 1;', 'TEST QUERY 2;']) @pytest.mark.parametrize('db_name', ['test_db_1', 'test_db_2']) @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql.get_db_engine') def test_execute_sql(get_db_engine, db_name, sql, query_params): """Test _execute_sql utility function.""" mock_engine = mock.MagicMock() mock_result = mock.MagicMock() mock_engine.execute.return_value = mock_result get_db_engine.return_value = mock_engine assert mysql._execute_sql(db_name, sql, query_params) == mock_result get_db_engine.assert_called_with(db_name) mock_engine.execute.assert_called_with(sql, query_params) @pytest.mark.parametrize('period_id, acc_flat_name, art_rel_name', [ (222, 'test_flat', 'test_art'), (223, 'flat_test', 'art_test'), ]) @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql.alchemy_sql') @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql._execute_sql') @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql.sql') @mock.patch('accounting.flows.reserve_payouts.connectors.mysql.setting') def test_get_physical_transactions_sum( setting, sql, execute_sql, alchemy_sql, period_id, acc_flat_name, art_rel_name): """Test get_physical_transactions_sum function.""" mock_phys_trans_sql = mock.MagicMock() mock_phys_trans_sql.format.return_value = mock_phys_trans_sql sql.PHYSICAL_TRANSACTIONS_SUM_SQL = mock_phys_trans_sql setting.DB_ACC_FLAT_NAME = acc_flat_name setting.DB_ART_RELATIONS_NAME = art_rel_name mock_result = mock.MagicMock() expected_query_params = { 'period_id': period_id } alchemy_sql.text.return_value = mock_phys_trans_sql execute_sql.return_value = mock_result result = mysql.get_physical_transactions_sum(period_id) assert result == mock_result execute_sql.assert_called_with( acc_flat_name, mock_phys_trans_sql, expected_query_params) alchemy_sql.text.assert_called_with(mock_phys_trans_sql) mock_phys_trans_sql.format.assert_called_with( accountingflat=acc_flat_name, art_relations=art_rel_name) @pytest.mark.parametrize('label_ids', [ (123, 321, 1234), (111, 222, 333, 444, 555), ]) @pytest.mark.parametrize('period_id, art_rel_name', [ (222, 'test_art_rel'), (223, 'art_rel_test'), ]) @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql.alchemy_sql') @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql.sql') @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql._execute_sql') @mock.patch('accounting.flows.reserve_payouts.connectors.mysql.setting') def test_get_vendor_contracts( setting, execute_sql, sql, alchemy_sql, period_id, art_rel_name, label_ids): """Test get_vendor_contracts function.""" vendor_contract_sql = mock.MagicMock() sql.VENDOR_CONTRACT_SQL = vendor_contract_sql setting.DB_ART_RELATIONS_NAME = art_rel_name mock_result = mock.MagicMock() execute_sql.return_value = mock_result alchemy_sql.text.return_value = vendor_contract_sql expected_query_params = { 'label_ids': label_ids, 'period_id': period_id} result = mysql.get_vendor_contracts(period_id, label_ids) assert result == mock_result execute_sql.assert_called_with( art_rel_name, vendor_contract_sql, expected_query_params) alchemy_sql.text.assert_called_with(vendor_contract_sql) @pytest.mark.parametrize('keys, data_tuple, expected_result', [ ((1, 2, 3), ('a', 'b', 'c'), {1: 'a', 2: 'b', 3: 'c'}), (('a', 'b', 'c'), (1, 2, 3), {'a': 1, 'b': 2, 'c': 3}) ]) def test_make_dict_from_sql_result(keys, data_tuple, expected_result): """Test _make_dict_from_sql_result utility function.""" actual_result = mysql._make_dict_from_sql_result(keys, data_tuple) assert actual_result == expected_result def test_vendor_query_result_to_dict(): """Test query_result_to_dict utility function.""" keys = ('vendor_id', 'some_attr') items = (1, 'some_value') expected = { 1: {'vendor_id': 1, 'some_attr': 'some_value'} } query_result = mock.MagicMock() query_result.keys.return_value = keys query_result.__iter__.return_value = [items] actual = mysql.vendor_query_result_to_dict(query_result) assert actual == expected query_result.close.assert_has_calls([mock.call()]) def test_vendor_query_result_to_dict_exception(): """Test query_result_to_dict with exception.""" keys = ('vendor_id', 'some_attr') items = [(1, 'some_value'), (1, 'some_value')] query_result = mock.MagicMock() query_result.keys.return_value = keys query_result.__iter__.return_value = items with pytest.raises(ValueError): mysql.vendor_query_result_to_dict(query_result) query_result.close.assert_has_calls([mock.call()]) @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql.alchemy_sql') @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql.sql') @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql._execute_sql') @mock.patch('accounting.flows.reserve_payouts.connectors.mysql.setting') def test_truncate_reserves_temp_table(setting, execute_sql, sql, alchemy_sql): """Test _truncate_reserves_temp_table utility function.""" db_name = 'test_acounting_flat' setting.DB_ACC_FLAT_NAME = db_name truncate_query = 'truncate' sql.TRUNCATE_TEMP_TABLE = truncate_query alchemy_sql.text.side_effect = lambda x: x mock_result = mock.MagicMock() execute_sql.return_value = mock_result mysql.truncate_reserves_temp_table() alchemy_sql.text.assert_called_once_with(truncate_query) execute_sql.assert_has_calls([mock.call(db_name, truncate_query, {})]) mock_result.close.assert_has_calls([mock.call()]) @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql.alchemy_sql') @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql.sql') @mock.patch( 'accounting.flows.reserve_payouts.connectors.mysql._execute_sql') @mock.patch('accounting.flows.reserve_payouts.connectors.mysql.setting') def test_populate_reserves_temp_table(setting, execute_sql, sql, alchemy_sql): """Test _populate_reserves_temp_table utility function.""" period_id = 4242 db_name = 'test_acounting_flat' setting.DB_ACC_FLAT_NAME = db_name insert_query = 'insert' sql.INSERT_PHYS_TRANSACTIONS_FOR_PERIOD = insert_query alchemy_sql.text.side_effect = lambda x: x mock_result = mock.MagicMock() execute_sql.return_value = mock_result mysql.populate_reserves_temp_table(period_id) alchemy_sql.text.assert_called_once_with(insert_query) execute_sql.assert_has_calls( [mock.call(db_name, insert_query, {'period_id': period_id})]) mock_result.close.assert_has_calls([mock.call()])