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 import sources.localbitcoins import util from tests.common import (cg_prices, expected_to_update_lbtc, fake_public_ads_lbtc) from twisted.internet.defer import inlineCallbacks 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): logging.disable(logging.CRITICAL) self.markets = lib.markets.Markets() self.lbtc = sources.localbitcoins.LBTC() self.money = lib.money.Money() self.sources = sources.Sources() setattr(self.markets, "sources", self.sources) 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.es = MagicMock() self.lbtc.es.index = 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.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.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 = yield self.lbtc.get_all_public_ads() self.assertDictEqual(public_ads, fake_public_ads_lbtc) 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) @inlineCallbacks def test_update_prices(self): settings.settings.LocalBitcoins.MinMargin = 1.10 settings.settings.LocalBitcoins.MaxMargin = 1.3 settings.settings.LocalBitcoins.Username = "Harrey" # Override the providers settings.settings.LocalBitcoins.ProviderList = '["national-bank-transfer"]' # Override enum_public_ads self.lbtc.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.lbtc.money.cg.get_price = MagicMock() self.lbtc.money.cg.get_price.return_value = cg_prices self.lbtc.slow_ad_update = MagicMock() yield self.lbtc.update_prices() call_args = self.lbtc.slow_ad_update.call_args_list[0][0][0] self.assertCountEqual(call_args, expected_to_update_lbtc) @inlineCallbacks def test_enum_public_ads(self): # Override enum_public_ads self.lbtc.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.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.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.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) 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.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.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)