shopify_test_helpers.py

import os
import sys
import unittest
from pprint import pformat
from pyactiveresource.activeresource import ActiveResource
from pyactiveresource.connection import Request
from pyactiveresource.testing import http_fake
import shopify
from pyactiveresource.testing.http_fake import TestHandler, FakeResponse, Error
from typing import Callable


#TODO - remove 0 not used right now (idea was to be able to accept mocks)
from shopify_app import shopify_settings


class HttpFakeImprovedTestHandler(TestHandler):
    def respond_to(self, method, path,
                   request_headers, body, assert_lambda: Callable[[Request], None] = lambda *args: None, code=200, response_headers=None):
        before = set(dict(self._response_map))
        super().respond_to(self, method, path, request_headers, body, code, response_headers)
        after = set(dict(self._response_map))
        new_key_dict :dict  = after - before
        added_key = new_key_dict.keys()[0]

        #let's add the assert_lambda to the map that is created on the above call
        added_value = self._response_map[added_key]

        added_value = (*added_value, assert_lambda)


    def do_open(self, http_class, request, **http_conn_args):
        """Return the response object for the given request.

               Overrides the HTTPHandler method of the same name to return a
               FakeResponse instead of creating any network connections.

               Args:
                   http_class: The http protocol being used.
                   request: A urllib.request.Request object.
               Returns:
                   A FakeResponse object.
               """
        self.__class__.request = request  # Store the most recent request object
        if self._response_map:
            key = self.create_response_key(
                request.get_method(), request.get_full_url(), request.headers)
            if str(key) in self._response_map:
                value = self._response_map[str(key)]
                if len(value) == 4:
                    (code, body, response_headers, fun_request) = value
                    fun_request(request)
                else:

                    (code, body, response_headers) = value
                return FakeResponse(code, body, response_headers)
            else:
                raise Error('Unknown request %s %s'
                            '\nrequest:%s\nresponse_map:%s' % (
                                request.get_method(), request.get_full_url(),
                                str(key), pformat(list(self._response_map.keys()))))
        elif isinstance(self._response, Exception):
            raise (self._response)
        else:
            return self._response




class ShopifyTestCase(unittest.TestCase):

    def setUp(self):
        ActiveResource.site = None
        ActiveResource.headers=None

        shopify.ShopifyResource.clear_session()
        shopify.ShopifyResource.site = f"https://phony.myshopify.com/admin/api/{shopify_settings.SHOPIFY_VERSION}"
        shopify.ShopifyResource.password = None
        shopify.ShopifyResource.user = None

        http_fake.initialize()
        self.http = http_fake.TestHandler
        self.http.set_response(Exception('Bad request'))
        self.http.site = 'https://phony.myshopify.com'

    def load_fixture(self, name, format='json'):
        with open(os.path.dirname(__file__)+'/test_fixtures/%s.%s' % (name, format), 'rb') as f:
            return f.read()

    def fake(self, endpoint, **kwargs):
        body = kwargs.pop('body', None) or self.load_fixture(endpoint)
        format = kwargs.pop('format','json')
        method = kwargs.pop('method','GET')
        prefix = kwargs.pop('prefix', f'/admin/api/{shopify_settings.SHOPIFY_VERSION}')
        assert_lambda: function = kwargs.pop('assert_lambda', lambda *args:None)

        if ('extension' in kwargs and not kwargs['extension']):
            extension = ""
        else:
            extension = ".%s" % (kwargs.pop('extension', 'json'))

        url = "https://phony.myshopify.com%s/%s%s" % (prefix, endpoint, extension)
        try:
           url = kwargs['url']
        except KeyError:
           pass

        headers = {}
        if kwargs.pop('has_user_agent', True):
            userAgent = 'ShopifyPythonAPI/%s Python/%s' % (shopify.VERSION, sys.version.split(' ', 1)[0])
            headers['User-agent'] = userAgent
            headers['X-shopify-access-token'] = 'NA'

        try:
            headers.update(kwargs['headers'])
        except KeyError:
           pass

        code = kwargs.pop('code', 200)

# assert lambda support - not used RN
        # self.http.respond_to(method=method, path=url, request_headers=headers, body=body, assert_lambda=assert_lambda, code=code,
        #     response_headers=kwargs.pop('response_headers', None))
        self.http.respond_to(method=method, path=url, request_headers=headers, body=body, code=code,
                             response_headers=kwargs.pop('response_headers', None))

Work in progress - to allow lambdas with asserts when doing a self.fake(