"""Integration test configuration.""" import os import pytest from abacus_common_logic.connectors.database import db from abacus_worksheet.api import create_app from abacus_worksheet.config import Config from tests.integration.utils.ows_abacus_worksheet_api_client import ( AbacusWorksheetAPIClient, ) QA_BASE_URL = os.environ.get( 'QA_BASE_URL', 'http://abacus-worksheet-ows-abacus-worksheet:8080' ) class TestConfig(Config): """Test configuration.""" MYSQL_DB_NAME = os.environ.get('MYSQL_DB_NAME', Config.MYSQL_DB_NAME) MYSQL_DB_USER = os.environ.get('MYSQL_DB_USER', 'royalties') MYSQL_DB_HOST = os.environ.get('MYSQL_DB_HOST', 'abacus-worksheet-mysql') MYSQL_DB_PORT = os.environ.get('MYSQL_DB_PORT', '3306') MYSQL_DB_PASS = os.environ.get('MYSQL_DB_PASS', '1234') @pytest.fixture(scope='session', autouse=True) def test_app(): """Create a test application.""" return create_app(TestConfig) @pytest.fixture(autouse=True) def test_app_in_context(test_app): """Push the test app onto the context.""" with test_app.app_context(): yield test_app @pytest.fixture def basic_headers(): """Return basic headers.""" return {'Content-Type': 'application/json'} def ows_abacus_worksheet_api_client(basic_headers): """Create ows-abacus-worksheet client.""" return AbacusWorksheetAPIClient(QA_BASE_URL, basic_headers) @pytest.fixture def drop_fks(): """Drop FKs.""" fks = [ """ALTER TABLE worksheet_adjustment DROP FOREIGN KEY fk_worksheet_adjustment_account;""", """ALTER TABLE worksheet_adjustment DROP FOREIGN KEY fk_worksheet_adjustment_contract;""", """ALTER TABLE worksheet_adjustment_detail DROP FOREIGN KEY fk_worksheet_adjustment_detail_account;""", """ALTER TABLE worksheet_adjustment_detail DROP FOREIGN KEY fk_worksheet_adjustment_detail_contract;""", ] for i in fks: try: db.engine.execute(i) except Exception as ex: print('FK was already deleted: {}'.format(ex)) @pytest.fixture(autouse=True) def clear_db(): """Clean tables between runs.""" tables = [ 'account', 'account_contract', 'account_payment_term', 'abacus_event', 'abacus_state', 'contract', 'statement_period_adjustment_file', 'statement_period', 'ledger_correction', 'worksheet_adjustment', 'worksheet_adjustment_detail', 'worksheet_correction', ] con = db.engine.connect() con.execute('SET FOREIGN_KEY_CHECKS = 0;') trans = con.begin() for table in tables: con.execute(f'TRUNCATE TABLE `{table}`') trans.commit() con.execute('SET FOREIGN_KEY_CHECKS = 1;') @pytest.fixture def create_account_fixture(): """Create a DB fixture in account table.""" query = """ INSERT INTO royalty_accounting.account ( account_id, account_name, created_by, created_at, last_modified_by, last_modified ) VALUES ( 1, 'Test Account 1', 'Test', '2021-01-01 00:00:00', 'Test', '2021-01-01 00:00:00' ), ( 2, 'Yet Another Account', 'Test', '2021-01-01 00:00:00', 'Test', '2021-01-01 00:00:00' ) """ db.engine.execute(query) @pytest.fixture def create_contract_fixture(): """Create a DB fixture in contract table.""" contract = """ INSERT INTO contract ( `contract_id`, `reference_signing_entity_id`, `reference_sap_profit_center_id`, `contract_name`, `term_start`, `term_end`, `created_by`, `created_at`, `last_modified_by`, `last_modified` ) VALUES ( 1, 1, 1, 'test', '2021-05-06', '2021-05-27', 'vz', NOW(), 'vz', NOW() ), ( 2, 1, 1, 'yet another contract', '2021-05-06', '2021-05-27', 'vz', NOW(), 'vz', NOW() ); """ db.engine.execute(contract) @pytest.fixture def create_account_contract_fixture(): """Create a DB fixture in account_contract table.""" account_contract = """ INSERT INTO account_contract( `account_contract_id`, `account_id`, `contract_id` ) VALUES ( 1, 1, 1 ), ( 2, 2, 2 ); """ db.engine.execute(account_contract) @pytest.fixture def create_abacus_event_fixture(): """Create a DB fixture in abacus_event table.""" abacus_event = """ INSERT INTO royalty_accounting.abacus_event ( abacus_event_id, event_name, target_type, target_id, event_date, previous_abacus_event_id, rolled_back_at ) VALUES ( 1, 'apply_royalty_reversal', 'statement_period', 1, '2021-04-12 02:38:45.000000', null, null )""" db.engine.execute(abacus_event) @pytest.fixture def create_statement_period_fixture(): """Create a DB fixture in statement_period table.""" statement_period_insert = """ INSERT INTO royalty_accounting.statement_period( statement_period_name, statement_period_status, statement_month, statement_year, closed_date, closed_by ) VALUES ('Period 1', 'closed', NULL, NULL, '2022-01-01', 'default_user_id'), ('Period 2', 'closed', 1, 2022, '2023-01-01', 'default_user_id'), ('Period 3', 'closed', 2, 2022, '2023-01-01', 'default_user_id'), ('Period 4', 'closed', 3, 2022, '2023-01-01', 'default_user_id'), ('Period 5', 'closed', 4, 2022, '2023-01-01', 'default_user_id'), ('Period 6', 'closed', 5, 2022, '2023-01-01', 'default_user_id'), ('Period 7', 'closed', 6, 2022, '2023-01-01', 'default_user_id'), ('Period 8', 'closed', 7, 2022, '2023-01-01', 'default_user_id'), ('Period 9', 'closed', 8, 2022, '2023-01-01', 'default_user_id'), ('Period 10', 'closed', 9, 2022, '2023-01-01', 'default_user_id'), ('Period 11', 'closed', 10, 2022, '2023-01-01', 'default_user_id'), ('Period 12', 'closed', 11, 2022, '2023-01-01', 'default_user_id'), ('Period 13', 'closed', 12, 2022, '2023-01-01', 'default_user_id'), ('Period 14', 'open', 1, 2023, NULL, NULL), ('Period 15', 'open', 2, 2023, NULL, NULL), ('Period 16', 'open', 3, 2023, NULL, NULL), ('Period 17', 'open', 4, 2023, NULL, NULL), ('Period 18', 'open', 6, 2023, NULL, NULL), ('Period 19', 'open', 9, 2023, NULL, NULL), ('Period 20', 'open', 10, 2023, NULL, NULL), ('Period 21', 'open', 11, 2023, NULL, NULL), ('Period 22', 'open', 10, 2023, NULL, NULL), ('Period 23', 'open', 1, 2024, NULL, NULL), ('Period 24', 'open', 2, 2024, NULL, NULL), ('Period 25', 'open', 4, 2024, NULL, NULL), ('Period 26', 'open', 6, 2024, NULL, NULL), ('Period 27', 'open', 9, 2024, NULL, NULL), ('Period 28', 'open', 11, 2024, NULL, NULL), ('Period 29', 'open', 12, 2024, NULL, NULL) """ db.engine.execute(statement_period_insert) def create_ledger_correction_fixture(): """Create a DB fixture in ledger_correction table.""" ledger_correction_insert = """ INSERT INTO royalty_accounting.ledger_correction( ledger_correction_id, abacus_event_id, worksheet_correction_id, account_id, contract_id, statement_period_id, correction_statement_period_id, correction_type, currency_code, gross_revenue, distribution_fee, net_revenue, note, created_at, created_by, last_modified, last_modified_by ) VALUES ( 1, 1, 1, 1, 1, 1, 1, 'royalty_correction', 'USD', 200.00, 0.00, 200.00, 'vz', '2023-05-04 17:06:33', 'vz', '2023-05-04 17:06:35', 'vz' )""" db.engine.execute(ledger_correction_insert) @pytest.fixture def create_statement_period_adjustment_file_fixture(): """Create a DB fixture in statement_period_adjustment_file table.""" statement_period_adjustment_file_insert = """ INSERT INTO royalty_accounting.statement_period_adjustment_file( statement_period_adjustment_file_id, statement_period_id, file_name, valid_file_location, invalid_file_location, valid_row_count, invalid_row_count, md5sum, created_by, created_at, last_modified_by, last_modified, deleted_by, deleted_at ) VALUES ( 1, 1, 'test', 's3://qa-abacus-adjustments/test_fixtures/valid_adjustments.xlsx', null, null, null, null, 'vz', '2023-10-09 13:48:10', 'vz', '2023-10-09 13:48:11', null, null ), ( 2, 1, 'test', 's3://qa-abacus-adjustments/test_fixtures/valid_adjustments.xlsx', null, null, null, null, 'vz', '2023-10-09 13:48:10', 'vz', '2023-10-09 13:48:11', '2023-12-09 13:48:11', '2023-12-09 13:48:11' )""" db.engine.execute(statement_period_adjustment_file_insert) @pytest.fixture def create_worksheet_adjustment_fixture(): """Create a DB fixture in worksheet_adjustment table.""" worksheet_adjustment_insert = """ INSERT INTO royalty_accounting.worksheet_adjustment( worksheet_adjustment_id, statement_period_adjustment_file_id, abacus_event_id, account_id, contract_id, activity_statement_period_id, apply_to_statement_period_id, reference_adjustment_type_id, adjustment_amount, adjustment_currency_code, note, internal_note, created_by, created_at, last_modified_by, last_modified, deleted_by, deleted_at ) VALUES ( 1, 1, 1, 1, 1, 1, 1, 1, 4.00, 'USD', null, null, 'vz', '2023-10-09 13:49:00', 'vz', '2023-10-09 13:49:01', null, null ), ( 2, 2, 1, 1, 1, 1, 1, 1, 4.00, 'USD', null, null, 'vz', '2023-10-09 13:49:00', 'vz', '2023-10-09 13:49:01', null, null ), ( 3, 1, 1, 2, 2, 1, 1, 1, 7.00, 'USD', null, null, 'vz', '2023-10-09 13:49:00', 'vz', '2023-10-09 13:49:01', null, null )""" db.engine.execute(worksheet_adjustment_insert) @pytest.fixture def create_worksheet_adjustment_detail_fixture(): """Create a DB fixture in worksheet_adjustment_detail table.""" worksheet_adjustment_detail_insert = """ INSERT INTO royalty_accounting.worksheet_adjustment_detail( worksheet_adjustment_detail_id, statement_period_adjustment_file_id, worksheet_adjustment_id, account_id, contract_id, activity_statement_period_id, apply_to_statement_period_id, reference_adjustment_type_id, currency_code, amount, upc, distribution_type, note, internal_note, created_by, created_at, last_modified_by, last_modified, deleted_by, deleted_at ) VALUES ( 1, 1, 1, 1, 1, 1, 1, 1, 'USD', 123.00, 'TEST124123451', 'digital', 'e2e_test', 'e2e_test', 'e2e', '2023-10-19 15:28:11', 'e2e', '2023-10-19 15:28:13', null, null ), ( 2, 2, 2, 1, 1, 1, 1, 1, 'USD', 321.00, 'TEST_new', 'digital', 'e2e_test', 'e2e_test', 'e2e', '2023-10-19 15:28:11', 'e2e', '2023-10-19 15:28:13', null, null ), ( 3, 1, 3, 2, 2, 1, 1, 1, 'USD', 321.00, 'TEST_new', 'digital', 'e2e_test', 'e2e_test', 'e2e', '2023-10-19 15:28:11', 'e2e', '2023-10-19 15:28:13', null, null )""" db.engine.execute(worksheet_adjustment_detail_insert) @pytest.fixture def create_account_payment_term_fixture(): """Create a DB fixture in account_payment_term table.""" account_payment_term_insert = """ INSERT INTO account_payment_term( `account_payment_term_id`, `account_id`, `currency_code`, `payment_minimum`, `payment_entity_id`, `payment_schedule`, `created_by`, `created_at`, `last_modified_by`, `last_modified` ) VALUES (1, 1, 'AUD', 42.00, 1, null, 'Test', NOW(), 'Test', NOW()) """ db.engine.execute(account_payment_term_insert) def check_db_empty(compare_func, table='worksheet_adjustment'): """Check whether table is empty.""" rowcount = db.engine.execute('SELECT * FROM {}'.format(table)).rowcount assert compare_func(rowcount, 0)