Source code for drf_api_checker.unittest

# -*- coding: utf-8 -*-
import inspect
import os

from rest_framework.test import APIClient

from drf_api_checker.fs import clean_url, get_filename
from drf_api_checker.recorder import BASE_DATADIR, DEFAULT_CHECKS, Recorder
from drf_api_checker.utils import dump_fixtures, load_fixtures


[docs]class ApiCheckerMixin: """ Mixin to enable API contract check How to use it: - implement get_fixtures() to create data for test. It should returns a dictionary - use self.assert<METHOD>(url) to check urls contract Example: class TestAPIAgreements(ApiCheckerMixin, TestCase): def get_fixtures(self): return {'customer': CustomerFactory()} def test_customer_detail(self): url = reverse("customer-detail", args=[self.get_fixture('customer').pk]) self.assertGET(url) """ recorder_class = Recorder client_class = APIClient expect_errors = False allow_empty = False checks = list(DEFAULT_CHECKS) def setUp(self): super().setUp() # self.client = self.client_class() if self.client_class else APIClient() self._process_fixtures() self.recorder = self.recorder_class(self.data_dir, self) if hasattr(self, 'check_headers'): raise DeprecationWarning("'check_headers' has been deprecated. Use 'checks' instead.") if hasattr(self, 'check_status'): raise DeprecationWarning("'check_status' has been deprecated. Use 'checks' instead.") @property def data_dir(self): cls = type(self) return os.path.join(os.path.dirname(inspect.getfile(cls)), BASE_DATADIR, cls.__module__, cls.__name__) def get_fixtures_filename(self, basename='fixtures'): return get_filename(self.data_dir, f'{basename}.json') def get_fixtures(self): """ returns test fixtures. Should returns a dictionary where any key is a fixture name the value should be a Model instance (or a list). {'user' : UserFactory(username='user'), 'partners': [PartnerFactory(), PartnerFactory()], } fixtures can be accessed using `get_fixture(<name>)` """ return {} # pragma: no cover def get_fixture(self, name): """ returns fixture `name` loaded by `get_fixtures()` """ return self.__fixtures[name] # pragma: no cover def _process_fixtures(self): """ store or retrieve test fixtures """ fname = self.get_fixtures_filename() if os.path.exists(fname) and not os.environ.get('API_CHECKER_RESET'): self.__fixtures = load_fixtures(fname) else: self.__fixtures = self.get_fixtures() if self.__fixtures: dump_fixtures(self.__fixtures, fname) def _assertCALL(self, url, *, allow_empty=None, checks=None, expect_errors=None, name=None, method='get', data=None, **kwargs): if 'check_headers' in kwargs: raise DeprecationWarning("'check_headers' has been deprecated. Use 'checks' instead.") if 'check_status' in kwargs: raise DeprecationWarning("'check_status' has been deprecated. Use 'checks' instead.") if kwargs: raise AttributeError("Unknown arguments %s" % kwargs.keys()) expect_errors = self.expect_errors if expect_errors is None else expect_errors allow_empty = self.allow_empty if allow_empty is None else allow_empty self.recorder.assertCALL(url, method=method, allow_empty=allow_empty, checks=checks, expect_errors=expect_errors, name=name, data=data) def assertGET(self, url, allow_empty=None, checks=None, expect_errors=None, name=None, data=None, **kwargs): if 'check_headers' in kwargs: raise DeprecationWarning("'check_headers' has been deprecated. Use 'checks' instead.") if 'check_status' in kwargs: raise DeprecationWarning("'check_status' has been deprecated. Use 'checks' instead.") if kwargs: raise AttributeError("Unknown arguments %s" % kwargs.keys()) self._assertCALL(url, method='get', allow_empty=allow_empty, checks=checks, expect_errors=expect_errors, name=name, data=data) def assertPUT(self, url, data, allow_empty=None, checks=None, expect_errors=None, name=None, **kwargs): if 'check_headers' in kwargs: raise DeprecationWarning("'check_headers' has been deprecated. Use 'checks' instead.") if 'check_status' in kwargs: raise DeprecationWarning("'check_status' has been deprecated. Use 'checks' instead.") if kwargs: raise AttributeError("Unknown arguments %s" % kwargs.keys()) self._assertCALL(url, method='put', allow_empty=allow_empty, checks=checks, expect_errors=expect_errors, name=name, data=data) def assertPOST(self, url, data, allow_empty=None, checks=None, expect_errors=None, name=None, **kwargs): if 'check_headers' in kwargs: raise DeprecationWarning("'check_headers' has been deprecated. Use 'checks' instead.") if 'check_status' in kwargs: raise DeprecationWarning("'check_status' has been deprecated. Use 'checks' instead.") if kwargs: raise AttributeError("Unknown arguments %s" % kwargs.keys()) self._assertCALL(url, data=data, method='post', allow_empty=allow_empty, checks=checks, expect_errors=expect_errors, name=name) def assertDELETE(self, url, allow_empty=None, checks=None, expect_errors=None, name=None, **kwargs): if 'check_headers' in kwargs: raise DeprecationWarning("'check_headers' has been deprecated. Use 'checks' instead.") if 'check_status' in kwargs: raise DeprecationWarning("'check_status' has been deprecated. Use 'checks' instead.") if kwargs: raise AttributeError("Unknown arguments %s" % kwargs.keys()) self._assertCALL(url, method='delete', allow_empty=allow_empty, checks=checks, expect_errors=expect_errors, name=name)
[docs]class ApiCheckerBase(type): """ Custom _type_, intended to be used as metaclass. It will create a test for each url defined in URLS in the format ``test__<normalized_url_path>``, if a method with the same name is found the creation is skipped reading this as an intention to have a custom test for that url. class TestAPIIntervention(TestCase, metaclass=ApiCheckerBase): URLS = [ reverse("intervention-list"), reverse("intervention-detail", args=[101]), ] def get_fixtures(cls): return {'intervention': InterventionFactory(id=101), 'result': ResultFactory(), } running this code will produce... ... test_url__api_v2_interventions (etools.applications.partners.tests.test_api.TestAPIIntervention) ... ok test_url__api_v2_interventions_101 (etools.applications.partners.tests.test_api.TestAPIIntervention) ... ok ... """ mixin = ApiCheckerMixin def __new__(cls, clsname, superclasses, attributedict): superclasses = (cls.mixin,) + superclasses clazz = type.__new__(cls, clsname, superclasses, attributedict) def check_url(url): if isinstance(url, (list, tuple)): url, data = url else: data = None def _inner(self): self.assertGET(url, data=data) _inner.__name__ = "test_url__" + clean_url('get', url, data) return _inner if 'URLS' not in attributedict: # pragma: no cover raise ValueError(f"Error creating {clsname}. " f"ApiCheckerBase requires URLS attribute ") for u in attributedict['URLS']: m = check_url(u) if not hasattr(clazz, m.__name__): setattr(clazz, m.__name__, m) return clazz