"""Integration test configuration.""" import os from os.path import dirname, join from typing import Any from abacus_common_logic.connectors.database import db from dotenv import load_dotenv from py.xml import html import pytest from pytest_html.extras import url from sqlalchemy import text # Load environment variables from a .env file if present from payment.api import create_app from payment.config import Config from tests.integration.consts import headers from tests.integration.utils import auth from tests.integration.utils.generic_helper import generate_random_string from tests.integration.utils.ows_abacus_account_api_client import AbacusAccountAPIClient from tests.integration.utils.ows_abacus_state_api_client import AbacusStateAPIClient from tests.integration.utils.ows_payment_api_client import PaymentAPIClient dotenv_path = join(dirname(__file__), '.env') load_dotenv(dotenv_path) QA_BASE_URL = os.environ.get('QA_BASE_URL', 'http://localhost:6251') print(f'Using payment URL {QA_BASE_URL}') QA_ABACUS_STATE_BASE_URL = os.environ.get( 'QA_ABACUS_STATE_BASE_URL', 'http://localhost:6252' ) print(f'Using abacus state URL {QA_ABACUS_STATE_BASE_URL}') QA_ABACUS_ACCOUNT_BASE_URL = os.environ.get( 'QA_ABACUS_ACCOUNT_BASE_URL', 'http://localhost:6253' ) print(f'Using abacus account URL {QA_ABACUS_ACCOUNT_BASE_URL}') JIRA_PREFIX_URL = 'https://theorchard.atlassian.net/browse/' PYTEST_REPORT_PREFIX = 'ows-payment microservice integration tests results' PYTEST_REPORT_SUMMARY = 'integration tests of ows-payment microservice' def _exec_sql(sql): """Execute raw SQL using SA2-compatible engine.connect().""" with db.engine.connect() as conn: conn.execute(text(sql)) conn.commit() 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', 'mysql-royalties-container') 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(scope='session', autouse=True) def test_app_in_context(test_app): """Push the test app onto the context.""" with test_app.app_context(): yield test_app def ows_payment_api_client(headers): """Create ows-payment APIClient object.""" return PaymentAPIClient(QA_BASE_URL, headers) def ows_abacus_state_api_client(headers): """Create ows-abacus-state APIClient object.""" return AbacusStateAPIClient(QA_ABACUS_STATE_BASE_URL, headers) def ows_abacus_account_api_client(headers): """Create ows-abacus-account APIClient object.""" return AbacusAccountAPIClient(QA_ABACUS_ACCOUNT_BASE_URL, headers) @pytest.fixture(scope='session') def basic_headers(): """Return basic headers.""" return {'Content-Type': 'application/json'} @pytest.fixture def mock_account_payees_integration(): """Create mock account payees.""" account_payee_insert = """ INSERT INTO account_payee( account_payee_id, account_id, payoneer_program_id, payoneer_payee_id, payoneer_payee_name, payoneer_iframe_url, payoneer_iframe_url_date, payoneer_session_id, created_by, created_at, last_modified_by, last_modified ) VALUES (1, 123123, 1001, 1, NULL, NULL, NULL, NULL, 'Test', NOW(), 'Test', NOW()) """ _exec_sql(account_payee_insert) @pytest.fixture def mock_abacus_state(): """Create mock account payees.""" account_payee_insert = """ INSERT INTO abacus_state( `parent_table_name`, `parent_table_id`, `action_name`, `action_status`, `message`, `created_by`, `created_at`, `last_modified_by`, `last_modified` ) VALUES ('payment_group_payment', 1, 'generate_payments', 'complete', NULL, '2', NOW(), '2', NOW()), ('payment_group_payment', 1, 'generate_export', 'complete', NULL, '2', NOW(), '2', NOW()), ('payment_group_payment', 1, 'upload_approval', 'complete', NULL, '2', NOW(), '2', NOW()), ('payment_group_payment', 1, 'send_payments', 'complete', NULL, '2', NOW(), '2', NOW()), ('payment_group_payment', 2, 'generate_payments', 'complete', NULL, '2', NOW(), '2', NOW()), ('payment_group_payment', 2, 'generate_export', 'complete', NULL, '2', NOW(), '2', NOW()), ('payment_group_payment', 2, 'upload_approval', 'init', NULL, '2', NOW(), '2', NOW()), ('payment_group_payment', 2, 'send_payments', 'init', NULL, '2', NOW(), '2', NOW()), ('payment_group_payment_account', 1, 'send_payments', 'complete', NULL, '2', NOW(), '2', NOW()), ('payment_group_payment_account', 2, 'send_payments', 'init', NULL, '2', NOW(), '2', NOW()), ('payment_group_payment_account', 3, 'send_payments', 'error', 'Insufficient funds', '2', NOW(), '2', NOW()), ('payment_group_payment_batch', 1, 'send_payment', 'complete', NULL, '2', NOW(), '2', NOW()), ('payment_group_payment_batch', 2, 'send_payment', 'error', 'Batch Failed', '2', NOW(), '2', NOW()), ('payment_group_payment_batch', 3, 'send_payment', 'complete', NULL, '2', NOW(), '2', NOW()), ('worksheet_payment_contract_advance', 1, 'send_payments', 'init', NULL, '2', NOW(), '2', NOW()) """ _exec_sql(account_payee_insert) @pytest.fixture def mock_reference_payoneer_program_integration(): """Create mock reference_payoneer_program.""" reference_payoneer_program_insert = """ INSERT INTO reference_payoneer_program( payoneer_program_id, payoneer_program_name, funding_currency, reference_payment_type_id ) VALUES ('100158970', 'test_program', 'USD', 3), ('1001', 'test_program', 'AUS', 3); """ _exec_sql(reference_payoneer_program_insert) @pytest.fixture def mock_exchange_rate(): """Create mock exchange_rate.""" exchange_rate_insert = """ INSERT INTO exchange_rate ( exchange_rate_id, statement_period_id, rate, from_currency_code, to_currency_code, created_by, created_at, last_modified_by, last_modified ) VALUES ( 1, 300, 5.0000000000000000000, 'USD', 'GBP', 'integration_tests', '2024-02-23 15:46:34', 'integration_tests', '2024-02-23 15:46:39' ) """ _exec_sql(exchange_rate_insert) @pytest.fixture def drop_fks(): """Drop FKs.""" fks = [ """ALTER TABLE account_payee DROP FOREIGN KEY fk_payoneer_program_id;""", """ALTER TABLE payment_group_payment_batch DROP FOREIGN KEY fk_payment_batch_reference_payoneer_program;""", """ALTER TABLE account_payee DROP FOREIGN KEY fk_account_payee_account;""", """ALTER TABLE worksheet_payment_contract_advance DROP FOREIGN KEY fk_statement_period;""", ] for i in fks: try: _exec_sql(i) except Exception as ex: print('FK was already deleted: {}'.format(ex)) def generate_payment_group_body(): """Return payment_group body.""" name = 'orcd_autotest_{}'.format(generate_random_string(16)) return { 'group_name': name, 'payment_name': 'Payment Name', 'is_reusable': True, 'group_criteria': { 'currency_codes': ['CAD'], 'payment_schedules': ['30_days_after_month_end'], 'reference_payment_entities': [1], }, } def create_abacus_event(statement_period_id, target_id): """Insert abacus_event fixtures directly into the DB.""" mock_abacus_event_fixture = """ INSERT INTO `abacus_event` ( `abacus_event_id`, `statement_period_id`, `event_name`, `target_type`, `target_id`, `event_date`, `previous_abacus_event_id`, `rolled_back_at`, `created_by` ) VALUES ( 1, {}, 'payment_group_payment', 'payment_group_payment', {}, '2022-01-01', NULL, NULL, 'default_user_id' ) """.format(statement_period_id, target_id) # noqa _exec_sql(mock_abacus_event_fixture) def create_ledger_account_contract(event_id, account_id, contract_id): """Insert Ledger Account Contract test data into the DB.""" sql = """ INSERT INTO `ledger_account_contract`( ledger_account_contract_id, abacus_event_id, account_id, contract_id, currency_code, currency_amount, previous_balance, current_balance, note, created_by, created_at, last_modified_by, last_modified ) VALUES (1, {}, {}, {}, 'USD', '1000.00', '0.00', '1000.00', 'note', 'default_user_id', '2022-01-01', 'default_user_id', '2022-01-01') """.format(event_id, account_id, contract_id) _exec_sql(sql) def query_the_db(query): """Execute a query against local db.""" _exec_sql(query) def generate_abacus_states(payment_id): """Return POST /abacus-state body.""" table_name = 'payment_group_payment' return [ { 'action_name': 'generate_payments', 'parent_table_name': table_name, 'parent_table_id': payment_id, }, { 'action_name': 'generate_export', 'parent_table_name': table_name, 'parent_table_id': payment_id, }, { 'action_name': 'upload_approval', 'parent_table_name': table_name, 'parent_table_id': payment_id, }, { 'action_name': 'send_payments', 'parent_table_name': table_name, 'parent_table_id': payment_id, }, ] @pytest.fixture(autouse=True) def fresh_db(): """Refresh the test database.""" top_level_tables = ( 'abacus_event', 'abacus_state', 'account', 'account_payee', 'contract', 'contract_advance', 'account_contract', 'exchange_rate', 'payment_group', 'payment_group_payment', 'payment_group_payment_account', 'payment_group_payment_batch', 'payment_group_payment_batch_account', 'payment_group_payment_account_detail', 'payment_minimum', 'account_payment_term', 'reference_payoneer_program', 'ledger_account_contract', 'report_payment_group_payment', 'worksheet_payment_contract_advance', 'worksheet_account_contract_closing_balance', 'worksheet_account_contract_payable_after_tax', 'worksheet_account_contract_payable_details', 'worksheet_account_contract_taxable_revenue', 'worksheet_tax_correction', 'worksheet_tax_correction_vat', ) with db.engine.connect() as con: con.execute(text('SET FOREIGN_KEY_CHECKS = 0;')) for table_name in top_level_tables: con.execute(text(f'TRUNCATE TABLE `{table_name}`')) con.execute(text('SET FOREIGN_KEY_CHECKS = 1;')) con.commit() def create_account(headers, account_id=123123): """Create account fixture.""" ows_account_client = ows_abacus_account_api_client(headers) post_account = {'account_name': 'Test Account', 'account_id': account_id} response_post = ows_account_client.post_account(post_account) assert response_post.status_code == 201 def create_contract_fixture(account_id=123123, contract_id=123): """Create contract fixtures.""" contract_insert = """ INSERT INTO contract ( contract_id, reference_signing_entity_id, reference_sap_profit_center_id, contract_name, contract_type, sap_created_at, term_start, term_end, created_by, created_at, last_modified_by, last_modified ) VALUES ( {}, 1, 1, 'test', 'distribution', null, '2022-11-14', '2022-11-30', 'vz', '2022-11-14 02:05:15', 'vz', '2022-11-14 02:05:16' ) """.format(contract_id) _exec_sql(contract_insert) contract_account_insert = """ INSERT INTO account_contract ( account_contract_id, account_id, contract_id, is_primary_for_calc ) VALUES ( 1, {}, {}, null ) """.format(account_id, contract_id) _exec_sql(contract_account_insert) def create_contract_advance_fixture(): """Create contract_advance fixtures.""" contract_advance_insert = """ INSERT INTO contract_advance ( contract_advance_id, contract_id, reference_payment_type_id, reference_advance_payment_method_id, advance_description, amount, vat_amount, withholding_tax_amount, amount_after_withholding_and_vat, us_source_income_rate, currency_code, milestone, milestone_description, milestone_date, advance_status, note, created_by, created_at, last_modified_by, last_modified, deleted_at, deleted_by ) VALUES ( 1, 123, 1, 1, 'integration_tests', 42.00, 24.00, null, null, null, 'USD', 'Delivery', 'integration_tests', '2024-02-01', DEFAULT, 'testing', 'integration_tests', '2024-02-23 14:49:13', 'integration_tests', '2024-02-23 14:49:18', null, null ) """ _exec_sql(contract_advance_insert) def check_db_field_value(table_name, column_name, condition): """Retrieve a value of a specific entry from a table.""" with db.engine.connect() as con: try: query = f'SELECT `{column_name}` FROM `{table_name}` WHERE {condition}' result = con.execute(text(query)).fetchone() con.commit() if result: return result[0] else: return None except Exception as e: raise e def get_table_row_count(table_name): """Retrieve the row count from a given table.""" with db.engine.connect() as con: try: query = f'SELECT COUNT(*) FROM `{table_name}`' result = con.execute(text(query)).fetchone() con.commit() if result: return result[0] else: return 0 except Exception as e: raise e def create_payment_group_payment(headers): """Create payment_group_payment entry and related fixtures.""" ows_payment_client = ows_payment_api_client(headers) payment_group_body = generate_payment_group_body() response_post_payment_group = ows_payment_client.post_payment_group( payment_group_body ) assert response_post_payment_group.status_code == 201 payment_group_id = response_post_payment_group.json()['payment_group_id'] payment_group_payment_body = { 'payment_group_id': payment_group_id, 'payment_name': 'orcd_autotest_{}'.format(generate_random_string(16)), } response_post_group_payment = ows_payment_client.post_group_payment( payment_group_payment_body ) assert response_post_group_payment.status_code == 201 payment_group_payment_id = response_post_group_payment.json()[ 'payment_group_payment_id' ] return payment_group_payment_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) try: description = report.description except Exception: description = '' cells.insert(2, html.td(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 = [] @pytest.fixture(scope='session') def bearer_token_read_only_user() -> str: """Fetch Auth0 secrets from AWS Secret Manager.""" login_secrets = auth.login_from_secrets_manager( password_secret_name=auth.READONLY_USER_PASSWORD_SECRET_NAME, auth_client_id_secret_name=auth.AUTH0_CLIENT_ID_SECRET_NAME, auth_client_secret_secret_name=auth.AUTH0_CLIENT_SECRET_SECRET_NAME, ) token = auth.generate_auth_token( auth.LoginInfo( auth_client_id=login_secrets.auth_client_id, auth_client_secret=login_secrets.auth_client_secret, password=login_secrets.password, username=auth.READONLY_USER, ) ) return token @pytest.fixture(scope='session') def admin_headers() -> dict[str, str]: """Return profile headers.""" return headers.ADMIN_HEADERS @pytest.fixture(params=['read_only', 'admin'], scope='session') def auth_headers( request: Any, bearer_token_read_only_user: str, admin_headers: dict[str, str] ) -> dict[str, str]: """Return appropriate headers based on the parameter.""" if request.param == 'read_only': return { 'Authorization': f'Bearer {bearer_token_read_only_user}', 'Content-Type': 'application/json', 'Orchard-Identity-Id': 'ff3737b4-7734-4339-9798-5ddd1c9998b0', 'Orchard-Profile-Id': '1005', 'Orchard-Profile-Type': 'Account360Profile', 'Orchard-Requestor-Service': 'graphql-abacus', 'Orchard-Roles': 'account360', } else: return admin_headers @pytest.fixture def minimum_a360_read_only_user_headers( bearer_token_read_only_user: str, ) -> dict[str, str]: """Return minimum headers for A360 RO user.""" return { 'Authorization': f'Bearer {bearer_token_read_only_user}', 'Orchard-Requestor-Service': 'graphql-abacus', } @pytest.fixture() def unauthorized_headers(bearer_token_read_only_user_unauthorized) -> dict[str, str]: """Return headers for unauthorized user.""" return { 'Authorization': f'Bearer {bearer_token_read_only_user_unauthorized}', 'Orchard-Requestor-Service': 'graphql-abacus', } @pytest.fixture(scope='session') def bearer_token_read_only_user_unauthorized() -> str: """Fetch Auth0 secrets from AWS Secret Manager.""" login_secrets = auth.login_from_secrets_manager( password_secret_name=auth.READONLY_USER_PASSWORD_SECRET_NAME, auth_client_id_secret_name=auth.AUTH0_CLIENT_ID_SECRET_NAME, auth_client_secret_secret_name=auth.AUTH0_CLIENT_SECRET_SECRET_NAME, ) token = auth.generate_auth_token( auth.LoginInfo( auth_client_id=login_secrets.auth_client_id, auth_client_secret=login_secrets.auth_client_secret, password=login_secrets.password, username=auth.READONLY_USER_UNAUTHORIZED, ) ) return token @pytest.fixture def no_authorization_headers() -> dict[str, Any]: """Return headers that have no authorization headers present. The orchard-requestor-service is set because this triggers checking on authorization headers. """ return { 'Content-Type': 'application/json', 'orchard-requestor-service': 'graphql-abacus', } @pytest.fixture def some_other_app_profile_headers() -> dict[str, Any]: """Return some other profile headers.""" return { 'Content-Type': 'application/json', 'orchard-requestor-service': 'graphql-abacus', 'Orchard-Profile-Type': 'ContentProfile', 'Orchard-Profile-Id': '123456', 'Orchard-Roles': 'content-reviewer', } @pytest.fixture def abacus_headers() -> dict[str, Any]: """Return abacus profile headers.""" return { 'Content-Type': 'application/json', 'orchard-requestor-service': 'graphql-abacus', 'Orchard-Profile-Type': 'AbacusProfile', 'Orchard-Profile-Id': '123456', 'Orchard-Roles': 'administrator', } @pytest.fixture def a360_headers() -> dict[str, Any]: """Return a360 profile headers.""" return { 'Content-Type': 'application/json', 'orchard-requestor-service': 'graphql-abacus', 'Orchard-Profile-Type': 'Account360Profile', 'Orchard-Profile-Id': '123456', 'Orchard-Roles': 'account360', }