import gzip import io import math import zipfile from datetime import date from itertools import islice from os import path from pathlib import Path from unittest.mock import ANY, MagicMock, patch import openpyxl import pytest from process.constants import XLSX_REPORTS_FEATURE from process.process import add_reports_to_royalties, generate_reports, process from process.utils.report_writer import Report from process.utils.transaction import Transaction MOCK_REPORT_RUN_UUID = "test_run" MOCK_BUCKET = "test_bucket" MOCK_KEY = f"{MOCK_REPORT_RUN_UUID}.xlsx" MOCK_REPORT_INFO = {"collaborator_1": {"id": 1, "name": "Test Collaborator"}} MOCK_REPORT = Report( report_id=1, collaborator_id=1, collaborator_name="collab_1", vendor_id=1, total="100.00", currency="USD", filename="report_1", report_run_uuid=MOCK_REPORT_RUN_UUID, number_format="us", created_date=(2023, 10, 1, 12, 0, 0), contract_totals={}, ) MOCK_REPORT_RUN = { "name": "report_run", "period_name": "period_name", "number_format": "us", "trigger_type": "MANUAL", "notification_email": None, } MOCK_REPORT_RUN_WITH_EMAIL = {**MOCK_REPORT_RUN, "notification_email": "test@example.com"} MOCK_AUTO_REPORT_RUN = { "name": "report_run", "period_name": "period_name", "number_format": "us", "trigger_type": "AUTO", "notification_email": None, } MOCK_AUTO_REPORT_RUN_WITH_EMAIL = {**MOCK_AUTO_REPORT_RUN, "notification_email": "test@example.com"} EXPECTED_EARNINGS_BY_COLLABORATOR_ID = { "15583": 0.4646767405, "15584": 1.028617435, "15585": 3.077286369, "15586": 0.8388146177, "15587": 0.8911198898, "15588": 0.4074387432, "15589": 2.224163233, "15590": 0.07413101551, "15591": 0.5289317924, "15592": 0.04954295259, "15593": 0.001527298723, "15594": 31.943365045, "15595": 0.004858539753, "15596": 0.01692864787, "15597": 0.09033065632, "15598": 1.45917107, "15599": 0.9621253459, "15600": 0.06971734562, "15613": 0.004139710127, "15661": 0.01775926781, "15662": 0.06659168496, "15663": 0.00539150331, "15664": 0.005691086701, "15672": 0.1149138866, "15673": 1.362660309, "15674": 0.01710125369, "15675": 0.0004301277057, "15676": 0.2658464574, "15677": 0.0550312077, "15678": 1.055967169, "15679": 0.01732486667, "15680": 0.01753858894, "15681": 0.08665939364, "15746": 0.03292727831, "15747": 0.03799899477, "15748": 0.03855176052, "15749": 0.1400551149, "15750": 0.4726858048, "15751": 0.1233352435, "15752": 0.2064891912, "15753": 0.01408317435, "15754": 0.01408317435, "15755": 0.1015033543, "15756": 0.2517728394, "15757": 0.03596754849, } def _make_mock_report(collab_id: str): return { "id": "1", "vendor_id": "1", "currency": "GBP", "collaborator_internal_id": "1", "collaborator_name": f"collab_{collab_id}", "filename": f"report_{collab_id}", "status": "REQUESTED", } def _verify_report_xlsx(xlsx_filename: str, key: str, expected_tab_count: int): report_basename = path.splitext(path.basename(key.split("/")[-1]))[0] collab_id = report_basename.split("report_")[1] def unpad(row): """Removes None padding from end of row.""" return tuple(col for col in row if col is not None) with open(xlsx_filename, "rb") as f: workbook = openpyxl.load_workbook(f) sheet = workbook.active assert sheet is not None iter = sheet.iter_rows(values_only=True) assert len(workbook.sheetnames) == expected_tab_count # Assert the general shape of the summary summary = [ unpad(item) for item in islice(iter, 9) ] # get the next 9 items from the iterator assert summary == [ ("Report Run Name", MOCK_REPORT_RUN["name"]), ("Report Date", ANY), ("Report Accounting Period", "period_name"), (), ("Label ID", "1"), ("Collaborator Name", f"collab_{collab_id}"), ("Collaborator Internal ID", "1"), ("Earned This Report", ANY, "GBP"), (), ] # Check the earnings match - we use round w/ isclose to account for weird # string -> float issues. earnings_row = summary[7] expected_earnings = round(EXPECTED_EARNINGS_BY_COLLABORATOR_ID[collab_id], 6) reported_earnings = round(float(earnings_row[1]), 6) assert math.isclose(expected_earnings, reported_earnings) def _verify_report_csv(zip_filename: str, key: str, expected_file_count: int): """Verify a CSV-path report: the upload is a zip of tab-delimited .xls files.""" report_basename = path.splitext(path.basename(key.split("/")[-1]))[0] collab_id = report_basename.split("report_")[1] with open(zip_filename, "rb") as f: archive = zipfile.ZipFile(io.BytesIO(f.read())) names = archive.namelist() assert len(names) == expected_file_count text = archive.read(names[0]).decode("utf-8") # Every cell is wrapped in double-quotes and tab-terminated. rows = [[cell.strip('"') for cell in line.split("\t")] for line in text.split("\n")] assert rows[0][0] == "Report Run Name" assert rows[0][1] == MOCK_REPORT_RUN["name"] assert rows[4][0] == "Label ID" assert rows[7][0] == "Earned This Report" expected_earnings = round(EXPECTED_EARNINGS_BY_COLLABORATOR_ID[collab_id], 6) reported_earnings = round(float(rows[7][1]), 6) assert math.isclose(expected_earnings, reported_earnings) @patch("process.process.get_report_run_info") @patch("process.process.get_report_info_by_collaborator_id") @patch("process.process.s3") @patch("process.process.add_reports_to_royalties") @patch("process.utils.db.mysql_connection") def test_csv_output( mysql_connection_mock, add_reports_to_royalties_mock, s3_mock, get_report_info_by_collaborator_id_mock, get_report_run_info_mock, mock_features, ): """CSV/zip output path.""" def upload_file_mock(zip_filename: str, bucket: str, key: str): return _verify_report_csv(zip_filename, key, expected_file_count=1) # XLSX feature deliberately disabled -> CsvReportWriter is used. add_reports_to_royalties_mock.return_value = None get_report_run_info_mock.return_value = MOCK_REPORT_RUN mock_report_info_by_collaborator_id = MagicMock() mock_report_info_by_collaborator_id.items = lambda: [] mock_report_info_by_collaborator_id.pop = lambda cid, _: _make_mock_report(cid) get_report_info_by_collaborator_id_mock.return_value = mock_report_info_by_collaborator_id mock_results_fixture_path = Path(__file__).parent / "input/test.csv" mock_results_gz_file = io.BytesIO() with ( open(mock_results_fixture_path, "rb") as mock_results_file, gzip.GzipFile(fileobj=mock_results_gz_file, mode="wb") as mock_results_gz, ): mock_results_gz.write(mock_results_file.read()) mock_results_gz_file.seek(0) s3_mock.get_object.return_value = mock_results_gz_file s3_mock.upload_file = MagicMock() s3_mock.upload_file.side_effect = upload_file_mock process( "bucket", f"{MOCK_REPORT_RUN_UUID}.xlsx", skip_update=True, skip_notification=True, skip_upload=False, # intercepted by s3_mock ) s3_mock.upload_file.assert_called() @pytest.mark.parametrize( "mock_report_run, limit_output_row_count", [ (MOCK_REPORT_RUN, None), (MOCK_AUTO_REPORT_RUN, None), (MOCK_REPORT_RUN, 50), (MOCK_AUTO_REPORT_RUN, 50), ], ) @patch("process.process.get_report_run_info") @patch("process.process.get_report_info_by_collaborator_id") @patch("process.process.s3") @patch("process.process.add_reports_to_royalties") @patch("process.utils.db.mysql_connection") def test_xlsx_output( mysql_connection_mock, add_reports_to_royalties_mock, s3_mock, get_report_info_by_collaborator_id_mock, get_report_run_info_mock, mock_features, mocker, mock_report_run, limit_output_row_count, ): if limit_output_row_count is not None: mocker.patch("process.utils.report_writer.MAX_OUTPUT_ROW_COUNT", limit_output_row_count) # Define upload mock for report output verification def upload_file_mock(xlsx_filename: str, bucket: str, key: str): if not limit_output_row_count: return _verify_report_xlsx(xlsx_filename, key, expected_tab_count=1) elif s3_mock.upload_file.call_count == 1: # Only check the first file, as they will all differ! return _verify_report_xlsx(xlsx_filename, key, expected_tab_count=4) # Set up mocks mock_features({XLSX_REPORTS_FEATURE: True}) add_reports_to_royalties_mock.return_value = None get_report_run_info_mock.return_value = mock_report_run mock_report_info_by_collaborator_id = MagicMock() mock_report_info_by_collaborator_id.items = lambda: [] mock_report_info_by_collaborator_id.pop = lambda cid, _: _make_mock_report(cid) get_report_info_by_collaborator_id_mock.return_value = mock_report_info_by_collaborator_id # Create gzipped query results file from fixture # Based on report run 6f29778e-dc46-4510-8145-827a8b7c5dd9 mock_results_fixture_path = Path(__file__).parent / "input/test.csv" mock_results_gz_file = io.BytesIO() with ( open(mock_results_fixture_path, "rb") as mock_results_file, gzip.GzipFile(fileobj=mock_results_gz_file, mode="wb") as mock_results_gz, ): mock_results_gz.write(mock_results_file.read()) mock_results_gz_file.seek(0) # Set gzipped file as the S3 get_object return value s3_mock.get_object.return_value = mock_results_gz_file s3_mock.upload_file = MagicMock() s3_mock.upload_file.side_effect = upload_file_mock process( "bucket", f"{MOCK_REPORT_RUN_UUID}.xlsx", skip_update=True, skip_notification=True, skip_upload=False, # This is intercepted by s3_mock ) s3_mock.upload_file.assert_called() ( add_reports_to_royalties_mock.assert_called() if mock_report_run["trigger_type"] == "AUTO" else add_reports_to_royalties_mock.assert_not_called() ) @patch("process.process.create_transactions") @patch("process.process.get_open_statement_period_ids_for_vendors") def test_add_reports_to_royalties( mock_get_open_statement_period_ids_for_vendors, mock_create_transactions ): """Test that add_reports_to_royalties is called with the correct parameters.""" mock_reports = [ Report( report_id=1, collaborator_id=1, collaborator_name="collab_1", vendor_id=1, total="100.00", currency="USD", filename="report_1", report_run_uuid=MOCK_REPORT_RUN_UUID, number_format="us", created_date=(2023, 10, 1, 12, 0, 0), contract_totals={}, ), Report( report_id=2, collaborator_id=2, collaborator_name="collab_2", vendor_id=1, total="200.00", currency="USD", filename="report_2", report_run_uuid=MOCK_REPORT_RUN_UUID, number_format="us", created_date=(2023, 10, 1, 12, 0, 0), contract_totals={}, ), ] mock_report_run = { "name": "test_run", "period_name": "test_period", "number_format": "us", "trigger_type": "MANUAL", } mock_get_open_statement_period_ids_for_vendors.return_value = { 1: 123, # Mock statement period ID for vendor_id 1 } date_today = date.today().strftime("%Y-%m-%d") add_reports_to_royalties( mock_reports, mock_report_run, ) mock_create_transactions.assert_called_once_with( [ Transaction( collaborator_id=1, report_id=1, statement_period_id=123, type="REVENUE", original_amount="100.00", chargeable_amount="100.00", currency="USD", date=date_today, description=f"test_run_collab_1_test_period_{date_today}", creation_batch_uuid=ANY, ), Transaction( collaborator_id=2, report_id=2, statement_period_id=123, type="REVENUE", original_amount="200.00", chargeable_amount="200.00", currency="USD", date=date_today, description=f"test_run_collab_2_test_period_{date_today}", creation_batch_uuid=ANY, ), ] ) @patch("process.process.send_notification_email") @patch("process.process.add_reports_to_royalties") @patch("process.process.create_empty_reports") @patch("process.process.create_reports_from_query_results") @patch("process.process.clear_report_contract") @patch("process.process.get_report_info_by_collaborator_id") @patch("process.process.get_report_run_info") def test_generate_reports( mock_get_report_run_info, mock_get_report_info_by_collaborator_id, mock_clear_report_contract, mock_create_reports_from_query_results, mock_create_empty_reports, mock_add_reports_to_royalties, mock_send_notification_email, ): """Test generate_reports for manual report runs.""" mock_report_run = MOCK_REPORT_RUN_WITH_EMAIL mock_get_report_run_info.return_value = mock_report_run mock_get_report_info_by_collaborator_id.return_value = MOCK_REPORT_INFO mock_create_reports_from_query_results.return_value = [mock_report_run] mock_create_empty_reports.return_value = [] generate_reports( bucket=MOCK_BUCKET, key=MOCK_KEY, skip_update=False, skip_upload=False, skip_notification=False, skip_add_to_royalties=False, ) mock_get_report_run_info.assert_called_once_with(MOCK_REPORT_RUN_UUID) mock_get_report_info_by_collaborator_id.assert_called_once_with(MOCK_REPORT_RUN_UUID) mock_clear_report_contract.assert_called_once_with(MOCK_REPORT_RUN_UUID) mock_create_reports_from_query_results.assert_called_once_with( bucket=MOCK_BUCKET, key=MOCK_KEY, report_run_uuid=MOCK_REPORT_RUN_UUID, report_run_name=mock_report_run["name"], number_format=mock_report_run["number_format"], period_name=mock_report_run["period_name"], report_info_by_collaborator_id=MOCK_REPORT_INFO, skip_upload=False, skip_update=False, ) mock_create_empty_reports.assert_called_once_with( report_run_uuid=MOCK_REPORT_RUN_UUID, report_run_name=mock_report_run["name"], number_format=mock_report_run["number_format"], period_name=mock_report_run["period_name"], report_info_by_collaborator_id=MOCK_REPORT_INFO, skip_upload=False, skip_update=False, ) mock_add_reports_to_royalties.assert_not_called() mock_send_notification_email.assert_called_once_with("test@example.com", "report_run") @patch("process.process.send_notification_email") @patch("process.process.add_reports_to_royalties") @patch("process.process.create_empty_reports") @patch("process.process.create_reports_from_query_results") @patch("process.process.clear_report_contract") @patch("process.process.get_report_info_by_collaborator_id") @patch("process.process.get_report_run_info") def test_generate_reports_auto_report_run( mock_get_report_run_info, mock_get_report_info_by_collaborator_id, mock_clear_report_contract, mock_create_reports_from_query_results, mock_create_empty_reports, mock_add_reports_to_royalties, mock_send_notification_email, ): """Test generate_reports with auto trigger, adds royalties and transaction fees.""" mock_report_run = MOCK_AUTO_REPORT_RUN_WITH_EMAIL mock_get_report_run_info.return_value = mock_report_run mock_get_report_info_by_collaborator_id.return_value = MOCK_REPORT_INFO mock_create_reports_from_query_results.return_value = [mock_report_run] mock_create_empty_reports.return_value = [] generate_reports( bucket=MOCK_BUCKET, key=MOCK_KEY, skip_update=False, skip_upload=False, skip_notification=False, skip_add_to_royalties=False, ) mock_get_report_run_info.assert_called_once_with(MOCK_REPORT_RUN_UUID) mock_get_report_info_by_collaborator_id.assert_called_once_with(MOCK_REPORT_RUN_UUID) mock_clear_report_contract.assert_called_once_with(MOCK_REPORT_RUN_UUID) mock_create_reports_from_query_results.assert_called_once_with( bucket=MOCK_BUCKET, key=MOCK_KEY, report_run_uuid=MOCK_REPORT_RUN_UUID, report_run_name=mock_report_run["name"], number_format=mock_report_run["number_format"], period_name=mock_report_run["period_name"], report_info_by_collaborator_id=MOCK_REPORT_INFO, skip_upload=False, skip_update=False, ) mock_create_empty_reports.assert_called_once_with( report_run_uuid=MOCK_REPORT_RUN_UUID, report_run_name=mock_report_run["name"], number_format=mock_report_run["number_format"], period_name=mock_report_run["period_name"], report_info_by_collaborator_id=MOCK_REPORT_INFO, skip_upload=False, skip_update=False, ) mock_add_reports_to_royalties.assert_called_once_with([mock_report_run], mock_report_run) mock_send_notification_email.assert_called_once_with("test@example.com", "report_run") @patch("process.process.send_notification_email") @patch("process.process.add_reports_to_royalties") @patch("process.process.create_empty_reports") @patch("process.process.create_reports_from_query_results") @patch("process.process.clear_report_contract") @patch("process.process.get_report_info_by_collaborator_id") @patch("process.process.get_report_run_info") def test_generate_reports_skip_all_flags( mock_get_report_run_info, mock_get_report_info_by_collaborator_id, mock_clear_report_contract, mock_create_reports_from_query_results, mock_create_empty_reports, mock_add_reports_to_royalties, mock_send_notification_email, ): """Test generate_reports with all skip flags enabled.""" mock_report_run = MOCK_AUTO_REPORT_RUN_WITH_EMAIL mock_get_report_run_info.return_value = mock_report_run mock_get_report_info_by_collaborator_id.return_value = MOCK_REPORT_INFO mock_create_reports_from_query_results.return_value = [mock_report_run] mock_create_empty_reports.return_value = [] generate_reports( bucket=MOCK_BUCKET, key=MOCK_KEY, skip_update=True, skip_upload=True, skip_notification=True, skip_add_to_royalties=True, ) mock_get_report_run_info.assert_called_once_with(MOCK_REPORT_RUN_UUID) mock_get_report_info_by_collaborator_id.assert_called_once_with(MOCK_REPORT_RUN_UUID) mock_clear_report_contract.assert_not_called() mock_create_reports_from_query_results.assert_called_once_with( bucket=MOCK_BUCKET, key=MOCK_KEY, report_run_uuid=MOCK_REPORT_RUN_UUID, report_run_name=mock_report_run["name"], number_format=mock_report_run["number_format"], period_name=mock_report_run["period_name"], report_info_by_collaborator_id=MOCK_REPORT_INFO, skip_upload=True, skip_update=True, ) mock_create_empty_reports.assert_called_once_with( report_run_uuid=MOCK_REPORT_RUN_UUID, report_run_name=mock_report_run["name"], number_format=mock_report_run["number_format"], period_name=mock_report_run["period_name"], report_info_by_collaborator_id=MOCK_REPORT_INFO, skip_upload=True, skip_update=True, ) mock_add_reports_to_royalties.assert_not_called() mock_send_notification_email.assert_not_called() @patch("process.process.send_notification_email") @patch("process.process.add_reports_to_royalties") @patch("process.process.create_empty_reports") @patch("process.process.create_reports_from_query_results") @patch("process.process.clear_report_contract") @patch("process.process.get_report_info_by_collaborator_id") @patch("process.process.get_report_run_info") def test_generate_reports_no_notification_email( mock_get_report_run_info, mock_get_report_info_by_collaborator_id, mock_clear_report_contract, mock_create_reports_from_query_results, mock_create_empty_reports, mock_add_reports_to_royalties, mock_send_notification_email, ): """Test generate_reports when notification_email is None.""" mock_report_run = MOCK_REPORT_RUN mock_get_report_run_info.return_value = mock_report_run mock_get_report_info_by_collaborator_id.return_value = MOCK_REPORT_INFO mock_create_reports_from_query_results.return_value = [MOCK_REPORT_RUN] mock_create_empty_reports.return_value = [] generate_reports( bucket=MOCK_BUCKET, key=MOCK_KEY, skip_update=False, skip_upload=False, skip_notification=False, skip_add_to_royalties=False, ) mock_send_notification_email.assert_not_called()