passerelle/tests/utils.py

import json

import httmock
import mock
from django.contrib.contenttypes.models import ContentType
from django.urls import reverse
from django.utils.six.moves.urllib import parse as urlparse

from passerelle.base.models import AccessRight, ApiUser
from passerelle.compat import json_loads


def generic_endpoint_url(connector, endpoint, slug='test'):
    return reverse('generic-endpoint', kwargs={'connector': connector, 'slug': slug, 'endpoint': endpoint})


def setup_access_rights(obj):
    api, created = ApiUser.objects.get_or_create(username='all', keytype='', key='')
    obj_type = ContentType.objects.get_for_model(obj)
    AccessRight.objects.create(codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=obj.pk)
    return obj


class FakedResponse(mock.Mock):
    headers = {}

    def json(self):
        return json_loads(self.content)


def mock_url(url=None, response='', status_code=200, headers=None, reason=None, exception=None, qs=None):
    urlmatch_kwargs = {}
    if url:
        parsed = urlparse.urlparse(url)
        if parsed.netloc:
            urlmatch_kwargs['netloc'] = parsed.netloc
        if parsed.path:
            urlmatch_kwargs['path'] = parsed.path

    if not isinstance(response, str):
        response = json.dumps(response)

    @httmock.remember_called
    @httmock.urlmatch(**urlmatch_kwargs)
    def mocked(url, request):
        if qs is not None:
            qs.update(urlparse.parse_qsl(url.query))
        if exception:
            raise exception
        return httmock.response(status_code, response, headers, reason, request=request)

    return httmock.HTTMock(mocked)


def make_resource(model_class, **kwargs):
    resource = model_class.objects.create(**kwargs)
    setup_access_rights(resource)
    return resource


def endpoint_get(expected_url, app, resource, endpoint, **kwargs):
    url = generic_endpoint_url(
        connector=resource.__class__.get_connector_slug(), endpoint=endpoint, slug=resource.slug
    )
    assert url == expected_url, 'endpoint URL has changed'
    return app.get(url, **kwargs)