from unittest import TestCase from unittest.mock import MagicMock from copy import deepcopy import transactions class TestTransactions(TestCase): def setUp(self): self.transactions = transactions.Transactions() self.test_data = { "event": "TransactionCreated", "timestamp": "2022-02-24T20:26:15.232342Z", "data": { "id": "6217e9e7-43e1-a809-8500-0a5b0170e6e4", "type": "transfer", "state": "completed", "request_id": "8a15213e-a7d2-4738-bfb5-b1d037b75a57", "created_at": "2022-02-24T20:26:15.238218Z", "updated_at": "2022-02-24T20:26:15.238218Z", "completed_at": "2022-02-24T20:26:15.238453Z", "reference": "TEST-1", "legs": [ { "leg_id": "80b35daf-409c-41be-8755-15982b7633a6", "account_id": "7185593b-d9ad-4456-920e-d9db109a5172", "counterparty": {"account_type": "revolut"}, "amount": 1, "currency": "GBP", "description": "From Mark Veidemanis", "balance": 3832.3, } ], }, } # Mock redis calls transactions.r.hgetall = self.mock_hgetall transactions.r.hmset = self.mock_hmset transactions.r.keys = self.mock_keys transactions.r.get = self.mock_get # Mock some callbacks self.transactions.irc = MagicMock() self.transactions.irc.sendmsg = MagicMock() self.transactions.release_funds = MagicMock() self.transactions.notify = MagicMock() self.transactions.notify.notify_complete_trade = MagicMock() # Mock the rates self.transactions.money = MagicMock() self.transactions.money.get_rates_all = MagicMock() self.transactions.money.get_rates_all.return_value = {"GBP": 0.8} self.trades = { 1: { "id": "uuid1", "buyer": "test_buyer_1", "currency": "GBP", "amount": "1", "amount_xmr": "0.3", "reference": "TEST-1", }, 2: { "id": "uuid2", "buyer": "test_buyer_2", "currency": "GBP", "amount": "1", "amount_xmr": "0.3", "reference": "TEST-2", }, 3: { "id": "uuid3", "buyer": "test_buyer_3", "currency": "GBP", "amount": "1000", "amount_xmr": "3", "reference": "TEST-3", }, 4: { "id": "uuid4", "buyer": "test_buyer_4", "currency": "GBP", "amount": "10", "amount_xmr": "0.5", "reference": "TEST-4", }, } 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["data"]["reference"] = reference test_data["data"]["legs"][0]["amount"] = amount test_data["data"]["legs"][0]["currency"] = currency return test_data def mock_hgetall(self, string): print("MOCK HGERALL CALLED WITH", string) ref = string.split(".")[1] for num, trade in self.trades.items(): if trade["reference"] == ref: return trade def mock_hmset(self, string, data): print("HMSET", string, data) 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): print("MOCK GET", string) for num, trade in self.trades.items(): if trade["id"] == string: return trade["reference"] 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["data"]["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["data"] self.transactions.transaction(malformed_data) self.transactions.release_funds.assert_not_called() malformed_data = self.test_data_copy del malformed_data["data"]["type"] 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["data"]["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["data"]["reference"] = "none" 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