"""Fixtures for lambda testing.""" import os import textwrap import mysql.connector import pytest from py.xml import html from pytest_html.extras import url from utils.dockerized_lambda_api_client import DockerizedLambdaAPIClient QA_BASE_URL = os.environ.get( 'QA_BASE_URL', 'http://commit-ledger-account-contract-lambda:8080' ) JIRA_PREFIX_URL = 'https://theorchard.atlassian.net/browse/' PYTEST_REPORT_PREFIX = 'commit_ledger_account_contract lambda integration tests results' PYTEST_REPORT_SUMMARY = textwrap.dedent(""" Lambda for committing to ledger_account_contract (AKA "the general ledger"). """).splitlines() def dockerized_lambda_api_client(headers): """Create adjustments_apply dockerized lambda object.""" return DockerizedLambdaAPIClient(QA_BASE_URL, headers) def get_db_field_value(table, column, condition): """Retrieve the value of the specified column from a table based on a condition.""" cnx = mysql.connector.connect( host='commit-ledger-account-contract-mysql', port='3306', database='royalty_accounting', user='db_user', password='db_pass', ) cursor = cnx.cursor() query = 'SELECT {} FROM {} WHERE {}'.format(column, table, condition) cursor.execute(query) result = cursor.fetchone() if result is not None: value = result[0] else: value = None cursor.close() cnx.close() return value def query_db(query, values=None): """Execute a query against the db.""" cnx = mysql.connector.connect( host='commit-ledger-account-contract-mysql', port='3306', database='royalty_accounting', user='db_user', password='db_pass', ) cursor = cnx.cursor() try: if values: cursor.execute(query, values) else: cursor.execute(query) cnx.commit() return cursor.lastrowid finally: cursor.close() cnx.close() @pytest.fixture(scope='session') def basic_headers(): """Return basic headers.""" return {'Content-Type': 'application/json'} @pytest.fixture def clean_db(): """Clean tables between runs.""" tables = [ 'abacus_event', 'account', 'account_contract', 'contract', 'contract_advance', 'payment_group', 'payment_group_payment', 'payment_group_payment_account', 'payment_group_payment_account_detail', 'payment_group_payment_batch', 'payment_group_payment_batch_account', 'ledger_account', 'ledger_account_contract', 'ledger_contract_advance_applied', 'worksheet_account_contract_closing_balance', 'worksheet_account_contract_payable_after_tax', 'worksheet_payment_contract_advance', ] cnx = mysql.connector.connect( host='commit-ledger-account-contract-mysql', port='3306', database='royalty_accounting', user='db_user', password='db_pass', ) cursor = cnx.cursor() try: cursor.execute('SET FOREIGN_KEY_CHECKS = 0;') for table in tables: cursor.execute(f'TRUNCATE TABLE `{table}`') cursor.execute('SET FOREIGN_KEY_CHECKS = 1;') cnx.commit() finally: cursor.close() cnx.close() def create_abacus_event_fixture(event_name, target_type, target_id): """Insert an abacus_event into the db.""" query = """ INSERT INTO royalty_accounting.abacus_event ( event_name, target_type, target_id, statement_period_id, event_date, created_by ) VALUES ( %s, %s, %s, 300, NOW(), 'integration tests' ) """ return query_db(query, (event_name, target_type, target_id)) def create_account_fixture(account_id): """Insert an account into the db.""" query = """ INSERT INTO royalty_accounting.account ( account_id, account_name, created_by, created_at, last_modified_by, last_modified ) VALUES ( %s, 'TEST ACCOUNT', 'integration tests', NOW(), 'integration tests', NOW() ) """ return query_db(query, (account_id,)) def create_contract_fixture(): """Insert a contract into the db.""" query = """ INSERT INTO royalty_accounting.contract ( reference_signing_entity_id, contract_name, term_start, term_end, created_by, created_at, last_modified_by, last_modified ) VALUES ( 1, 'TEST CONTRACT', '2020-01-01', '2030-01-01', 'integration tests', NOW(), 'integration tests', NOW() ) """ return query_db(query) def create_account_contract_fixture(account_id, contract_id): """Insert an account_contract into the db.""" query = """ INSERT INTO royalty_accounting.account_contract ( account_id, contract_id ) VALUES ( %s, %s ) """ return query_db(query, (account_id, contract_id)) def create_payment_group_fixture(): """Insert a payment_group into the db.""" query = """ INSERT INTO royalty_accounting.payment_group ( group_name, created_at, created_by, last_modified, last_modified_by ) VALUES ( 'TEST PAYMENT GROUP', NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db(query) def create_payment_group_payment_fixture(payment_group_id): """Insert a payment_group_payment into the db.""" query = """ INSERT INTO royalty_accounting.payment_group_payment ( payment_group_id, statement_period_id, payment_name, created_at, created_by, last_modified, last_modified_by ) VALUES ( %s, 300, 'TEST PAYMENT GROUP PAYMENT', NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db(query, (payment_group_id,)) def create_payment_group_payment_batch_fixture(payment_group_payment_id): """Insert a payment_group_payment_batch into the db.""" query = """ INSERT INTO royalty_accounting.payment_group_payment_batch ( payment_group_payment_id, payoneer_program_id, batch_num, created_at, created_by, last_modified, last_modified_by ) VALUES ( %s, 100176930, 1, NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db(query, (payment_group_payment_id,)) def create_payment_group_payment_account_fixture(payment_group_payment_id, account_id): """Insert a payment_group_payment_account into the db.""" query = """ INSERT INTO royalty_accounting.payment_group_payment_account ( payment_group_payment_id, account_id, payoneer_program_id, current_statement_period_id, currency_code, last_payment, current_balance, balance_after_tax, created_at, created_by, last_modified, last_modified_by ) VALUES ( %s, %s, 100176930, 300, 'USD', 0.00, 100.00, 100.00, NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db(query, (payment_group_payment_id, account_id)) def create_payment_group_payment_batch_account_fixture( payment_group_payment_batch_id, payment_group_payment_account_id ): """Insert a payment_group_payment_batch_account into the db.""" query = """ INSERT INTO royalty_accounting.payment_group_payment_batch_account ( payment_group_payment_batch_id, payment_group_payment_account_id, created_at, created_by, last_modified, last_modified_by ) VALUES ( %s, %s, NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db( query, (payment_group_payment_batch_id, payment_group_payment_account_id) ) def create_ledger_account_contract_fixture(abacus_event_id, account_id, contract_id): """Insert a ledger_account_contract into the db.""" query = """ INSERT INTO royalty_accounting.ledger_account_contract ( abacus_event_id, account_id, contract_id, currency_code, currency_amount, previous_balance, current_balance, created_at, created_by, last_modified, last_modified_by ) VALUES ( %s, %s, %s, 'USD', 100.00, 0.00, 100.00, NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db(query, (abacus_event_id, account_id, contract_id)) def create_worksheet_account_contract_closing_balance_fixture( account_id, contract_id, abacus_event_id, ledger_account_contract_id ): """Insert a worksheet_account_contract_closing_balance into the db.""" query = """ INSERT INTO royalty_accounting.worksheet_account_contract_closing_balance ( account_id, contract_id, abacus_event_id, ledger_account_contract_id, reference_payment_entity_id, statement_period_id, currency_code, amount, created_at, created_by, last_modified, last_modified_by ) VALUES ( %s, %s, %s, %s, 2, 300, 'USD', 100.00, NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db( query, (account_id, contract_id, abacus_event_id, ledger_account_contract_id) ) def create_worksheet_account_contract_payable_after_tax_fixture( worksheet_account_contract_closing_balance_id, account_id, contract_id, abacus_event_id, ): """Insert a worksheet_account_contract_payable_after_tax into the db.""" query = """ INSERT INTO royalty_accounting.worksheet_account_contract_payable_after_tax ( worksheet_account_contract_closing_balance_id, account_id, contract_id, abacus_event_id, statement_period_id, payable_amount_pre_tax, payable_amount_post_tax, currency_code, country_of_tax_residence, created_at, created_by, last_modified, last_modified_by ) VALUES ( %s, %s, %s, %s, 300, 100.00, 100.00, 'USD', 'USA', NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db( query, ( worksheet_account_contract_closing_balance_id, account_id, contract_id, abacus_event_id, ), ) def create_payment_group_payment_account_detail_fixture( payment_group_payment_account_id, worksheet_account_contract_payable_after_tax_id, account_id, contract_id, ): """Insert a payment_group_payment_account_detail into the db.""" query = """ INSERT INTO royalty_accounting.payment_group_payment_account_detail ( payment_group_payment_account_id, worksheet_account_contract_payable_after_tax_id, account_id, contract_id, payable_amount_pre_tax, vat_amount, payable_amount_post_tax, currency_code, created_at, created_by, last_modified, last_modified_by ) VALUES ( %s, %s, %s, %s, 100.00, 20.00, 120.00, 'USD', NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db( query, ( payment_group_payment_account_id, worksheet_account_contract_payable_after_tax_id, account_id, contract_id, ), ) def create_contract_advance_fixture(contract_id): """Insert a contract_advance into the db.""" query = """ INSERT INTO royalty_accounting.contract_advance ( contract_id, advance_description, amount, currency_code, milestone, milestone_description, created_at, created_by, last_modified, last_modified_by ) VALUES ( %s, 'TEST CONTRACT ADVANCE', 100.00, 'USD', 'delivery', 'TEST', NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db(query, (contract_id,)) def create_worksheet_payment_contract_advance_fixture(contract_advance_id): """Insert a worksheet_payment_contract_advance into the db.""" query = """ INSERT INTO royalty_accounting.worksheet_payment_contract_advance ( contract_advance_id, statement_period_id, exchange_rate_statement_period_id, payment_name, amount, currency_code, amount_payee_currency, payee_currency_code, exchange_rate, withholding_tax_amount, vat_amount, amount_after_withholding_and_vat, withholding_tax_amount_payee_currency, vat_amount_payee_currency, amount_after_withholding_and_vat_payee_currency, created_at, created_by, last_modified, last_modified_by ) VALUES ( %s, 300, 300, 'TEST PAYMENT', 100.00, 'USD', 100.00, 'USD', 1, 0.00, 0.00, 100.00, 0.00, 0.00, 100.00, NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db(query, (contract_advance_id,)) def create_ledger_contract_advance_applied_fixture( contract_advance_id, worksheet_payment_contract_advance_id, abacus_event_id, account_id, contract_id, ): """Insert a ledger_contract_advance_applied into the db.""" query = """ INSERT INTO royalty_accounting.ledger_contract_advance_applied ( contract_advance_id, worksheet_payment_contract_advance_id, abacus_event_id, account_id, contract_id, statement_period_id, advance_amount, advance_currency_code, advance_amount_payee_currency, advance_payee_currency_code, created_at, created_by, last_modified, last_modified_by ) VALUES ( %s, %s, %s, %s, %s, 300, 100.00, 'USD', 100.00, 'USD', NOW(), 'integration tests', NOW(), 'integration tests' ) """ return query_db( query, ( contract_advance_id, worksheet_payment_contract_advance_id, abacus_event_id, account_id, contract_id, ), ) @pytest.mark.optionalhook def pytest_html_results_summary(prefix, summary, postfix): """Populate report with info and summary.""" prefix.extend([html.p(PYTEST_REPORT_PREFIX)]) summary.extend([html.p(PYTEST_REPORT_SUMMARY)]) def pytest_html_results_table_header(cells): """Create results table header.""" cells.insert(1, html.th('Jira ID')) cells.insert(2, html.th('Description')) cells.pop() def pytest_html_results_table_row(report, cells): """Populate results table row.""" jira_ids = getattr(report, 'jira_ids', []) jira_links = [] for index, item in enumerate(jira_ids): link = url('{}{}'.format(JIRA_PREFIX_URL, item), item) if index == 0: jira_links.append(html.a(link['name'], href=link['content'])) else: jira_links.append(', ') jira_links.append(html.a(link['name'], href=link['content'])) jira_links_html = html.td(*jira_links) cells.insert(1, jira_links_html) cells.insert(2, html.td(report.description)) cells.pop() @pytest.hookimpl(hookwrapper=True) def pytest_runtest_makereport(item, call): """Populate results table.""" outcome = yield report = outcome.get_result() report.description = str(item.function.__doc__) if item.get_closest_marker('jira') is not None: to_populate = [] for item in item.get_closest_marker('jira').args: to_populate.append(item) report.jira_ids = to_populate else: report.jira_id = []