"""Fixtures for lambda testing.""" import os import mysql.connector from py.xml import html import pytest from pytest_html.extras import url from tests.integration.utils.dockerized_lambda_client import DockerizedLambdaAPIClient QA_BASE_URL = os.environ.get('QA_BASE_URL', 'http://payments-generate-lambda:8080') JIRA_PREFIX_URL = 'https://theorchard.atlassian.net/browse/' PYTEST_REPORT_PREFIX = 'commit_vat lambda integration tests results' PYTEST_REPORT_SUMMARY = 'This lambda generates a payment' def dockerized_lambda_api_client(headers): """Create snapshot_adjustments dockerized lambda object.""" return DockerizedLambdaAPIClient(QA_BASE_URL, headers) def mysqlclient_docker(): """Initialize mysql client for docker image.""" return mysql.connector.connect( host='payments-generate-mysql', port='3306', database='royalty_accounting', user='db_user', password='db_pass' ) @pytest.fixture(scope='session') def basic_headers(): """Return basic headers.""" return {'Content-Type': 'application/json'} @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 = []