pluto/handler/tests/test_lbtc.py

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_lbtc, cg_prices, expected_to_update_lbtc
from sources.localbitcoins import LBTC
from markets import Markets
from money import Money
import util
import settings
import sources


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()
        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.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()
        print("public_ads", public_ads)
        # print("fake_public_ads", fake_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)

    def test_update_prices(self):
        settings.settings.LocalBitcoins.Username = "Harrey"
        # Override the providers
        settings.settings.LocalBitcoins.ProviderList = '["national-bank-transfer"]'

        # 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()
        call_args = self.lbtc.slow_ad_update.call_args_list[0][0][0]
        self.assertCountEqual(call_args, expected_to_update_lbtc)

    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"]["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.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)