"""Tests for the TransactionPersister class.""" from datetime import date, datetime from decimal import Decimal from unittest.mock import ANY from uuid import uuid4 import pytest from collaborator.constants import error from collaborator.constants import transaction as constants from collaborator.models.rds.transaction_persister import TransactionPersister from collaborator.utils.error import OwsError from tests.testutils import db, transaction_fixtures @db.test_schema_default_seed def test_create_transaction(): """Test creating a transaction.""" payload = { "collaborator_id": 1, "transaction_type": constants.TYPE_REVENUE, "transaction_date": date(2020, 12, 31), "description": "Something cool", "original_amount": Decimal(100.0), "collaborator_share": 1.0, "chargeable_amount": Decimal(100.0), "transferwise_transaction_id": None, "report_id": None, "voided_transaction_id": None, "currency": "USD", "statement_period_id": 1, } expected = { "id": ANY, "collaborator_id": 1, "type": constants.TYPE_REVENUE, "date": "2020-12-31", "description": "Something cool", "original_amount": 100.0, "collaborator_share": 1.0, "chargeable_amount": 100.0, "transferwise_transaction_id": None, "report_id": None, "created_date": ANY, "voided_transaction_id": None, "credited_payment_id": None, "currency": "USD", "deleted_date": None, "current_balance": 0, "statement_period_id": 1, "creation_batch_uuid": None, } result = TransactionPersister.create_transaction(**payload) result["created_date"] = "now" assert result == expected @db.test_schema_default_seed def test_create_transactions(): """Test creating several transactions.""" created = datetime.now() creation_batch_uuid = str(uuid4()) created_date_str = created.strftime("%Y-%m-%dT%H:%M:%S.%f") date_str = created.date().strftime("%Y-%m-%d") payload = [ { "collaborator_id": 1, "created_date": created, "date": created.date(), "type": constants.TYPE_REVENUE, "description": "Something cool", "original_amount": Decimal(100.0), "collaborator_share": None, "chargeable_amount": Decimal(100.0), "transferwise_transaction_id": None, "report_id": None, "voided_transaction_id": None, "currency": "USD", "statement_period_id": 1, "creation_batch_uuid": creation_batch_uuid, }, { "collaborator_id": 2, "created_date": created, "date": created.date(), "type": constants.TYPE_PAYMENT, "description": "Something cool", "original_amount": Decimal(100.0), "collaborator_share": 1.0, "chargeable_amount": Decimal(100.0), "transferwise_transaction_id": 1, "report_id": None, "voided_transaction_id": None, "currency": "USD", "statement_period_id": 1, "creation_batch_uuid": creation_batch_uuid, }, { "collaborator_id": 3, "created_date": created, "date": created.date(), "type": constants.TYPE_REVENUE, "description": "Something cool", "original_amount": Decimal(100.0), "collaborator_share": 0.5, "chargeable_amount": Decimal(50.0), "transferwise_transaction_id": None, "report_id": 1, "voided_transaction_id": None, "currency": "USD", "statement_period_id": 1, "creation_batch_uuid": creation_batch_uuid, }, { "collaborator_id": 4, "created_date": created, "date": created.date(), "type": constants.TYPE_WHT_ALLOCATION, "description": "Something cool", "original_amount": Decimal(100.0), "collaborator_share": 0.5, "chargeable_amount": Decimal(50.0), "transferwise_transaction_id": None, "report_id": 1, "voided_transaction_id": None, "currency": "USD", "statement_period_id": 1, "creation_batch_uuid": creation_batch_uuid, }, ] result = TransactionPersister.create_transactions( payload, creation_batch_uuid=creation_batch_uuid ) assert result == [ { "id": ANY, "collaborator_id": 1, "type": constants.TYPE_REVENUE, "date": date_str, "description": "Something cool", "original_amount": 100.0, "collaborator_share": None, "chargeable_amount": 100.0, "transferwise_transaction_id": None, "report_id": None, "created_date": created_date_str, "voided_transaction_id": None, "credited_payment_id": None, "currency": "USD", "deleted_date": None, "current_balance": 0, "statement_period_id": 1, "creation_batch_uuid": creation_batch_uuid, }, { "id": ANY, "collaborator_id": 2, "type": constants.TYPE_PAYMENT, "date": date_str, "description": "Something cool", "original_amount": 100.0, "collaborator_share": 1.0, "chargeable_amount": 100.0, "transferwise_transaction_id": 1, "report_id": None, "created_date": created_date_str, "voided_transaction_id": None, "credited_payment_id": None, "currency": "USD", "deleted_date": None, "current_balance": 0, "statement_period_id": 1, "creation_batch_uuid": creation_batch_uuid, }, { "id": ANY, "collaborator_id": 3, "type": constants.TYPE_REVENUE, "date": date_str, "description": "Something cool", "original_amount": 100.0, "collaborator_share": 0.5, "chargeable_amount": 50.0, "transferwise_transaction_id": None, "report_id": 1, "created_date": created_date_str, "voided_transaction_id": None, "credited_payment_id": None, "currency": "USD", "deleted_date": None, "current_balance": 0, "statement_period_id": 1, "creation_batch_uuid": creation_batch_uuid, }, { "id": ANY, "collaborator_id": 4, "type": constants.TYPE_WHT_ALLOCATION, "date": date_str, "description": "Something cool", "original_amount": 100.0, "collaborator_share": 0.5, "chargeable_amount": 50.0, "transferwise_transaction_id": None, "report_id": 1, "created_date": created_date_str, "voided_transaction_id": None, "credited_payment_id": None, "currency": "USD", "deleted_date": None, "current_balance": 0, "statement_period_id": 1, "creation_batch_uuid": creation_batch_uuid, }, ] @db.test_schema_default_seed @pytest.mark.parametrize( "collaborator_id, statement_period_id, limit, offset, expected_result", [ ( 1, 1, 0, 0, ( [ transaction for transaction in transaction_fixtures.SORTED_TRANSACTIONS if transaction["statement_period_id"] == 1 ], 3, ), ) ], ) def test_get_transactions( collaborator_id, statement_period_id, limit, offset, expected_result ): """Test get transactions for a collaborator.""" result = TransactionPersister.get_transactions( collaborator_id=collaborator_id, statement_period_id=statement_period_id, limit=limit, offset=offset, ) assert result == expected_result @db.test_schema_default_seed @pytest.mark.parametrize( "collaborator_id, limit, offset, expected_transactions, expected_count", [ (1, 1, 0, [transaction_fixtures.SORTED_TRANSACTIONS[0]], 4), (1, 2, 1, transaction_fixtures.SORTED_TRANSACTIONS[1:3], 4), ], ) def test_get_transactions_with_limit_and_offset( collaborator_id, limit, offset, expected_transactions, expected_count ): """Test get transactions for a collaborator.""" collaborator_id = 1 statement_period_id = None transactions, count = TransactionPersister.get_transactions( collaborator_id=collaborator_id, statement_period_id=statement_period_id, limit=limit, offset=offset, ) assert len(transactions) == limit assert transactions == expected_transactions assert count == expected_count @db.test_schema_default_seed @pytest.mark.parametrize( "collaborator_id, expected_result", [ ( 1, { "id": 5, "collaborator_id": 1, "date": "2020-02-03", "type": constants.TYPE_REVENUE, "description": "Report revenue", "original_amount": Decimal(1234), "collaborator_share": 1.0, "chargeable_amount": Decimal(1234), "transferwise_transaction_id": None, "report_id": 1, "created_date": "2020-02-03T04:05:06", "voided_transaction_id": None, "credited_payment_id": None, "currency": "USD", "deleted_date": None, "current_balance": 0, "statement_period_id": 2, "creation_batch_uuid": None, }, ), (2, None), ( 3, { "id": 6, "collaborator_id": 3, "date": "2019-10-09", "type": constants.TYPE_REVENUE, "description": "Report revenue", "original_amount": Decimal(9001), "collaborator_share": 1.0, "chargeable_amount": Decimal(9001), "transferwise_transaction_id": None, "report_id": 5, "created_date": "2020-02-03T04:05:06", "voided_transaction_id": None, "credited_payment_id": None, "currency": "USD", "deleted_date": None, "current_balance": 0, "statement_period_id": 2, "creation_batch_uuid": None, }, ), ], ) def test_latest_transaction(collaborator_id, expected_result): """Test getting the latest transaction.""" result = TransactionPersister.latest_transaction(collaborator_id) assert result == expected_result @db.test_schema_default_seed @pytest.mark.parametrize("transaction_id, void_status", [(3, True), (4, False)]) def test_check_void_transaction(transaction_id, void_status): """Test that transaction is voided.""" result = TransactionPersister.check_void_transaction(transaction_id) assert result == void_status @db.test_schema_default_seed @pytest.mark.parametrize( "transaction_id, expected_result", [(1, transaction_fixtures.TRANSACTIONS[0]), (99999, None)], ) def test_get__by_id(transaction_id, expected_result): """Test get transactions by a transaction_id.""" result = TransactionPersister.get_by_id(transaction_id) assert result == expected_result @db.test_schema_default_seed @pytest.mark.parametrize( "transferwise_transaction_id, expected_result", [ ( 1, { "id": 2, "collaborator_id": 1, "date": "2019-10-09", "type": constants.TYPE_PAYMENT, "description": "Payment", "original_amount": -50.0, "collaborator_share": 1.0, "chargeable_amount": -50.0, "transferwise_transaction_id": 1, "report_id": 1, "created_date": datetime(2019, 10, 9, 8, 7, 6).isoformat(), "voided_transaction_id": None, "credited_payment_id": None, "currency": "USD", "deleted_date": None, "current_balance": 0, "statement_period_id": 1, "creation_batch_uuid": None, }, ) ], ) def test_get_transaction_by_transferwise_transaction_id( transferwise_transaction_id, expected_result ): """Test get transaction by a transfwise_transaction_id.""" result = TransactionPersister.get_transaction_by_transferwise_transaction_id( transferwise_transaction_id ) assert result == expected_result @db.test_schema_default_seed @pytest.mark.parametrize("transferwise_transaction_id", [15]) def test_get_transaction_by_transferwise_transaction_id_none( transferwise_transaction_id, ): """Test get transaction by a transfwise_transaction_id.""" result = TransactionPersister.get_transaction_by_transferwise_transaction_id( transferwise_transaction_id ) assert result is None @db.test_schema_default_seed @pytest.mark.parametrize( "payment_id, is_credited", [ (1, False), (2, True), ], ) def test_is_transaction_credited(payment_id, is_credited): """Test that failed payment transaction is credited.""" result = TransactionPersister.is_transaction_credited(payment_id) assert result == is_credited @db.test_schema_default_seed def test_get_transactions_for_statement_period(): """Test get transactions for statement period.""" expected_result = [ { "chargeable_amount": 9001.0, "collaborator_id": 20, "collaborator_share": 1.0, "created_date": "2020-02-03T04:05:06", "currency": "USD", "current_balance": 0, "date": "2018-10-09", "deleted_date": None, "description": "Report revenue", "id": 7, "original_amount": 9001.0, "report_id": 5, "statement_period_id": 1, "transferwise_transaction_id": 2, "type": "CREDIT", "voided_transaction_id": None, "credited_payment_id": None, "creation_batch_uuid": None, }, { "chargeable_amount": -100.0, "collaborator_id": 1, "collaborator_share": 1.0, "created_date": "2019-11-10T09:08:07", "currency": "USD", "current_balance": 0, "date": "2019-11-10", "deleted_date": None, "description": "Expense", "id": 3, "original_amount": -100.0, "report_id": None, "statement_period_id": 1, "transferwise_transaction_id": None, "type": "EXPENSE", "voided_transaction_id": None, "credited_payment_id": None, "creation_batch_uuid": None, }, { "chargeable_amount": -50.0, "collaborator_id": 1, "collaborator_share": 1.0, "created_date": "2019-10-09T08:07:06", "currency": "USD", "current_balance": 0, "date": "2019-10-09", "deleted_date": None, "description": "Payment", "id": 2, "original_amount": -50.0, "report_id": 1, "statement_period_id": 1, "transferwise_transaction_id": 1, "type": "PAYMENT", "voided_transaction_id": None, "credited_payment_id": None, "creation_batch_uuid": None, }, { "chargeable_amount": 100.0, "collaborator_id": 1, "collaborator_share": 1.0, "created_date": "2019-09-08T07:06:05", "currency": "USD", "current_balance": 0, "date": "2019-09-08", "deleted_date": None, "description": "Manual revenue", "id": 1, "original_amount": 100.0, "report_id": None, "statement_period_id": 1, "transferwise_transaction_id": None, "type": "REVENUE", "voided_transaction_id": None, "credited_payment_id": None, "creation_batch_uuid": None, }, ] statement_period_id = 1 ( transactions, total_records, ) = TransactionPersister.get_transactions_for_statement_period( statement_period_id, 0, 0 ) assert transactions == expected_result assert total_records == 4 @db.test_schema_default_seed def test_get_transactions_for_open_period(): """Test get transactions for report ids.""" expected_result = [ { "id": 5, "collaborator_id": 1, "date": "2020-02-03", "type": constants.TYPE_REVENUE, "description": "Report revenue", "original_amount": 1234.0, "collaborator_share": 1.0, "chargeable_amount": 1234.0, "transferwise_transaction_id": None, "report_id": 1, "current_balance": 0, "created_date": "2020-02-03T04:05:06", "voided_transaction_id": None, "credited_payment_id": None, "currency": "USD", "statement_period_id": 2, "deleted_date": None, "creation_batch_uuid": None, } ] txns, count = TransactionPersister.get_transactions_for_open_period(vendor_id=24601) assert txns == expected_result assert count == len(expected_result) @db.test_schema_default_seed def test_get_transactions_with_report_ids(): """Test get transactions for report ids.""" expected_result = [ { "chargeable_amount": -50.0, "collaborator_id": 1, "collaborator_share": 1.0, "created_date": "2019-10-09T08:07:06", "currency": "USD", "current_balance": 0, "date": "2019-10-09", "deleted_date": None, "description": "Payment", "id": 2, "original_amount": -50.0, "report_id": 1, "statement_period_id": 1, "transferwise_transaction_id": 1, "type": "PAYMENT", "voided_transaction_id": None, "credited_payment_id": None, "creation_batch_uuid": None, }, { "chargeable_amount": 1234.0, "collaborator_id": 1, "collaborator_share": 1.0, "created_date": "2020-02-03T04:05:06", "currency": "USD", "current_balance": 0, "date": "2020-02-03", "deleted_date": None, "description": "Report revenue", "id": 5, "original_amount": 1234.0, "report_id": 1, "statement_period_id": 2, "transferwise_transaction_id": None, "type": "REVENUE", "voided_transaction_id": None, "credited_payment_id": None, "creation_batch_uuid": None, }, { "chargeable_amount": 9001.0, "collaborator_id": 3, "collaborator_share": 1.0, "created_date": "2020-02-03T04:05:06", "currency": "USD", "current_balance": 0, "date": "2019-10-09", "deleted_date": None, "description": "Report revenue", "id": 6, "original_amount": 9001.0, "report_id": 5, "statement_period_id": 2, "transferwise_transaction_id": None, "type": "REVENUE", "voided_transaction_id": None, "credited_payment_id": None, "creation_batch_uuid": None, }, { "chargeable_amount": 9001.0, "collaborator_id": 20, "collaborator_share": 1.0, "created_date": "2020-02-03T04:05:06", "currency": "USD", "current_balance": 0, "date": "2018-10-09", "deleted_date": None, "description": "Report revenue", "id": 7, "original_amount": 9001.0, "report_id": 5, "statement_period_id": 1, "transferwise_transaction_id": 2, "type": "CREDIT", "voided_transaction_id": None, "credited_payment_id": None, "creation_batch_uuid": None, }, ] report_ids = [1, 5] transactions = TransactionPersister.get_transactions_by_report_ids(report_ids) assert transactions == expected_result @db.test_schema_default_seed def test_delete_by_ids(): """Test delete transactions by a list of transaction_ids.""" transaction_ids = [5, 6] TransactionPersister.delete_by_ids(transaction_ids) for transaction_id in transaction_ids: result = TransactionPersister.get_by_id(transaction_id, include_deleted=True) assert result["deleted_date"] is not None @db.test_schema_default_seed def test_delete_by_ids_not_found_error(): """Test delete transactions with non-existent transaction_ids.""" transaction_ids = [99999, 88888] with pytest.raises(OwsError) as err: TransactionPersister.delete_by_ids(transaction_ids) assert err.value.code == error.ERROR_CODE_TRANSACTION_NOT_FOUND assert err.value.message == error.ERROR_MESSAGE_TRANSACTION_NOT_FOUND @db.test_schema_default_seed def test_delete_by_ids_closed_period_error(): """Test delete transactions for txns in a closed period.""" transaction_ids = [transaction_fixtures.TRANSACTIONS[0]["id"]] with pytest.raises(OwsError) as err: TransactionPersister.delete_by_ids(transaction_ids) assert err.value.code == error.ERROR_CODE_CANNOT_DELETE_CLOSED_TRANSACTION assert err.value.message == error.ERROR_MESSAGE_CANNOT_DELETE_CLOSED_TRANSACTION @db.test_schema_default_seed @pytest.mark.parametrize( "params, expected_result", [ ( { "report_run_id": 1, "collaborator_dp_enabled": True, }, { "total_count": 1, "non_zero_count": 1, "currency_agnostic_total": Decimal("9001.000000"), "completed_date": datetime(2020, 2, 3, 4, 5, 6), }, ), ( { "report_run_id": 1, "collaborator_dp_enabled": False, }, { "total_count": 3, "non_zero_count": 3, "currency_agnostic_total": Decimal("19236.000000"), "completed_date": datetime(2020, 2, 3, 4, 5, 6), }, ), ], ) def test_get_aggregations_for_report_run(params, expected_result): """Test get_aggregations_for_report_run.""" result = TransactionPersister.get_aggregations_for_report_run(**params) assert result._mapping == expected_result @db.test_schema_default_seed def test_get_transactions_for_participations(): """Test fetching transactions for a list of (collaborator_id, statement_period_id) pairs.""" participations = [(1, 1)] result = TransactionPersister.get_transactions_for_participations(participations) assert set(result.keys()) == {(1, 1)} entry = result[(1, 1)] # Three non-deleted transactions for collaborator_id=1, statement_period_id=1 assert entry["total_count"] == 3 assert len(entry["transactions"]) == 3 # Ordered by created_date desc, date desc assert [t["id"] for t in entry["transactions"]] == [3, 2, 1] # Join with TransferwiseTransaction is present on tx2 tx2 = next(t for t in entry["transactions"] if t["id"] == 2) assert tx2["transferwise_transaction_status"] == "outgoing_payment_sent" assert tx2["transferwise_batch_id"] == 1 # No TW join on tx1 tx1 = next(t for t in entry["transactions"] if t["id"] == 1) assert tx1["transferwise_transaction_status"] is None assert tx1["transferwise_batch_id"] is None @db.test_schema_default_seed def test_get_transactions_for_participations_multiple_pairs(): """Test fetching transactions for multiple participation pairs.""" participations = [(1, 1), (1, 2)] result = TransactionPersister.get_transactions_for_participations(participations) assert set(result.keys()) == {(1, 1), (1, 2)} assert result[(1, 1)]["total_count"] == 3 assert result[(1, 2)]["total_count"] == 1 assert result[(1, 2)]["transactions"][0]["id"] == 5 @db.test_schema_default_seed def test_get_transactions_for_participations_no_match(): """Test fetching transactions for a participation pair with no transactions.""" participations = [(1, 999)] result = TransactionPersister.get_transactions_for_participations(participations) assert result[(1, 999)] == {"transactions": [], "total_count": 0}