import logging from copy import deepcopy from unittest import TestCase from unittest.mock import MagicMock import lib.antifraud import lib.money import lib.transactions class TestTransactions(TestCase): def setUp(self): logging.disable(logging.CRITICAL) self.transactions = lib.transactions.Transactions() self.test_data = { "timestamp": "2022-03-14T19:34:13.501Z", "description": "Received Rebiere Matthieu", "transaction_type": "CREDIT", "transaction_category": "CREDIT", "transaction_classification": [], "amount": 1, "currency": "GBP", "transaction_id": "ec4df5248c750c30301a1da71024ac0b", "provider_transaction_id": "27373011.TU9ORVRBUllfQUNUSVZJVFk6OjI1MDE4MDQyOjpUUkFOU0ZFUjo6MzgwMDM2NDY2", "normalised_provider_transaction_id": "txn-c8c12c308789bd980", "meta": { "provider_reference": "TEST-1", "transaction_type": "Credit", "provider_id": "27373011.TU9ORVRBUllfQUNUSVZJVFk6OjI1MDE4MDQyOjpUUkFOU0ZFUjo6MzgwMDM2NDY2", "counter_party_preferred_name": "Rebiere Matthieu", }, "subclass": "truelayer", } # Mock redis calls lib.transactions.db.r.hgetall = self.mock_hgetall lib.transactions.db.r.hmset = self.mock_hmset lib.transactions.db.r.keys = self.mock_keys lib.transactions.db.r.get = self.mock_get # Mock some callbacks self.transactions.irc = MagicMock() self.transactions.irc.sendmsg = MagicMock() self.transactions.release_funds = MagicMock() self.transactions.ux = MagicMock() self.transactions.ux.notify = MagicMock() self.transactions.ux.notify.notify_complete_trade = MagicMock() self.transactions.antifraud = lib.antifraud.AntiFraud() # Mock the rates self.transactions.money = MagicMock() self.transactions.money.to_usd = self.mock_to_usd self.transactions.money.get_rates_all = MagicMock() self.transactions.money.get_rates_all.return_value = {"GBP": 0.8} # Don't mock the functions we want to test self.money = lib.money.Money() self.money.get_rates_all = MagicMock() self.money.get_rates_all.return_value = {"GBP": 0.8} self.transactions.money.get_acceptable_margins = ( self.money.get_acceptable_margins ) self.trades = { 1: { "id": "uuid1", "buyer": "test_buyer_1", "currency": "GBP", "amount": "1", "amount_xmr": "0.3", "reference": "TEST-1", "subclass": "agora", }, 2: { "id": "uuid2", "buyer": "test_buyer_2", "currency": "GBP", "amount": "1", "amount_xmr": "0.3", "reference": "TEST-2", "subclass": "agora", }, 3: { "id": "uuid3", "buyer": "test_buyer_3", "currency": "GBP", "amount": "1000", "amount_xmr": "3", "reference": "TEST-3", "subclass": "agora", }, 4: { "id": "uuid4", "buyer": "test_buyer_4", "currency": "GBP", "amount": "10", "amount_xmr": "0.5", "reference": "TEST-4", "subclass": "agora", }, } self.return_trades = [1, 2, 3] @property def test_data_copy(self): return deepcopy(self.test_data) def data_custom(self, amount, currency, reference): test_data = self.test_data_copy test_data["meta"]["provider_reference"] = reference test_data["amount"] = amount test_data["currency"] = currency return test_data def mock_hgetall(self, string): ref = string.split(".")[1] for num, trade in self.trades.items(): if trade["reference"] == ref: return trade def mock_hmset(self, string, data): pass def mock_keys(self, string): return [v["id"] for k, v in self.trades.items() if k in self.return_trades] def mock_get(self, string): for num, trade in self.trades.items(): if trade["id"] == string: return trade["reference"] def mock_to_usd(self, amount, currency): if currency == "GBP": return amount * 1.3 elif currency == "USD": return amount # fuck it who cares elif currency == "SEK": return 100 elif currency == "EUR": return 10 def test_transaction(self): self.transactions.transaction(self.test_data) self.transactions.release_funds.assert_called_once_with("uuid1", "TEST-1") self.transactions.release_funds = MagicMock() ref_2 = self.test_data_copy ref_2["meta"]["provider_reference"] = "TEST-2" self.transactions.transaction(ref_2) self.transactions.release_funds.assert_called_once_with("uuid2", "TEST-2") def test_transaction_invalid(self): invalid_data = self.test_data_copy invalid_data = self.data_custom(2000, "SEK", "sss") self.transactions.transaction(invalid_data) self.transactions.release_funds.assert_not_called() def test_transaction_malformed(self): malformed_data = self.test_data_copy del malformed_data["amount"] self.transactions.transaction(malformed_data) self.transactions.release_funds.assert_not_called() malformed_data = self.test_data_copy del malformed_data["currency"] self.transactions.transaction(malformed_data) self.transactions.release_funds.assert_not_called() def test_transaction_no_reference_fail(self): no_reference_fail = self.data_custom(1, "GBP", "none") no_reference_fail["meta"]["provider_reference"] = "none" self.transactions.transaction(no_reference_fail) self.transactions.release_funds.assert_not_called() def test_transaction_no_reference_pass(self): no_reference_pass = self.data_custom(1, "GBP", "none") no_reference_pass["meta"]["provider_reference"] = "THIS_ONE_FAILS" self.return_trades = [1] self.transactions.transaction(no_reference_pass) self.transactions.release_funds.assert_called_with("uuid1", "TEST-1") def test_transaction_large(self): exceeds_max = self.data_custom(1000, "GBP", "TEST-3") self.transactions.transaction(exceeds_max) self.transactions.release_funds.assert_called_once_with("uuid3", "TEST-3") def test_transaction_no_reference_exceeds_max(self): exceeds_max = self.data_custom(1000, "GBP", "noref") self.transactions.transaction(exceeds_max) self.transactions.release_funds.assert_not_called() def test_transaction_wrong_currency(self): wrong_currency = self.data_custom(1, "EUR", "TEST-1") self.transactions.transaction(wrong_currency) self.transactions.release_funds.assert_not_called() wrong_currency = self.data_custom(1, "EUR", "none") self.transactions.transaction(wrong_currency) self.transactions.release_funds.assert_not_called() def test_transaction_wrong_amount(self): self.transactions.money.get_acceptable_margins = MagicMock() self.transactions.money.get_acceptable_margins.return_value = (0.8, 1.8) wrong_amount = self.data_custom(10, "GBP", "TEST-1") self.transactions.transaction(wrong_amount) self.transactions.release_funds.assert_not_called() wrong_amount = self.data_custom(10, "GBP", "none") self.transactions.transaction(wrong_amount) self.transactions.release_funds.assert_not_called() # def test_transaction_pending(self): # pending_tx = self.test_data_copy # pending_tx["data"]["state"] = "pending" # self.transactions.transaction(pending_tx) # self.transactions.release_funds.assert_not_called() def test_transaction_too_low(self): too_low = self.data_custom(5, "GBP", "TEST-1") self.transactions.transaction(too_low) self.transactions.release_funds.assert_not_called() def test_transaction_too_high(self): too_high = self.data_custom(15, "GBP", "TEST-1") self.transactions.transaction(too_high) self.transactions.release_funds.assert_not_called() # def test_transaction_pending_then_completed(self): # pass def test_transaction_store_incomplete_trade(self): pass def test_transaction_release_incomplete_trade(self): pass def test_transaction_card_payment(self): pass def test_transaction_negative_amount(self): pass def test_release_funds(self): pass def test_new_trade(self): pass def test_find_trade(self): pass def test_get_refs(self): pass def test_get_ref_map(self): pass def test_get_ref(self): pass def test_del_ref(self): pass def test_cleanup(self): pass def test_tx_to_ref(self): pass def test_ref_to_tx(self): pass def test_get_total_usd(self): pass def test_get_total(self): pass def test_write_to_es(self): pass def test_get_remaining(self): pass def test_get_open_trades_usd(self): pass def test_get_total_remaining(self): pass def get_total_with_trades(self): pass