pluto/handler/tests/test_agora.py

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

from tests.common import fake_public_ads, cg_prices, expected_to_update
from agora import Agora
from markets import Markets


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
        self.agora.last_online_recent = MagicMock()
        self.agora.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()
        # print("PUBADS", 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):
        return  # TODO: FIX
        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):
        return  # TODO: FIX
        self.agora.cg.get_price = MagicMock()
        self.agora.cg.get_price.return_value = cg_prices
        self.agora.wrap_public_ads = self.mock_wrap_public_ads

        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
        self.agora.last_online_recent = MagicMock()
        self.agora.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)
                    # print("expecting", 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):
        # print("lookup rates ran")
        pass