Produits Entr'ouvert » Passerelle

import itertools

from django.apps import apps

from django.conf import settings

from django.core.exceptions import ValidationError, ObjectDoesNotExist, PermissionDenied

from django.core.urlresolvers import reverse

from django.db import connection, models, transaction

from django.db.models import Q

from django.test import override_settings

from django.utils.text import slugify

from django.utils import timezone, six

from django.utils import six

from django.utils.encoding import force_text

from django.utils.six.moves.urllib.parse import urlparse

from django.utils.translation import ugettext_lazy as _

from django.utils.timezone import now

    @property

    def level(self):

        return slugify(logging.getLevelName(self.levelno))

    def __str__(self):

        return '%s %s %s %s' % (self.timestamp, self.levelno, self.appname, self.slug)

STATUS_CHOICES = (

    ('unknown', _('Unknown')),

    ('up', _('Up')),

    ('down', _('Down')),

)

class ResourceStatus(models.Model):

        return self.__class__(self.connector, extra=dict(self.extra or {}, **kwargs))

    @property

    def level(self):

        return self._logger.getEffectiveLevel()

    def _log(self, levelname, message, *args, **kwargs):

        force = kwargs.pop('force', False)

        if self.connector.down() and not force:

            # don't log if the connector is known to be down

            return

        # hide credentials in urls

        for url in re.findall(r'(https?://\S+)', message):

            try:

                parsed = urlparse(url)

            else:

                sourceip = None

            attr['sourceip'] = sourceip

            if kwargs.get('exc_info'):

                (exc_type, exc_value, tb) = sys.exc_info()

                attr['extra']['error_summary'] = traceback.format_exception_only(exc_type, exc_value)

        admins = settings.ADMINS

        logging_parameters = self.connector.logging_parameters

        if logging_parameters.trace_emails:

            admins = [('', x) for x in logging_parameters.trace_emails.splitlines()]

        with override_settings(ADMINS=admins):

            getattr(self._logger, levelname.lower())(message, *args, **kwargs)

    def exception(self, message, *args, **kwargs):

        kwargs['exc_info'] = 1

    def debug(self, message, *args, **kwargs):

    def info(self, message, *args, **kwargs):

    def warning(self, message, *args, **kwargs):

    def critical(self, message, *args, **kwargs):

    def error(self, message, *args, **kwargs):

    def fatal(self, message, *args, **kwargs):

class HTTPResource(models.Model):

    '''Mixin to add basic TLS/Basic HTTP authentication fields to any

       resource.'''

    basic_auth_username = models.CharField(

        max_length=128,

        verbose_name=_('Basic authentication username'),

        connector_name, endpoint_name = kwargs['connector'], kwargs['endpoint']

        connector = self.get_object()

        url = request.get_full_path()

        payload = request.body[:connector.logging_parameters.requests_max_size]

        try:

            payload = payload.decode('utf-8')

        except UnicodeDecodeError:

            payload = '<BINARY PAYLOAD>'

                              (request.method, url, payload),

                              extra={

                                  'request': request,

                                  'connector': connector_name,

                                  'connector_endpoint': endpoint_name,

                                  'connector_endpoint_method': request.method,

                                  'connector_endpoint_url': url,

                                  'connector_payload': payload

        params = self.get_params(request, *args, **kwargs)

        if request.method == 'GET' and self.endpoint.endpoint_info.cache_duration:

            cache_key = hashlib.md5(

                force_bytes(repr(self.get_object().slug) + repr(self.endpoint) + repr(params))

            ).hexdigest()

            result = cache.get(cache_key)

            if result is not None:

                return result

        result = self.endpoint(request, **params)

        if request.method == 'GET' and self.endpoint.endpoint_info.cache_duration:

            cache.set(cache_key, result, self.endpoint.endpoint_info.cache_duration)

        return result

    def get(self, request, *args, **kwargs):

        if self.endpoint.endpoint_info.pattern:

            pattern = url(self.endpoint.endpoint_info.pattern, self.endpoint)

            match = pattern.resolve(kwargs.get('rest') or '')

import pytest

import utils

from django.contrib.contenttypes.models import ContentType

from django.core.urlresolvers import reverse

from passerelle.apps.arcgis.models import ArcGIS

from passerelle.base.models import ResourceLog, ProxyLogger, BaseResource, HTTPResource, LoggingParameters

from passerelle.base.models import ResourceStatus

from passerelle.apps.mdel.models import MDEL

from passerelle.contrib.stub_invoices.models import StubInvoicesConnector

from passerelle.utils.api import endpoint

from test_manager import login, admin_user, simple_user

@pytest.fixture

def arcgis(db):

    instance = utils.setup_access_rights(ArcGIS.objects.create(slug='test'))

    instance.set_log_level('DEBUG')

    return instance

DEMAND_STATUS = {

    'closed': True,

    'status': 'accepted',

    'comment': 'dossier trait\xe9.'

}

@mock.patch('passerelle.apps.mdel.models.Demand.get_status', lambda x: DEMAND_STATUS)

@mock.patch('passerelle.utils.Request.get')

def test_proxy_logger(mocked_get, caplog, app, arcgis):

    payload = open(os.path.join(os.path.dirname(__file__), 'data', 'nancy_arcgis', 'sigresponse.json')).read()

    mocked_get.return_value = utils.FakedResponse(content=payload, status_code=200)

    # simple logger

    logger = ProxyLogger(connector=arcgis)

    logger.debug('this is a debug test')

    logger.info('this is an info test')

    assert ResourceLog.objects.count() == 2

    for log in ResourceLog.objects.all():

        if log.levelno == 10:

            assert log.message == 'this is a debug test'

    log = ResourceLog.objects.filter(appname='arcgis', slug='test').first()

    assert log.appname == 'arcgis'

    assert log.slug == 'test'

    assert log.levelno == 20

    assert log.sourceip == '127.0.0.1'

    assert log.extra['connector'] == 'arcgis'

    assert log.extra['connector_endpoint'] == 'district'

    assert log.extra['connector_endpoint_method'] == 'GET'

    assert '/arcgis/test/district?' in log.extra['connector_endpoint_url']

    # Resource Generic Logger

    for record in caplog.records:

        if record.name != 'passerelle.resource.arcgis.test':

            continue

        assert record.levelno == 20

        assert record.levelname == 'INFO'

    app.get('/arcgis/test/district', params={'lon': 6.172122, 'lat': 48.673836}, status=200)

    assert ResourceLog.objects.count() == 1

    arcgis.logger.warning('first warning')

    assert ResourceLog.objects.count() == 2

    assert ResourceLog.objects.last().message == 'first warning'

    assert ResourceLog.objects.last().levelno == 30

class FakeConnectorBase(object):

    slug = 'connector'

    def get_connector_slug(self):

        return 'fake'

    @endpoint()