"""Test for Report model.""" from decimal import Decimal from typing import NamedTuple from unittest.mock import ANY from oto import status import pytest from collaborator.constants import error from collaborator.models.rds import report_persister from collaborator.models.rds.report import Report from collaborator.utils.error import OwsError from collaborator.utils.typing import Account, Sort from tests.testutils import db @db.test_schema_default_seed def test_get_items(mock_account): """Test getting reports.""" result = report_persister.ReportPersister.get_items(account=mock_account) assert [report["id"] for report in result] == [2, 1] @db.test_schema_default_seed def test_get_items_filtered(mock_account): """Test getting reports with filters.""" result = report_persister.ReportPersister.get_items( account=mock_account, status="GENERATED" ) assert [report["id"] for report in result] == [1] @db.test_schema_default_seed def test_get_items_sorted(mock_account): """Test getting reports with filters.""" result = report_persister.ReportPersister.get_items( account=mock_account, sort=Sort("amount", "DESC") ) assert [report["id"] for report in result] == [1, 2] @db.test_schema_default_seed @pytest.mark.parametrize( "has_transaction, expected_ids", [ (True, [1]), (False, [2]), ], ) def test_get_items_has_transaction(has_transaction, expected_ids, mock_account): """Test getting reports that have or don't have transactions.""" result = report_persister.ReportPersister.get_items( account=mock_account, has_transaction=has_transaction ) assert [item["id"] for item in result] == expected_ids @db.test_schema_default_seed def test_get_items_has_transaction_ignores_deleted_transactions(): """Test reports that don't have transactions (ignoring deleted).""" result = report_persister.ReportPersister.get_items( account=Account(type="vendor", id=120594), has_transaction=False ) assert [item["id"] for item in result] == [6, 7, 8] @db.test_schema_default_seed @pytest.mark.parametrize( "report_id, expected_result", [ ( 1, { "id": 1, "report_run_id": 1, "collaborator_id": 1, "exclude_transaction_types": None, "filename": "test.xls", "file_location": "s3://test.xls", "amount": 1234.0, "currency": "NOK", "status": "GENERATED", "requested_datetime": "2019-09-08T07:06:05", "generated_datetime": "2019-10-09T08:07:06", "transaction_id": 2, "deleted_by": None, "deleted_datetime": None, "collaborator_dp_enabled": False, }, ) ], ) def test_get_item(report_id, expected_result, mock_account): """Test getting a report.""" result = report_persister.ReportPersister.get_item(mock_account, report_id) assert result == expected_result @db.test_schema_default_seed @pytest.mark.parametrize("report_id", [(None), ("9999")]) def test_get_item_not_found(report_id, mock_account): """Test getting a report that doesn't exist.""" with pytest.raises(OwsError) as error_info: report_persister.ReportPersister.get_item(mock_account, report_id) assert error_info.value.status == status.NOT_FOUND assert error_info.value.code == error.ERROR_CODE_REPORT_NOT_FOUND assert error_info.value.message == error.ERROR_MESSAGE_REPORT_NOT_FOUND @pytest.mark.parametrize( "expected_result", [ [ { "amount": None, "collaborator_id": None, "currency": None, "exclude_transaction_types": None, "file_location": "s3/somewhere", "filename": None, "generated_datetime": None, "id": 123, "report_run_id": None, "requested_datetime": None, "status": None, "transaction_id": None, "deleted_by": None, "deleted_datetime": None, "collaborator_dp_enabled": False, }, { "amount": None, "collaborator_id": None, "currency": None, "exclude_transaction_types": None, "file_location": "s3/somewhere", "filename": None, "generated_datetime": None, "id": 345, "report_run_id": None, "requested_datetime": None, "status": None, "transaction_id": None, "deleted_by": None, "deleted_datetime": None, "collaborator_dp_enabled": False, }, ] ], ) def test_bulk_delete_items(expected_result, mocker, mock_account, mock_user): """Test bulk delete reports.""" mock_rows = [ Report( report_id=123, file_location="s3/somewhere", collaborator_dp_enabled=False ), Report( report_id=345, file_location="s3/somewhere", collaborator_dp_enabled=False ), ] mock_session = mocker.MagicMock() mock_session.query().join().filter().all.return_value = mock_rows report_ids = [123, 345] mock_log_event = mocker.patch.object(report_persister.logging, "log_event") result = report_persister.ReportPersister.bulk_delete_items( mock_account, report_ids, mock_user, session=mock_session ) assert result == expected_result mock_session.query().filter().update.assert_called_with( {"deleted_datetime": ANY, "deleted_by": mock_user.id}, synchronize_session=False ) mock_log_event.assert_called_with( "delete", "report", 345, { "amount": None, "collaborator_id": None, "currency": None, "exclude_transaction_types": None, "file_location": "s3/somewhere", "filename": None, "generated_datetime": None, "id": 345, "report_run_id": None, "requested_datetime": None, "status": None, "transaction_id": None, "deleted_by": None, "deleted_datetime": None, "collaborator_dp_enabled": False, }, None, mock_user, ) def test_bulk_delete_items_invalid_ids(mocker, mock_account, mock_user): """Test bulk delete reports.""" mock_rows = [ Report( report_id=123, file_location="s3/somewhere", collaborator_dp_enabled=False ), Report( report_id=345, file_location="s3/somewhere", collaborator_dp_enabled=False ), ] mock_session = mocker.MagicMock() mock_session.query().join().filter().all.return_value = mock_rows report_ids = [123, 345, 678] mock_log_event = mocker.patch.object(report_persister.logging, "log_event") with pytest.raises(OwsError) as err: report_persister.ReportPersister.bulk_delete_items( mock_account, report_ids, mock_user, session=mock_session ) assert err.value.code == error.ERROR_CODE_INVALID_REPORT_IDS assert err.value.message == error.ERROR_MESSAGE_INVALID_REPORT_IDS mock_session.query().filter().update.assert_not_called() mock_log_event.assert_not_called() @pytest.mark.parametrize( "text,expected", [ ("420lol", "420lol"), (" Hello, world! ", "Hello_world_"), ("--~~> 🆒 <~~--", "--_______--"), ], ) def test_sanitize(text, expected): """Test sanitizing text.""" assert report_persister.sanitize_text(text) == expected @db.test_schema_default_seed @pytest.mark.parametrize( ["transactions", "expected_result"], [ ( [{"report_id": 1, "transaction_id": 3}], [ { "id": 1, "report_run_id": 1, "collaborator_id": 1, "transaction_id": 3, "exclude_transaction_types": None, "filename": "test.xls", "file_location": "s3://test.xls", "amount": 1234.0, "currency": "NOK", "status": "GENERATED", "requested_datetime": "2019-09-08T07:06:05", "generated_datetime": "2019-10-09T08:07:06", "deleted_datetime": None, "deleted_by": None, "collaborator_dp_enabled": False, } ], ) ], ) def test_update_reports_with_transactions(transactions, expected_result, mock_user): """Test bulk add transaction_id to report.""" result = report_persister.ReportPersister.update_reports_with_transactions( transactions, mock_user ) assert result == expected_result @db.test_schema_default_seed def test_update_report_status(): """Test update report status.""" report = report_persister.ReportPersister.get_item_with_report_id(1) assert report["status"] == "GENERATED" report_persister.ReportPersister.update_report_status(1, "ERROR") report = report_persister.ReportPersister.get_item_with_report_id(1) assert report["status"] == "ERROR" @db.test_schema_default_seed def test_remove_transaction_from_report(): """Test remove transaction_id from report.""" report = report_persister.ReportPersister.get_item_with_report_id(1) assert report["transaction_id"] == 2 report_persister.ReportPersister.remove_transaction_from_report(2) report = report_persister.ReportPersister.get_item_with_report_id(1) assert report["transaction_id"] is None @db.test_schema_default_seed @pytest.mark.parametrize( "report_id, expected_result", [ ( 1, { "id": 1, "report_run_id": 1, "collaborator_id": 1, "exclude_transaction_types": None, "filename": "test.xls", "file_location": "s3://test.xls", "amount": 1234.0, "currency": "NOK", "status": "GENERATED", "requested_datetime": "2019-09-08T07:06:05", "generated_datetime": "2019-10-09T08:07:06", "transaction_id": 2, "deleted_datetime": None, "deleted_by": None, "collaborator_dp_enabled": False, }, ) ], ) def test_get_item_with_report_id(report_id, expected_result): """Test get report with a report_id.""" result = report_persister.ReportPersister.get_item_with_report_id(report_id) assert result == expected_result @db.test_schema_default_seed def test_get_reports_with_report_ids(): """Test getting reports.""" result = report_persister.ReportPersister.get_reports_with_report_ids([1, 2, 3]) assert result == [ { "id": 1, "report_run_id": 1, "collaborator_id": 1, "transaction_id": 2, "exclude_transaction_types": None, "filename": "test.xls", "file_location": "s3://test.xls", "amount": 1234.0, "currency": "NOK", "status": "GENERATED", "requested_datetime": "2019-09-08T07:06:05", "generated_datetime": "2019-10-09T08:07:06", "deleted_datetime": None, "deleted_by": None, "collaborator_dp_enabled": False, }, { "id": 2, "report_run_id": 2, "collaborator_id": 2, "transaction_id": None, "exclude_transaction_types": None, "filename": "test2.xls", "file_location": None, "amount": None, "currency": None, "status": "REQUESTED", "requested_datetime": "2019-10-09T08:07:06", "generated_datetime": None, "deleted_by": None, "deleted_datetime": None, "collaborator_dp_enabled": False, }, { "id": 3, "report_run_id": 3, "collaborator_id": 3, "transaction_id": None, "exclude_transaction_types": None, "filename": "test3.xls", "file_location": None, "amount": None, "currency": None, "status": "ERROR", "requested_datetime": "2019-09-08T07:06:05", "generated_datetime": None, "deleted_datetime": None, "deleted_by": None, "collaborator_dp_enabled": False, }, ] class MockReportToReportLine(NamedTuple): """Mock report to report line.""" this_report_id: int comp_report_id: int this_period_ids: str comp_period_ids: str @pytest.mark.parametrize( ["mock_report", "mock_period_ids", "expected_overlaps"], [ ({"id": 1}, ["", ""], []), ({"id": 1}, ["1", "1"], [2]), ({"id": 1}, ["1,2,3", "2,3,4"], [2]), ({"id": 1}, ["2,3,4", "1,2,3"], [2]), ], ) def test_get_report_ids_with_overlapping_period( mocker, mock_report, mock_period_ids, expected_overlaps ): """Test updating reports with overlaps.""" mock_reports = [mock_report["id"]] mock_report_to_report_lines = [ MockReportToReportLine( this_report_id=mock_report["id"], comp_report_id=(1 if mock_report["id"] == 2 else 2), this_period_ids=mock_period_ids[0], comp_period_ids=mock_period_ids[1], ) ] mock_session = mocker.MagicMock() mock_session.query().join().join().join().filter().all.return_value = ( mock_report_to_report_lines ) overlaps = report_persister.ReportPersister.get_report_ids_with_overlapping_period( mock_reports, session=mock_session ) assert overlaps == {mock_report["id"]: expected_overlaps} @db.test_schema_default_seed def test_get_report_contract_subtotals(): """Test getting report contract subtotals.""" requested_report_ids = [1, 2] results = report_persister.ReportPersister.get_report_contract_subtotals( report_ids=requested_report_ids, ) assert [row._mapping for row in results] == [ { "report_id": 1, "contract_id": 1001, "currency": "USD", "amount": Decimal("500.000000"), "vendor_id": 24601, }, { "report_id": 1, "contract_id": 1002, "currency": "USD", "amount": Decimal("300.000000"), "vendor_id": 24601, }, { "report_id": 2, "contract_id": 1003, "currency": "EUR", "amount": Decimal("800.000000"), "vendor_id": 24601, }, ] @db.test_schema_default_seed @pytest.mark.parametrize( "report_run_id, collaborator_dp_enabled, expected_count, expected_amount", [ (1, None, 2, 800.0), # Report run 1 with no filter (1, False, 2, 800.0), # Report run 1 with dp_enabled=False (1, True, 0, 0.0), # Report run 1 with dp_enabled=True (no matches) (2, None, 1, 800.0), # Report run 2 with no filter (999, None, 0, 0.0), # Non-existent report run ], ) def test_get_report_contract_subtotal_aggregations( report_run_id, collaborator_dp_enabled, expected_count, expected_amount ): """Test getting report contract subtotal aggregations with various filters.""" result = report_persister.ReportPersister.get_report_contract_subtotal_aggregations( report_run_id=report_run_id, collaborator_dp_enabled=collaborator_dp_enabled ) assert result.total_count == expected_count assert float(result.currency_agnostic_total_amount or 0) == expected_amount