224 lines
7.9 KiB
Python
224 lines
7.9 KiB
Python
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
|
|
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.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
|
|
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
|
|
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
|
|
|
|
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
|
|
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
|
|
|
|
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)
|