pluto/handler/tests/test_agora.py

from unittest import TestCase
from unittest.mock import MagicMock, patch

# from twisted.internet.defer import inlineCallbacks
from json import loads
from copy import deepcopy

from tests.common import fake_public_ads, cg_prices, expected_to_update
import sources.agora
import lib.markets
import lib.money
import util
import settings


class TestAgora(TestCase):
    def __init__(self, *args, **kwargs):
        self.test_return_data = {}
        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]

        super().__init__(*args, *kwargs)

    def setUp(self):
        self.markets = lib.markets.Markets()
        self.agora = sources.agora.Agora()
        self.money = lib.money.Money()
        setattr(self.agora, "markets", self.markets)
        setattr(self.money, "markets", self.markets)
        setattr(self.agora, "money", self.money)
        self.markets.sinks = MagicMock()
        self.markets.sinks.currencies = [
            "AUD",
            "CAD",
            "CHF",
            "CZK",
            "DKK",
            "EUR",
            "GBP",
            "HKD",
            "HUF",
            "JPY",
            "MXN",
            "NOK",
            "NZD",
            "PLN",
            "RUB",
            "SEK",
            "SGD",
            "THB",
            "TRY",
            "USD",
            "ZAR",
        ]
        self.agora.sinks = MagicMock()
        self.agora.sinks.currencies = self.markets.sinks.currencies

        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.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.agora.money.cg.get_price = MagicMock()
        self.agora.money.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 the providers
        settings.settings.Agora.ProviderList = '["REVOLUT", "NATIONAL_BANK"]'

        # Override enum_public_ads
        self.agora.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.agora.money.cg.get_price = MagicMock()
        self.agora.money.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.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 = 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.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.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.money.lookup_rates("agora", 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.agora.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.agora.money.cg.get_price = MagicMock()
        self.agora.money.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.money.lookup_rates("agora", enum_ads_return, rates=cg_prices)
        self.assertCountEqual(lookup_rates_return, expected_return)