pluto/handler/tests/test_agora.py

68 lines
2.3 KiB
Python

from unittest import TestCase
from unittest.mock import MagicMock, patch
from tests.common import fake_public_ads, cg_prices, expected_to_update
from agora import Agora
from markets import Markets
class TestAgora(TestCase):
def setUp(self):
self.markets = Markets()
self.agora = Agora()
setattr(self.agora, "markets", self.markets)
def mock_wrap_public_ads(self, asset, currency, providers, rates):
try:
return fake_public_ads[currency]
except KeyError:
return
def test_get_all_public_ads(self):
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
public_ads = self.agora.get_all_public_ads()
self.assertDictEqual(public_ads, fake_public_ads)
def test_get_all_public_ads_only_one(self):
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
public_ads = self.agora.get_all_public_ads()
for currency, ads in public_ads.items():
ad_ids = [ad[0] for ad in ads]
len_ad_ids = len(ad_ids)
ad_ids_dedup = set(ad_ids)
len_ad_ids_dedup = len(ad_ids_dedup)
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):
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)