from unittest import TestCase from unittest.mock import MagicMock, patch from json import loads from copy import deepcopy from tests.common import fake_public_ads, cg_prices, expected_to_update from agora import Agora from markets import Markets import util class TestAgora(TestCase): def __init__(self, *args, **kwargs): self.test_return_data = {} with open("tests/data/api_call_ads_return.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): self.markets = Markets() self.agora = Agora() setattr(self.agora, "markets", self.markets) 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)] def test_get_all_public_ads(self): # Override enum_public_ads self.agora.agora._api_call = self.mock_enum_public_ads_api_call self.agora.last_online_recent = MagicMock() self.agora.last_online_recent.return_value = True # Override get_price self.agora.cg.get_price = MagicMock() self.agora.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 = 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) def test_update_prices(self): # Override enum_public_ads self.agora.agora._api_call = self.mock_enum_public_ads_api_call self.agora.last_online_recent = MagicMock() self.agora.last_online_recent.return_value = True util.last_online_recent = MagicMock() util.last_online_recent.return_value = True # Override get_price self.agora.cg.get_price = MagicMock() self.agora.cg.get_price.return_value = cg_prices self.agora.slow_ad_update = MagicMock() self.agora.update_prices() call_args = self.agora.slow_ad_update.call_args_list[0][0][0] self.assertCountEqual(call_args, expected_to_update) def test_enum_public_ads(self): # Override enum_public_ads self.agora.agora._api_call = self.mock_enum_public_ads_api_call util.last_online_recent = MagicMock() util.last_online_recent.return_value = True enum_ads_return = 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.agora._api_call = self.mock_enum_public_ads_api_call self.agora.last_online_recent = MagicMock() self.agora.last_online_recent.return_value = True # Override get_price self.agora.cg.get_price = MagicMock() self.agora.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.lookup_rates(enum_ads_return) 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.agora._api_call = self.mock_enum_public_ads_api_call self.agora.last_online_recent = MagicMock() self.agora.last_online_recent.return_value = True # Override get_price self.agora.cg.get_price = MagicMock() self.agora.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.lookup_rates(enum_ads_return, rates=cg_prices) self.assertCountEqual(lookup_rates_return, expected_return)