path: root/test_acceptance.py

import requests
import requests_mock
import sys, os
import time, datetime
import unittest
from unittest import mock
from ssl import SSLError
from decimal import Decimal
import simplejson as json
import psycopg2
from io import StringIO

class FileMock:
    @classmethod
    def start(cls):
        cls.file_mock = mock.patch("market.open")
        cls.os_mock = mock.patch("os.makedirs")
        cls.stdout_mock = mock.patch('sys.stdout', new_callable=StringIO)
        cls.stdout_mock.start()
        cls.os_mock.start()
        cls.file_mock.start()

    @classmethod
    def check_calls(cls, tester):
        pass
        #raise NotImplementedError("Todo")

    @classmethod
    def stop(cls):
        cls.file_mock.stop()
        cls.stdout_mock.stop()
        cls.os_mock.stop()

class DatabaseMock:
    rows = []
    report_db = False
    db_patch = None
    cursor = None
    total_report_lines = 0
    db_mock = None
    requests = []

    @classmethod
    def start(cls, reports, report_db):
        cls.report_db = report_db
        cls.rows = []
        cls.total_report_lines=  0
        for user_id, market_id, http_requests, report_lines in reports.values():
            cls.rows.append( (market_id, { "key": "key", "secret": "secret" }, user_id) )
            cls.total_report_lines += len(report_lines)

        connect_mock = mock.Mock()
        cls.cursor = mock.MagicMock()
        connect_mock.cursor.return_value = cls.cursor
        def _execute(request, *args):
            cls.requests.append(request)
        cls.cursor.execute.side_effect = _execute
        cls.cursor.__iter__.return_value = cls.rows

        cls.db_patch = mock.patch("psycopg2.connect")
        cls.db_mock = cls.db_patch.start()
        cls.db_mock.return_value = connect_mock

    @classmethod
    def check_calls(cls, tester):
        if cls.report_db:
            tester.assertEqual(1 + len(cls.rows), cls.db_mock.call_count)
            tester.assertEqual(1 + len(cls.rows) + cls.total_report_lines, cls.cursor.execute.call_count)
        else:
            tester.assertEqual(1, cls.db_mock.call_count)
            tester.assertEqual(1, cls.cursor.execute.call_count)

    @classmethod
    def stop(cls):
        cls.db_patch.stop()
        cls.db_mock = None
        cls.cursor = None
        cls.total_report_lines = 0
        cls.rows = []
        cls.report_db = False
        cls.requests = []

class RequestsMock:
    adapter = None
    mocks = {}
    last_https = {}
    request_patch = []

    @classmethod
    def start(cls, reports):
        cls.adapter = requests_mock.Adapter()
        cls.adapter.register_uri(requests_mock.ANY, requests_mock.ANY,
                exc=requests_mock.exceptions.MockException("Not stubbed URL"))
        true_session = requests.Session

        cls.mocks = {}

        for market_id, elements in reports.items():
            for element in elements:
                method = element["method"]
                url = element["url"]
                cls.mocks \
                        .setdefault((method, url), {}) \
                        .setdefault(market_id, []) \
                        .append(element)

        for ((method, url), elements) in cls.mocks.items():
            cls.adapter.register_uri(method, url, text=cls.callback_func(elements), complete_qs=True)
        def _session():
            session = true_session()
            session.get_adapter = lambda url: cls.adapter
            return session
        cls.request_patch = [
                mock.patch.object(requests.sessions, "Session", new=_session),
                mock.patch.object(requests, "Session", new=_session)
                ]
        for patch in cls.request_patch:
            patch.start()

    @classmethod
    def stop(cls):
        for patch in cls.request_patch:
            patch.stop()
        cls.request_patch = []
        cls.last_https = {}
        cls.mocks = {}
        cls.adapter = None

    @classmethod
    def check_calls(cls, tester):
        for (method, url), elements in cls.mocks.items():
            for market_id, element in elements.items():
                tester.assertEqual(0, len(element), "Missing calls to {} {}, market_id {}".format(method, url, market_id))

    @classmethod
    def clean_body(cls, body):
        if body is None:
            return None
        import re
        if isinstance(body, bytes):
            body = body.decode()
        body = re.sub(r"&nonce=\d*$", "", body)
        body = re.sub(r"nonce=\d*&?", "", body)
        return body

    @classmethod
    def callback_func(cls, elements):
        def callback(request, context):
            try:
                element = elements[request.headers.get("X-market-id")].pop(0)
            except (IndexError, KeyError):
                raise RuntimeError("Unexpected call")
            if element["response"] is None and element["response_same_as"] is not None:
                element["response"] = cls.last_https[element["response_same_as"]]
            elif element["response"] is not None:
                cls.last_https[element["date"]] = element["response"]

            assert cls.clean_body(request.body) == \
                    cls.clean_body(element["body"]), "Body does not match"
            context.status_code = element["status"]
            if "error" in element:
                if element["error"] == "SSLError":
                    raise SSLError(element["error_message"])
                else:
                    raise getattr(requests.exceptions, element["error"])(element["error_message"])
            return element["response"]
        return callback

