"""contract_builder unit tests.""" import json from unittest.mock import MagicMock, patch import pandas from snapshot_contract import contract_builder as contract_builder from snapshot_contract.constants import IMPOSSIBLE_DATE_FUTURE from snapshot_contract.constants import PRODUCT from snapshot_contract.constants import TRACK @patch('snapshot_contract.contract_builder._flatten_contract_term_conditions') @patch('snapshot_contract.contract_builder.IsrcHelper') @patch('snapshot_contract.contract_builder._combine_contract_conditions') @patch('snapshot_contract.contract_builder._base_contract_detail') @patch('snapshot_contract.contract_builder._apply_term_type') def test_flatten_contract_data( apply_term_type, base_contract_detail, combine_contract_conditions, isrc_helper, flatten_contract_term_conditions, mock_contract, mock_contract_term, mock_isrcs): """Test main module method: flatten_contract_data.""" combined = { 'contract_id': mock_contract['contract_id'], 'payee_currency_id': mock_contract['currency_code'], 'payee_id': mock_contract['payee_id'], 'term_end': IMPOSSIBLE_DATE_FUTURE, 'term_start': mock_contract['term_start'], 'priority': mock_contract_term['priority'], 'contract_vendor_id': mock_contract_term['attachments'][0], 'base_store_id': mock_contract_term['conditions']['stores'][0], 'base_territory_id': mock_contract_term['conditions']['countries'][0], 'base_transaction_type_id': mock_contract_term['conditions']['transaction_types'][0] } accounting_run_id = 12 contract_df = pandas.DataFrame([mock_contract]) contract_term_df = pandas.DataFrame([mock_contract_term]) isrc_df = pandas.DataFrame(mock_isrcs) isrc_helper.return_value = MagicMock() base_contract_detail.return_value = mock_contract flatten_contract_term_conditions.return_value = [{'flat': True}] apply_term_type.return_value = [{'term_specific': True}] combine_contract_conditions.return_value = [combined] res = contract_builder.flatten_contract_data( accounting_run_id, contract_df, contract_term_df, isrc_df ) assert len(res) > 0 base_contract_detail.assert_called_once() flatten_contract_term_conditions.assert_called_once() apply_term_type.assert_called_once() combine_contract_conditions.assert_called_once() @patch('snapshot_contract.contract_builder.IsrcHelper') def test_apply_term_type_label(isrc_helper, mock_contract_term): """Test creating base / label specific dicts from "generic" term_conditions_list.""" mock_conditions = mock_contract_term['conditions'] term_conditions_list = [ { 'attachment': mock_contract_term['attachments'][0], 'country': mock_conditions['countries'][0], 'store': mock_conditions['stores'][0], 'txn_type': mock_conditions['transaction_types'][0] }, { 'attachment': mock_contract_term['attachments'][1], 'country': mock_conditions['countries'][1], 'store': mock_conditions['stores'][1], 'txn_type': mock_conditions['transaction_types'][0] }, { 'attachment': mock_contract_term['attachments'][1], 'country': mock_conditions['countries'][1], 'store': mock_conditions['stores'][0], 'txn_type': mock_conditions['transaction_types'][0] } ] contract_terms_df = pandas.DataFrame([mock_contract_term]) index = 0 res = contract_builder._apply_term_type( contract_terms_df, index, isrc_helper, term_conditions_list ) assert len(res) == len(term_conditions_list) for condition in res: assert condition['priority'] assert condition['contract_vendor_id'] assert condition['base_store_id'] assert condition['base_territory_id'] assert condition['base_transaction_type_id'] assert condition['base_rate'] @patch('snapshot_contract.contract_builder.IsrcHelper') def test_apply_term_type_product(isrc_helper, mock_contract_term): """Test creating product specific dicts from "generic" term_conditions_list.""" contract_term = mock_contract_term.copy() contract_term.update({ 'is_base_term': 0, 'priority': 2, 'term_type': PRODUCT }) mock_conditions = mock_contract_term['conditions'] term_conditions_list = [ { 'attachment': mock_contract_term['attachments'][0], 'country': mock_conditions['countries'][0], 'store': mock_conditions['stores'][0], 'txn_type': mock_conditions['transaction_types'][0] }, { 'attachment': mock_contract_term['attachments'][1], 'country': mock_conditions['countries'][1], 'store': mock_conditions['stores'][1], 'txn_type': mock_conditions['transaction_types'][0] }, { 'attachment': mock_contract_term['attachments'][1], 'country': mock_conditions['countries'][1], 'store': mock_conditions['stores'][0], 'txn_type': mock_conditions['transaction_types'][0] } ] contract_terms_df = pandas.DataFrame([contract_term]) index = 0 res = contract_builder._apply_term_type( contract_terms_df, index, isrc_helper, term_conditions_list ) assert len(res) == len(term_conditions_list) for condition in res: assert condition['priority'] assert condition['product_upc'] assert condition['product_store_id'] assert condition['product_territory_id'] assert condition['product_transaction_type_id'] assert condition['product_rate'] @patch('snapshot_contract.contract_builder.IsrcHelper') def test_apply_term_type_track(isrc_helper, mock_contract_term, mock_isrcs): """Test creating track specific dicts from "generic" term_conditions_list.""" contract_term = mock_contract_term.copy() contract_term.update({ 'is_base_term': 0, 'priority': 2, 'term_type': TRACK }) mock_conditions = mock_contract_term['conditions'] term_conditions_list = [ { 'attachment': mock_isrcs[0]['isrc'], 'country': mock_conditions['countries'][0], 'store': mock_conditions['stores'][0], 'txn_type': mock_conditions['transaction_types'][0] }, { 'attachment': mock_contract_term['attachments'][0], 'country': mock_conditions['countries'][1], 'store': mock_conditions['stores'][1], 'txn_type': mock_conditions['transaction_types'][0] }, { 'attachment': mock_contract_term['attachments'][1], 'country': mock_conditions['countries'][1], 'store': mock_conditions['stores'][0], 'txn_type': mock_conditions['transaction_types'][0] } ] contract_terms_df = pandas.DataFrame([contract_term]) index = 0 isrc_helper.find_isrc_id.return_value = mock_isrcs[0]['isrcId'] res = contract_builder._apply_term_type( contract_terms_df, index, isrc_helper, term_conditions_list ) assert len(res) == len(term_conditions_list) for condition in res: assert condition['priority'] assert condition['track_isrc_id'] assert condition['track_store_id'] assert condition['track_territory_id'] assert condition['track_transaction_type_id'] assert condition['track_rate'] @patch('snapshot_contract.contract_builder.Currency') def test_base_contract_detail(mock_currency, mock_contract): """Test base contract detail creates base dict of contract info.""" accounting_run_id = 12 contract_id = mock_contract['contract_id'] contract_df = pandas.DataFrame([mock_contract]) contract_row = contract_df.loc[contract_df['contract_id'] == contract_id] mock_currency_response = MagicMock() mock_currency_id = 840 mock_currency_response.number = mock_currency_id mock_currency.return_value = mock_currency_response res = contract_builder._base_contract_detail(accounting_run_id, contract_row) assert res['accounting_run_id'] == accounting_run_id assert res['contract_id'] == contract_id assert res['payee_currency_id'] == mock_currency_id assert res['payee_id'] == mock_contract['payee_id'] assert res['term_end'] == IMPOSSIBLE_DATE_FUTURE assert res['term_start'] == mock_contract['term_start'] assert len(res.keys()) > 10 mock_currency.assert_called_once_with(mock_contract['currency_code']) def test_combine_contract_conditions(mock_base_contract, mock_flat_contract_terms): """Test combining base contract data with each of it's flattened conditions.""" res = contract_builder._combine_contract_conditions( mock_base_contract, mock_flat_contract_terms ) assert len(res) == len(mock_flat_contract_terms) for contract in res: assert contract['accounting_run_id'] == mock_base_contract['accounting_run_id'] assert contract['contract_id'] == mock_base_contract['contract_id'] assert contract['payee_currency_id'] == \ mock_base_contract['payee_currency_id'] assert contract['payee_id'] == mock_base_contract['payee_id'] assert contract['priority'] assert contract['term_end'] == mock_base_contract['term_end'] assert contract['term_start'] == mock_base_contract['term_start'] assert res[0]['base_rate'] == mock_flat_contract_terms[0]['base_rate'] assert not res[0]['product_rate'] assert not res[0]['track_rate'] assert res[1]['product_rate'] == mock_flat_contract_terms[1]['product_rate'] assert not res[1]['base_rate'] assert not res[1]['track_rate'] assert res[2]['track_rate'] == mock_flat_contract_terms[2]['track_rate'] assert not res[2]['base_rate'] assert not res[2]['product_rate'] @patch('snapshot_contract.contract_builder.map_country_codes') def test_flatten_contract_term_conditions(mock_country_codes): """Test flattening nested lists of attachments and conditions.""" attachments = ['label_1', 'label_2'] conditions = { 'countries': ['CAD', 'MEX', 'USA'], 'stores': ['store_1', 'store_2'], 'transaction_types': ['txn_type_1'] } contract_terms = [{ 'attachments': json.dumps(attachments), 'conditions': json.dumps(conditions), }] contract_term_rows = pandas.DataFrame(contract_terms) country_map = {'CAD': 2, 'MEX': 7, 'USA': 1} mock_country_codes.return_value = country_map number_of_combos = len(attachments) * len(conditions['countries']) * \ len(conditions['stores']) * len(conditions['transaction_types']) res = contract_builder._flatten_contract_term_conditions(contract_term_rows, 0) assert len(res) == number_of_combos assert set([condition['attachment'] for condition in res]) == set(attachments) assert set([condition['country'] for condition in res]) == set(country_map.values()) assert set([condition['store'] for condition in res]) == set(conditions['stores']) assert set([condition['txn_type'] for condition in res]) == \ set(conditions['transaction_types']) mock_country_codes.assert_called_once_with(conditions['countries'])