Produits Entr'ouvert » Passerelle

# -*- coding: utf-8 -*-

# Generated by Django 1.11.17 on 2019-02-05 10:26

from __future__ import unicode_literals

from django.db import migrations, models

import django.db.models.deletion

class Migration(migrations.Migration):

    dependencies = [

        ('contenttypes', '0002_remove_content_type_name'),

        ('base', '0010_loggingparameters_trace_emails'),

    ]

    operations = [

        migrations.CreateModel(

            name='AvailabilityParameters',

            fields=[

                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),

                ('resource_pk', models.PositiveIntegerField()),

                ('run_check', models.BooleanField(default=True, verbose_name='Run regular availability checks')),

                ('resource_type', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='contenttypes.ContentType')),

            ],

        ),

        migrations.AlterUniqueTogether(

            name='availabilityparameters',

            unique_together=set([('resource_type', 'resource_pk')]),

        ),

    ]

    @property

    def availability_parameters(self):

        resource_type = ContentType.objects.get_for_model(self)

        try:

            return AvailabilityParameters.objects.get(

                    resource_type=resource_type,

                    resource_pk=self.id)

        except AvailabilityParameters.DoesNotExist:

            return AvailabilityParameters(

                    resource_type=resource_type,

                    resource_pk=self.id)

    check_status.not_implemented = True

        # eventually skip it

        if not self.availability_parameters.run_check:

            return

class AvailabilityParameters(models.Model):

    resource_type = models.ForeignKey(ContentType)

    resource_pk = models.PositiveIntegerField()

    resource = fields.GenericForeignKey('resource_type', 'resource_pk')

    run_check = models.BooleanField(

        default=True, verbose_name=_('Run regular availability checks'),

        help_text=_('Run an availability check every 5 minutes'))

    class Meta:

        unique_together = (('resource_type', 'resource_pk'))

        LoggingParametersUpdateView, ManageAvailabilityView

        LoggingParametersUpdateView.as_view(), name='logging-parameters'),

    url(r'manage/availability/(?P<resource_type>[\w,-]+)/(?P<resource_pk>[\w,-]+)/$',

        ManageAvailabilityView.as_view(), name='manage-availability')

from .models import ApiUser, AccessRight, LoggingParameters, AvailabilityParameters, ResourceStatus

class ManageAvailabilityView(FormView):

    template_name = 'passerelle/manage/manage_availability_form.html'

    def get_context_data(self, **kwargs):

        context = super(ManageAvailabilityView, self).get_context_data(**kwargs)

        connector = self.get_resource()

        context['connector'] = connector

        context['availability_status'] = connector.get_availability_status()

        return context

    def get_form_class(self):

        form_class = model_forms.modelform_factory(

                AvailabilityParameters,

                fields=['run_check'])

        return form_class

    def get_initial(self):

        d = self.initial.copy()

        d['resource_type'] = self.kwargs['resource_type']

        d['resource_pk'] = self.kwargs['resource_pk']

        d['run_check'] = self.get_resource().availability_parameters.run_check

        return d

    def get_resource(self):

        content_type = ContentType.objects.get_for_id(self.kwargs['resource_type'])

        return content_type.model_class().objects.get(pk=self.kwargs['resource_pk'])

    def get_success_url(self):

        return self.get_resource().get_absolute_url()

    def form_valid(self, form):

        resource = self.get_resource()

        parameters = resource.availability_parameters

        run_check = form.cleaned_data['run_check']

        if not run_check and resource.down():

            resource_type = ContentType.objects.get_for_model(resource)

            ResourceStatus(

                resource_type=resource_type,

                resource_pk=self.kwargs['resource_pk'],

                status='up',

                message='').save()

        if parameters.run_check != run_check:

            parameters.run_check = run_check

            parameters.save()

            resource.logger.info(u'availability checks %s', 'enabled' if run_check else 'disabled')

        return super(ManageAvailabilityView, self).form_valid(form)

    def check_status(self):

        auth_url = urlparse.urljoin(self.url, 'auth')

        response = self.requests.get(auth_url, headers={'MH-LOGIN': self.username})

        response.raise_for_status()

{% extends "passerelle/manage.html" %}

{% load i18n %}

{% block appbar %}

<h2>{% trans 'Connector Status' %} : {{availability_status.status}}</h2>

{% endblock %}

{% block content %}

<form method="post">

  {% csrf_token %}

  {{ form.as_p }}

  <div class="buttons">

    <button class="submit-button">{% trans 'Save' %}</button>

    <a href="{{ connector.get_absolute_url }}" class="cancel">{% trans 'Cancel' %}</a>

  </div>

</form>

{% endblock %}

{% if object|can_edit:request.user and has_check_status %}

<a rel="popup" href="{% url 'manage-availability' resource_type=object|resource_type resource_pk=object.id %}">{% trans 'availability check' %}</a>

{% endif %}

    def get_context_data(self, slug=None, **kwargs):

        context = super(GenericConnectorView, self).get_context_data(**kwargs)

        context['has_check_status'] = not hasattr(

            context['object'].check_status, 'not_implemented')

        return context

def test_availability_checks_disabled(app, connector):

    with HTTMock(up_mock):

        connector.availability()

    assert connector.get_availability_status().up()

    av = connector.availability_parameters

    av.run_check = False

    av.save()

    with HTTMock(down_mock):

        connector.availability()

    assert connector.get_availability_status().up()

    av.run_check = True

    av.save()

    with HTTMock(down_mock):

        connector.availability()

    assert connector.get_availability_status().down()

from passerelle.base.models import ApiUser, AccessRight, ResourceLog, ResourceStatus

def test_availability_parameters(app, admin_user, monkeypatch):

    data = StringIO('1;Foo\n2;Bar\n3;Baz')

    csv = CsvDataSource.objects.create(

        csv_file=File(data, 't.csv'),

        columns_keynames='id, text', slug='test', title='a title', description='a description')

    app = login(app)

    resp = app.get(csv.get_absolute_url())

    assert csv.availability_parameters.run_check

    # csv connector has the default check_status which does nothing

    # so availability check is hidden

    assert 'availability check' not in resp.text

    def check_status(*args, **kwargs):

        return True

    monkeypatch.setattr(CsvDataSource, 'check_status', check_status)

    resp = app.get(csv.get_absolute_url())

    assert 'availability check' in resp.text

    resp = resp.click('availability check')

    assert 'up' in resp.text

    resp.form['run_check'] = False

    resp = resp.form.submit()

    # Connector status not changed, availability parameters changed

    assert not csv.availability_parameters.run_check

    resp = app.get(csv.get_absolute_url())

    resp = resp.click('availability check')

    resp.form['run_check'] = True

    resp = resp.form.submit()

    # Connector down

    resource_type = ContentType.objects.get_for_model(csv)

    status = ResourceStatus(

        resource_type=resource_type, resource_pk=csv.pk,

        status='down', message='')

    status.save()

    assert csv.down()

    resp = app.get(csv.get_absolute_url())

    resp = resp.click('availability check')

    resp.form['run_check'] = False

    resp = resp.form.submit()

    # Connector is put back up

    assert not csv.availability_parameters.run_check

    assert not csv.down()

    status = csv.get_availability_status()

    assert status.status == 'up'