class TimeMock:
    delta = 0
    true_time = time.time
    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

class AcceptanceTestCase():
    def parse_file(self, report_file):
        with open(report_file, "rb") as f:
            json_content = json.load(f, parse_float=Decimal)
        config, user, date, market_id = self.parse_config(json_content)
        http_requests = self.parse_requests(json_content)

        return config, user, date, market_id, http_requests, json_content

    def parse_requests(self, json_content):
        http_requests = []
        for element in json_content:
            if element["type"] != "http_request":
                continue
            http_requests.append(element)
        return http_requests

    def parse_config(self, json_content):
        market_info = None
        for element in json_content:
            if element["type"] != "market":
                continue
            market_info = element
        assert market_info is not None, "Couldn't find market element"

        args = market_info["args"]
        config = []
        for arg in ["before", "after", "quiet", "debug"]:
            if args.get(arg, False):
                config.append("--{}".format(arg))
        for arg in ["parallel", "report_db"]:
            if not args.get(arg, False):
                config.append("--no-{}".format(arg.replace("_", "-")))
        for action in args.get("action", []):
            config.extend(["--action", action])
        if args.get("report_path") is not None:
            config.extend(["--report-path", args.get("report_path")])
        if args.get("user") is not None:
            config.extend(["--user", args.get("user")])
        config.extend(["--config", ""])

        user = market_info["user_id"]
        date = datetime.datetime.strptime(market_info["date"], "%Y-%m-%dT%H:%M:%S.%f")
        market_id = market_info["market_id"]
        return config, user, date, market_id

    def requests_by_market(self):
        r = {
                None: []
                }
        got_common = False
        for user_id, market_id, http_requests, report_lines in self.reports.values():
            r[str(market_id)] = []
            for http_request in http_requests:
                if http_request["market_id"] is None:
                    if not got_common:
                        r[None].append(http_request)
                else:
                    r[str(market_id)].append(http_request)
            got_common = True
        return r

    def setUp(self):
        if not hasattr(self, "files"):
            raise "This class expects to be inherited with a class defining self.files in setUp"

        self.reports = {}
        self.start_date = datetime.datetime.now()
        self.config = []
        for f in self.files:
            self.config, user_id, date, market_id, http_requests, report_lines = self.parse_file(f)
            if date < self.start_date:
                self.start_date = date
            self.reports[f] = [user_id, market_id, http_requests, report_lines]

        DatabaseMock.start(self.reports, "--no-report-db" not in self.config)
        RequestsMock.start(self.requests_by_market())
        FileMock.start()
        TimeMock.start(self.start_date)

    def base_test(self):
        import main
        main.main(self.config)
        RequestsMock.check_calls(self)
        DatabaseMock.check_calls(self)
        FileMock.check_calls(self)

    def tearDown(self):
        TimeMock.stop()
        FileMock.stop()
        RequestsMock.stop()
        DatabaseMock.stop()

import glob
for dirfile in glob.glob("tests/acceptance/**/*/", recursive=True):
    json_files = glob.glob("{}/*.json".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), {
            "files": json_files,
            "test_{}".format(name): AcceptanceTestCase.base_test
            })

if __name__ == '__main__':
    unittest.main()