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()
-