diff --git a/handler/tests/test_agora.py b/handler/tests/test_agora.py index 48924bc..aadc0eb 100644 --- a/handler/tests/test_agora.py +++ b/handler/tests/test_agora.py @@ -14,7 +14,7 @@ import settings class TestAgora(TestCase): def __init__(self, *args, **kwargs): self.test_return_data = {} - with open("tests/data/api_call_ads_return.json", "r") as f: + 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] @@ -52,6 +52,8 @@ class TestAgora(TestCase): "USD", "ZAR", ] + self.agora.sinks = MagicMock() + self.agora.sinks.currencies = self.markets.sinks.currencies self.all_providers = [ "XOOM", diff --git a/handler/tests/test_lbtc.py b/handler/tests/test_lbtc.py new file mode 100644 index 0000000..3b413c9 --- /dev/null +++ b/handler/tests/test_lbtc.py @@ -0,0 +1,208 @@ +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 sources.localbitcoins import LBTC +from markets import Markets +from money import Money +import util +import settings + + +class TestLBTC(TestCase): + def __init__(self, *args, **kwargs): + self.test_return_data = {} + with open("tests/data/lbtc_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): + self.markets = Markets() + self.lbtc = LBTC() + self.money = Money() + setattr(self.lbtc, "markets", self.markets) + setattr(self.money, "markets", self.markets) + setattr(self.lbtc, "money", self.money) + self.markets.sinks = MagicMock() + self.markets.sinks.currencies = [ + "GBP", + ] + self.lbtc.sinks = MagicMock() + self.lbtc.sinks.currencies = self.markets.sinks.currencies + + self.all_providers = [ + "national-bank-transfer", + ] + + def mock_enum_public_ads_api_call(self, api_method, query_values): + if "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.lbtc.lbtc._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.lbtc.money.cg.get_price = MagicMock() + self.lbtc.money.cg.get_price.return_value = cg_prices + + self.lbtc.markets.get_all_providers = MagicMock() + self.lbtc.markets.get_all_providers.return_value = self.all_providers + + public_ads = self.lbtc.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.lbtc.run_cheat_in_thread() + + asset2 = self.lbtc.run_cheat_in_thread() + self.assertEqual(set([asset1, asset2]), set(["BTC"])) + + asset3 = self.lbtc.run_cheat_in_thread() + + asset4 = self.lbtc.run_cheat_in_thread() + + self.assertEqual(set([asset3, asset4]), set(["BTC"])) + + # Only one asset so far for BTC + # self.assertNotEqual(asset1, asset2) + # self.assertNotEqual(asset3, asset4) + + def test_update_prices(self): + settings.settings.LocalBitcoins.Username = "Harrey" + # Override the providers + settings.settings.LocalBitcoins.ProviderList = '["REVOLUT", "NATIONAL_BANK"]' + + # Override enum_public_ads + self.lbtc.lbtc._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.lbtc.money.cg.get_price = MagicMock() + self.lbtc.money.cg.get_price.return_value = cg_prices + + self.lbtc.slow_ad_update = MagicMock() + self.lbtc.update_prices() + print("CALL ARGS", self.lbtc.slow_ad_update.call_args_list) + call_args = self.lbtc.slow_ad_update.call_args_list[0][0][0] + print("EXPECTED", expected_to_update) + self.assertCountEqual(call_args, expected_to_update) + + def test_enum_public_ads(self): + # Override enum_public_ads + self.lbtc.lbtc._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.lbtc.enum_public_ads("BTC", "GBP", 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 == "BTC": + continue + if not currency == "GBP": + continue + content = self.test_return_data[(asset, currency, page)] + ads = content["response"]["data"]["ad_list"] + for ad in ads: + ad_id = str(ad["data"]["ad_id"]) + username = ad["data"]["profile"]["username"] + temp_price = ad["data"]["temp_price"] + provider = ad["data"]["online_provider"] + asset = "BTC" + 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.lbtc.lbtc._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.lbtc.enum_public_ads("BTC", "GBP", self.all_providers) + + expected_return = [] + # Let's manually calculate what it's supposed to look like + price_btc = cg_prices["bitcoin"]["usd"] + for ad in deepcopy(enum_ads_return): + price = float(ad[2]) + margin = round(price / price_btc, 2) + ad.append(margin) + expected_return.append(ad) + + lookup_rates_return = self.lbtc.money.lookup_rates("lbtc", 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.lbtc.lbtc._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.lbtc.money.cg.get_price = MagicMock() + self.lbtc.money.cg.get_price.return_value = cg_prices + + enum_ads_return = self.lbtc.enum_public_ads("BTC", "GBP", self.all_providers) + + expected_return = [] + # Let's manually calculate what it's supposed to look like + price_btc = cg_prices["bitcoin"]["gbp"] + for ad in deepcopy(enum_ads_return): + price = float(ad[2]) + margin = round(price / price_btc, 2) + ad.append(margin) + expected_return.append(ad) + # Test specifying rates= + lookup_rates_return = self.lbtc.money.lookup_rates("lbtc", enum_ads_return, rates=cg_prices) + self.assertCountEqual(lookup_rates_return, expected_return)