import logging from copy import deepcopy from json import loads from unittest import TestCase from unittest.mock import MagicMock, patch import lib.markets import lib.money import settings import sources.agora import util from tests.common import cg_prices, expected_to_update, fake_public_ads from twisted.internet.defer import inlineCallbacks class TestAgora(TestCase): def __init__(self, *args, **kwargs): self.test_return_data = {} with open("tests/data/agora_ads.json", "r") as f: for line in f.readlines(): parsed = loads(line) self.test_return_data[(parsed[2], parsed[1], str(parsed[0]))] = parsed[ 3 ] super().__init__(*args, *kwargs) def setUp(self): logging.disable(logging.CRITICAL) self.markets = lib.markets.Markets() self.agora = sources.agora.Agora() self.money = lib.money.Money() setattr(self.agora, "markets", self.markets) setattr(self.money, "markets", self.markets) setattr(self.agora, "money", self.money) self.markets.sinks = MagicMock() self.markets.sinks.currencies = [ "AUD", "CAD", "CHF", "CZK", "DKK", "EUR", "GBP", "HKD", "HUF", "JPY", "MXN", "NOK", "NZD", "PLN", "RUB", "SEK", "SGD", "THB", "TRY", "USD", "ZAR", ] self.agora.sinks = MagicMock() self.agora.es = MagicMock() self.agora.es.index = MagicMock() self.agora.sinks.currencies = self.markets.sinks.currencies self.all_providers = [ "XOOM", "CRYPTOCURRENCY", "VIPPS", "PAYSAFECARD", "PAYPAL", "WU", "SQUARE_CASH", "CASH_DEPOSIT", "ADVCASH", "TRANSFERWISE", "GIFT_CARD_CODE_GLOBAL", "NETELLER", "INTERNATIONAL_WIRE_SWIFT", "CASH_BY_MAIL", "SEPA", "OTHER", "REVOLUT", "NATIONAL_BANK", "MONEYBOOKERS", "CREDITCARD", "APPLE_PAY", "ZELLE", "PERFECT_MONEY", "CASHIERS_CHECK", "GOOGLEWALLET", "STRIKE", "SPECIFIC_BANK", "CHIPPER_CASH", "REMITLY", "WORLDREMIT", "PAYEER", "MOBILE_TOP_UP", "VIRTUAL_VISA_MASTERCARD", "VANILLA", "MONEYGRAM", "VENMO", "SERVE2SERVE", "WEBMONEY", ] def mock_enum_public_ads_api_call(self, api_method, query_values): if "buy-monero-online" in api_method: asset = "XMR" elif "buy-bitcoins-online" in api_method: asset = "BTC" spl = api_method.split("/") currency = spl[1] page = str(query_values["page"]) return self.test_return_data[(asset, currency, page)] @inlineCallbacks def test_get_all_public_ads(self): # Override enum_public_ads self.agora.api._api_call = self.mock_enum_public_ads_api_call util.last_online_recent = MagicMock() util.last_online_recent.return_value = True # Override get_price self.agora.money.cg.get_price = MagicMock() self.agora.money.cg.get_price.return_value = cg_prices self.agora.markets.get_all_providers = MagicMock() self.agora.markets.get_all_providers.return_value = self.all_providers public_ads = yield self.agora.get_all_public_ads() self.assertDictEqual(public_ads, fake_public_ads) for currency, ads in public_ads.items(): ad_ids = [ad[0] for ad in ads] ad_ids_dedup = set(ad_ids) # Make sure there's no duplicate ads self.assertEqual(len(ad_ids), len(ad_ids_dedup)) @patch("twisted.internet.threads.deferToThread") def test_run_cheat_in_thread(self, defer): asset1 = self.agora.run_cheat_in_thread() asset2 = self.agora.run_cheat_in_thread() self.assertEqual(set([asset1, asset2]), set(["XMR", "BTC"])) asset3 = self.agora.run_cheat_in_thread() asset4 = self.agora.run_cheat_in_thread() self.assertEqual(set([asset3, asset4]), set(["XMR", "BTC"])) self.assertNotEqual(asset1, asset2) self.assertNotEqual(asset3, asset4) @inlineCallbacks def test_update_prices(self): # Override the providers settings.settings.Agora.MinMargin = 1.17 settings.settings.Agora.MaxMargin = 1.3 settings.settings.Agora.ProviderList = '["REVOLUT", "NATIONAL_BANK"]' # Override enum_public_ads self.agora.api._api_call = self.mock_enum_public_ads_api_call util.last_online_recent = MagicMock() util.last_online_recent.return_value = True # Override get_price self.agora.money.cg.get_price = MagicMock() self.agora.money.cg.get_price.return_value = cg_prices self.agora.slow_ad_update = MagicMock() yield self.agora.update_prices() call_args = self.agora.slow_ad_update.call_args_list[0][0][0] self.assertCountEqual(call_args, expected_to_update) @inlineCallbacks def test_enum_public_ads(self): # Override enum_public_ads self.agora.api._api_call = self.mock_enum_public_ads_api_call util.last_online_recent = MagicMock() util.last_online_recent.return_value = True enum_ads_return = yield self.agora.enum_public_ads( "XMR", "USD", self.all_providers ) # Ensure there are no duplicates enum_ads_return_ids = [ (x[0], x[1], x[2], x[3], x[4], x[5]) for x in enum_ads_return ] enum_ads_return_ids_dedup = set(enum_ads_return_ids) self.assertEqual(len(enum_ads_return_ids), len(enum_ads_return_ids_dedup)) expected_return = [] # ['94b399e2-2c96-480c-b399-e22c96180cf2', 'Jorge', '272.00', 'SEPA', 'XMR', 'USD'] for asset, currency, page in self.test_return_data: if not asset == "XMR": continue if not currency == "USD": continue content = self.test_return_data[(asset, currency, page)] ads = content["response"]["data"]["ad_list"] for ad in ads: ad_id = ad["data"]["ad_id"] username = ad["data"]["profile"]["username"] temp_price = ad["data"]["temp_price"] provider = ad["data"]["online_provider"] asset = "XMR" currency = ad["data"]["currency"] to_append = [ ad_id, username, temp_price, provider, asset, currency, ] if to_append not in expected_return: expected_return.append(to_append) self.assertCountEqual(enum_ads_return, expected_return) self.assertNotEqual(enum_ads_return[0][0], enum_ads_return[1][0]) ad_ids = [x[0] for x in enum_ads_return] ad_ids_dedup = set(ad_ids) self.assertEqual(len(ad_ids), len(ad_ids_dedup)) def test_lookup_rates(self): # Override enum_public_ads self.agora.api._api_call = self.mock_enum_public_ads_api_call util.last_online_recent = MagicMock() util.last_online_recent.return_value = True # Override get_price self.money.cg.get_price = MagicMock() self.money.cg.get_price.return_value = cg_prices enum_ads_return = self.agora.enum_public_ads("XMR", "USD", self.all_providers) expected_return = [] # Let's manually calculate what it's supposed to look like price_xmr = cg_prices["monero"]["usd"] for ad in deepcopy(enum_ads_return): price = float(ad[2]) margin = round(price / price_xmr, 2) ad.append(margin) expected_return.append(ad) lookup_rates_return = self.agora.money.lookup_rates( "agora", enum_ads_return ) # TODO: do this properly self.assertCountEqual(lookup_rates_return, expected_return) def test_lookup_rates_not_usd(self): """ Above test only tests USD which does not take into account Forex. Let's test both, and additionaly specify our own rates. """ # Override enum_public_ads self.agora.api._api_call = self.mock_enum_public_ads_api_call util.last_online_recent = MagicMock() util.last_online_recent.return_value = True # Override get_price self.agora.money.cg.get_price = MagicMock() self.agora.money.cg.get_price.return_value = cg_prices enum_ads_return = self.agora.enum_public_ads("XMR", "EUR", self.all_providers) expected_return = [] # Let's manually calculate what it's supposed to look like price_xmr = cg_prices["monero"]["eur"] for ad in deepcopy(enum_ads_return): price = float(ad[2]) margin = round(price / price_xmr, 2) ad.append(margin) expected_return.append(ad) # Test specifying rates= lookup_rates_return = self.agora.money.lookup_rates( "agora", enum_ads_return, rates=cg_prices ) self.assertCountEqual(lookup_rates_return, expected_return)