import unittest
+from tests.acceptance import TimeMock
-from tests.test_ccxt_wrapper import *
-from tests.test_main import *
-from tests.test_market import *
-from tests.test_store import *
-from tests.test_portfolio import *
+from tests.helper import limits
+
+if "unit" in limits:
+ from tests.test_ccxt_wrapper import *
+ from tests.test_main import *
+ from tests.test_market import *
+ from tests.test_store import *
+ from tests.test_portfolio import *
+
+if "acceptance" in limits:
+ from tests.test_acceptance import *
if __name__ == '__main__':
unittest.main()
import requests_mock
import sys, os
import time, datetime
-import unittest
from unittest import mock
from ssl import SSLError
from decimal import Decimal
from io import StringIO
import re
import functools
-import glob
+import threading
+class TimeMock:
+ delta = {}
+ delta_init = 0
+ true_time = time.time
+ true_sleep = time.sleep
+ time_patch = None
+ datetime_patch = None
+
+ @classmethod
+ def travel(cls, start_date):
+ cls.delta = {}
+ cls.delta_init = (datetime.datetime.now() - start_date).total_seconds()
+
+ @classmethod
+ def start(cls):
+ cls.delta = {}
+ cls.delta_init = 0
+
+ class fake_datetime(datetime.datetime):
+ @classmethod
+ def now(cls, tz=None):
+ if tz is None:
+ return cls.fromtimestamp(time.time())
+ else:
+ return tz.fromutc(cls.utcfromtimestamp(time.time()).replace(tzinfo=tz))
+
+ cls.time_patch = mock.patch.multiple(time, time=cls.fake_time, sleep=cls.fake_sleep)
+ cls.datetime_patch = mock.patch.multiple(datetime, datetime=fake_datetime)
+ cls.time_patch.start()
+ cls.datetime_patch.start()
+
+ @classmethod
+ def stop(cls):
+ cls.delta = {}
+ cls.delta_init = 0
+
+ @classmethod
+ def fake_time(cls):
+ cls.delta.setdefault(threading.current_thread(), cls.delta_init)
+ return cls.true_time() - cls.delta[threading.current_thread()]
+
+ @classmethod
+ def fake_sleep(cls, duration):
+ cls.delta.setdefault(threading.current_thread(), cls.delta_init)
+ cls.delta[threading.current_thread()] -= float(duration)
+ cls.true_sleep(min(float(duration), 0.1))
+
+TimeMock.start()
import main
class FileMock:
if not line.startswith("[Worker] "):
self.tester.fail("« {} » not found in log file {}".format(line, split_logs))
# Le fichier de log est écrit
- # Le fichier de log est printed uniquement si non quiet
# Le rapport est écrit si pertinent
# Le rapport contient le bon nombre de lignes
def stop(self):
self.mocker.stop()
+ lazy_calls = [
+ "https://cryptoportfolio.io/wp-content/uploads/portfolio/json/cryptoportfolio.json",
+ "https://poloniex.com/public?command=returnTicker",
+ ]
def check_calls(self):
self.tester.assertEqual([], self.error_calls)
for (method, url), elements in self.mocks.items():
for market_id, element in elements.items():
- self.tester.assertEqual(0, len(element), "Missing calls to {} {}, market_id {}".format(method, url, market_id))
+ if url not in self.lazy_calls:
+ self.tester.assertEqual(0, len(element), "Missing calls to {} {}, market_id {}".format(method, url, market_id))
def clean_body(self, body):
if body is None:
elif element["response"] is not None:
self.last_https[element["date"]] = element["response"]
+ time.sleep(element.get("duration", 0))
+
assert self.clean_body(request.body) == \
self.clean_body(element["body"]), "Body does not match"
context.status_code = element["status"]
pass
-class TimeMock:
- delta = 0
- true_time = time.time
- true_sleep = time.sleep
- time_patch = None
- datetime_patch = None
-
- @classmethod
- def start(cls, start_date):
- cls.delta = (datetime.datetime.now() - start_date).total_seconds()
-
- class fake_datetime(datetime.datetime):
- @classmethod
- def now(cls, tz=None):
- if tz is None:
- return cls.fromtimestamp(time.time())
- else:
- return tz.fromutc(cls.utcfromtimestamp(time.time()).replace(tzinfo=tz))
-
- cls.time_patch = mock.patch.multiple(time, time=cls.fake_time, sleep=cls.fake_sleep)
- cls.datetime_patch = mock.patch.multiple(datetime, datetime=fake_datetime)
- cls.time_patch.start()
- cls.datetime_patch.start()
-
- @classmethod
- def stop(cls):
- cls.delta = 0
- cls.datetime_patch.stop()
- cls.time_patch.stop()
-
- @classmethod
- def fake_time(cls):
- return cls.true_time() - cls.delta
-
- @classmethod
- def fake_sleep(cls, duration):
- cls.delta -= duration
- cls.true_sleep(0.2)
-
class AcceptanceTestCase():
def parse_file(self, report_file):
with open(report_file, "rb") as f:
self.requests_mock.start()
self.file_mock.start()
self.global_variables_mock.start()
- TimeMock.start(self.start_date)
+ TimeMock.travel(self.start_date)
def base_test(self):
main.main(self.config)
self.requests_mock.stop()
self.database_mock.stop()
-for dirfile in glob.glob("tests/acceptance/**/*/", recursive=True):
- json_files = glob.glob("{}/*.json".format(dirfile))
- log_files = glob.glob("{}/*.log".format(dirfile))
- if len(json_files) > 0:
- name = dirfile.replace("tests/acceptance/", "").replace("/", "_")[0:-1]
- cname = "".join(list(map(lambda x: x.capitalize(), name.split("_"))))
-
- globals()[cname] = type(cname,
- (AcceptanceTestCase,unittest.TestCase), {
- "log_files": log_files,
- "files": json_files,
- "test_{}".format(name): AcceptanceTestCase.base_test
- })
-
-if __name__ == '__main__':
- unittest.main()
-
from io import StringIO
import portfolio, market, main, store
-__all__ = ["unittest", "WebMockTestCase", "mock", "D",
+__all__ = ["limits", "unittest", "WebMockTestCase", "mock", "D",
"StringIO"]
+limits = ["acceptance", "unit"]
+for test_type in limits:
+ if "--no{}".format(test_type) in sys.argv:
+ sys.argv.remove("--no{}".format(test_type))
+ limits.remove(test_type)
+ if "--only{}".format(test_type) in sys.argv:
+ sys.argv.remove("--only{}".format(test_type))
+ limits = [test_type]
+ break
+
class WebMockTestCase(unittest.TestCase):
import time
--- /dev/null
+from .helper import limits
+from tests.acceptance import AcceptanceTestCase
+
+import unittest
+import glob
+
+__all__ = []
+
+for dirfile in glob.glob("tests/acceptance/**/*/", recursive=True):
+ json_files = glob.glob("{}/*.json".format(dirfile))
+ log_files = glob.glob("{}/*.log".format(dirfile))
+ if len(json_files) > 0:
+ name = dirfile.replace("tests/acceptance/", "").replace("/", "_")[0:-1]
+ cname = "".join(list(map(lambda x: x.capitalize(), name.split("_"))))
+
+ globals()[cname] = unittest.skipUnless("acceptance" in limits, "Acceptance skipped")(
+ type(cname, (AcceptanceTestCase, unittest.TestCase), {
+ "log_files": log_files,
+ "files": json_files,
+ "test_{}".format(name): AcceptanceTestCase.base_test
+ })
+ )
+ __all__.append(cname)
+
+
-from .helper import unittest, mock, D
+from .helper import limits, unittest, mock, D
import requests_mock
import market
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class poloniexETest(unittest.TestCase):
def setUp(self):
super().setUp()
from .helper import *
import main, market
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class MainTest(WebMockTestCase):
def test_make_order(self):
self.m.get_ticker.return_value = {
import market, store, portfolio
import datetime
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class MarketTest(WebMockTestCase):
def setUp(self):
super().setUp()
store_report.assert_called_once()
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class ProcessorTest(WebMockTestCase):
def test_values(self):
processor = market.Processor(self.m)
import portfolio
import datetime
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class ComputationTest(WebMockTestCase):
def test_compute_value(self):
compute = mock.Mock()
portfolio.Computation.compute_value("foo", "bid", compute_value="test")
compute.assert_called_with("foo", "bid")
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class TradeTest(WebMockTestCase):
def test_values_assertion(self):
self.assertEqual("ETH", as_json["currency"])
self.assertEqual("BTC", as_json["base_currency"])
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class BalanceTest(WebMockTestCase):
def test_values(self):
balance = portfolio.Balance("BTC", {
self.assertEqual(D(0), as_json["margin_available"])
self.assertEqual(D(0), as_json["margin_borrowed"])
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class OrderTest(WebMockTestCase):
def test_values(self):
order = portfolio.Order("buy", portfolio.Amount("ETH", 10),
result = order.retrieve_order()
self.assertFalse(result)
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class MouvementTest(WebMockTestCase):
def test_values(self):
mouvement = portfolio.Mouvement("ETH", "BTC", {
self.assertEqual("BTC", as_json["base_currency"])
self.assertEqual("ETH", as_json["currency"])
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class AmountTest(WebMockTestCase):
def test_values(self):
amount = portfolio.Amount("BTC", "0.65")
import threading
import market, portfolio, store
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class NoopLockTest(unittest.TestCase):
def test_with(self):
noop_lock = store.NoopLock()
with noop_lock:
self.assertTrue(True)
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class LockedVarTest(unittest.TestCase):
def test_values(self):
thread3.join()
self.assertEqual("Bar", locked_var.get()[0:3])
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class TradeStoreTest(WebMockTestCase):
def test_compute_trades(self):
self.m.balances.currencies.return_value = ["XMR", "DASH", "XVG", "BTC", "ETH"]
self.assertEqual([trade_mock1, trade_mock2], trade_store.pending)
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class BalanceStoreTest(WebMockTestCase):
def setUp(self):
super().setUp()
self.assertEqual(1, as_json["BTC"])
self.assertEqual(2, as_json["ETH"])
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class ReportStoreTest(WebMockTestCase):
def test_add_log(self):
with self.subTest(market=self.m):
'action': 'Hey'
})
+@unittest.skipUnless("unit" in limits, "Unit skipped")
class PortfolioTest(WebMockTestCase):
def setUp(self):
super().setUp()
projects / perso / Immae / Projets / Cryptomonnaies / Cryptoportfolio / Trader.git / commitdiff
