pluto/handler/tests/test_agora.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, cg_prices, expected_to_update
from agora import Agora
from markets import Markets
from money import Money
import util
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:
            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()
        self.money = Money()
        setattr(self.agora, "markets", self.markets)
        setattr(self.money, "markets", self.markets)
        setattr(self.agora, "money", self.money)

        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
        util.last_online_recent = MagicMock()
        util.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()
        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.agora._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.cg.get_price = MagicMock()
        self.agora.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.agora._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.agora._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(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.agora._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.cg.get_price = MagicMock()
        self.agora.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(enum_ads_return, rates=cg_prices)
        self.assertCountEqual(lookup_rates_return, expected_return)

    def test_format_ad(self):
        settings.settings.Agora.Ad = """* Set **Country of recipient's bank** to **"United Kingdom"**
$PAYMENT$
* Set **Company name** to **"PATHOGEN LIMITED"**"""
        payment_details = {"sort_code": "02-03-04", "account_number": "0023-0045"}
        payment_details_text = self.agora.format_payment_details("GBP", payment_details)
        ad_text = self.agora.format_ad("XMR", "GBP", payment_details_text)
        expected = """* Set **Country of recipient's bank** to **"United Kingdom"**
* Company name: **PATHOGEN LIMITED**
* Sort code: **02-03-04**
* Account number: **0023-0045**
* Please send in **GBP**
* If you are asked for address information, please use **24 Holborn Viaduct, London, England, EC1A 2BN**
* The post code is **EC1A 2BN**
* Set **Company name** to **"PATHOGEN LIMITED"**"""
        self.assertEqual(ad_text, expected)

    def test_format_payment_details(self):
        payment_details = {"sort_code": "02-03-04", "account_number": "0023-0045"}
        payment_details_text = self.agora.format_payment_details("GBP", payment_details)

        expected = """* Company name: **PATHOGEN LIMITED**
* Sort code: **02-03-04**
* Account number: **0023-0045**
* Please send in **GBP**
* If you are asked for address information, please use **24 Holborn Viaduct, London, England, EC1A 2BN**
* The post code is **EC1A 2BN**"""
        self.assertEqual(payment_details_text, expected)