Produits Entr'ouvert » Passerelle

# Generated by Django 2.2.19 on 2021-05-06 22:50

import django.contrib.postgres.fields.jsonb

from django.db import migrations, models

import django.db.models.deletion

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

    operations = [

        migrations.CreateModel(

            name='ToulouseSmartResource',

            fields=[

                (

                    'id',

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

                ),

                ('title', models.CharField(max_length=50, verbose_name='Title')),

                ('slug', models.SlugField(unique=True, verbose_name='Identifier')),

                ('description', models.TextField(verbose_name='Description')),

                (

                    'basic_auth_username',

                    models.CharField(

                        blank=True, max_length=128, verbose_name='Basic authentication username'

                    ),

                ),

                (

                    'basic_auth_password',

                    models.CharField(

                        blank=True, max_length=128, verbose_name='Basic authentication password'

                    ),

                ),

                (

                    'client_certificate',

                    models.FileField(

                        blank=True, null=True, upload_to='', verbose_name='TLS client certificate'

                    ),

                ),

                (

                    'trusted_certificate_authorities',

                    models.FileField(blank=True, null=True, upload_to='', verbose_name='TLS trusted CAs'),

                ),

                (

                    'verify_cert',

                    models.BooleanField(blank=True, default=True, verbose_name='TLS verify certificates'),

                ),

                (

                    'http_proxy',

                    models.CharField(blank=True, max_length=128, verbose_name='HTTP and HTTPS proxy'),

                ),

                ('webservice_base_url', models.URLField(verbose_name='Webservice Base URL')),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True,

                        related_name='_toulousesmartresource_users_+',

                        related_query_name='+',

                        to='base.ApiUser',

                    ),

                ),

            ],

            options={

                'verbose_name': 'Toulouse Smart',

            },

        ),

        migrations.CreateModel(

            name='Cache',

            fields=[

                (

                    'id',

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

                ),

                ('key', models.CharField(max_length=64, verbose_name='Key')),

                ('timestamp', models.DateTimeField(auto_now=True, verbose_name='Timestamp')),

                ('value', django.contrib.postgres.fields.jsonb.JSONField(default=dict, verbose_name='Value')),

                (

                    'resource',

                    models.ForeignKey(

                        on_delete=django.db.models.deletion.CASCADE,

                        to='toulouse_smart.ToulouseSmartResource',

                        verbose_name='Resource',

                    ),

                ),

            ],

        ),

    ]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2021 Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import datetime

from django.db import models

from django.utils.timezone import now

from django.utils.translation import ugettext_lazy as _

from django.contrib.postgres.fields import JSONField

import lxml.etree as ET

from passerelle.base.models import BaseResource, HTTPResource

from passerelle.utils import xml

from passerelle.utils.api import endpoint

class ToulouseSmartResource(BaseResource, HTTPResource):

    category = _('Business Process Connectors')

    webservice_base_url = models.URLField(_('Webservice Base URL'))

    log_requests_errors = False

    class Meta:

        verbose_name = _('Toulouse Smart')

    def get_intervention_types(self):

        try:

            return self.get('intervention-types', max_age=120)

        except KeyError:

            pass

        try:

            url = self.webservice_base_url + 'v1/type-intervention'

            response = self.requests.get(url)

            doc = ET.fromstring(response.content)

            intervention_types = []

            for xml_item in doc:

                item = xml.to_json(xml_item)

                for prop in item.get('properties', []):

                    prop['required'] = prop.get('required') == 'true'

                intervention_types.append(item)

            intervention_types.sort(key=lambda x: x['name'])

            for i, intervention_type in enumerate(intervention_types):

                intervention_type['order'] = i + 1

        except Exception:

            try:

                return self.get('intervention-types')

            except KeyError:

                raise

        self.set('intervention-types', intervention_types)

        return intervention_types

    def get(self, key, max_age=None):

        cache_entries = self.cache_entries

        if max_age:

            cache_entries = cache_entries.filter(timestamp__gt=now() - datetime.timedelta(seconds=max_age))

        try:

            return cache_entries.get(key='intervention-types').value

        except Cache.DoesNotExist:

            raise KeyError(key)

    def set(self, key, value):

        self.cache_entries.update_or_create(key=key, defaults={'value': value})

    @endpoint(

        name='type-intervention',

        description=_('Get intervention types'),

        perm='can_access',

    )

    def type_intervention(self, request):

        try:

            return {

                'data': [

                    {

                        'id': intervention_type['id'],

                        'text': intervention_type['name'],

                    }

                    for intervention_type in self.get_intervention_types()

                ]

            }

        except Exception:

            return {

                'data': [

                    {

                        'id': '',

                        'text': _('Service is unavailable'),

                        'disabled': True,

                    }

                ]

            }

class Cache(models.Model):

    resource = models.ForeignKey(

        verbose_name=_('Resource'),

        to=ToulouseSmartResource,

        on_delete=models.CASCADE,

        related_name='cache_entries',

    )

    key = models.CharField(_('Key'), max_length=64)

    timestamp = models.DateTimeField(_('Timestamp'), auto_now=True)

    value = JSONField(_('Value'), default=dict)

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

{% load i18n passerelle %}

{% block extra-sections %}

    <div class="section">

        <h3>{% trans "Details" %}</h3>

        <ul>

            <li><a href="{% url "toulouse-smart-type-intervention" slug=object.slug %}">{% trans "Intervention types" %}</a></li>

        </ul>

    </div>

{% endblock %}

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

{% load i18n gadjo %}

{% block breadcrumb %}

{{ block.super }}

<a href="{{ toulousesmartresource.get_absolute_url }}">{{ toulousesmartresource.title }}</a>

<a href="#">{% trans 'Intervention types' %}</a>

{% endblock %}

{% block appbar %}

<h2>{% trans 'Intervention types' %}</h2>

<span class="actions">

    <a href="{% url "toulouse-smart-type-intervention-as-blocks" slug=toulousesmartresource.slug %}">{% trans "Export to blocks" %}</a>

</span>

{% endblock %}

{% block content %}

<table class="main">

    <thead>

        <tr>

            <th>Nom du type d'intervention</th>

            <th>Nom</th>

            <th>Type</th>

            <th>Requis</th>

            <th>Valeur par défaut</th>

        </tr>

    </thead>

    <tbody>

        {% for intervention_type in toulousesmartresource.get_intervention_types %}

        <tr><td colspan="4">{{ intervention_type.order }} - {{ intervention_type.name }}</td></tr>

            {% for property in intervention_type.properties %}

                <tr>

                    <td></td>

                    <td>{{ property.name }}</td>

                    <td>{{ property.type }}</td>

                    <td title="{{ property.required|lower }}">{{ property.required|yesno:"✔,✘" }}</td>

                    <td>{{ property.defaultValue }}</td>

                </tr>

            {% endfor %}

        {% endfor %}

    </tbody>

</table>

<!--

{{ toulousesmartresource.get_intervention_types|pprint }}

-->

{% endblock %}

<?xml version="1.0"?>

<block id="{{ id }}">

  <name>{{ name }}</name>

  <slug>{{ name|slugify }}</slug>

  <fields>

    {% for property in properties %}

    <field>

      <id>{{ property.id }}</id>

      <label>{{ property.name }}</label>

      <type>{{ property.type }}</type>

      <required>{{ property.required }}</required>

      <varname>{{ property.name|slugify }}</varname>

      <display_locations>

        <display_location>validation</display_location>

        <display_location>summary</display_location>

      </display_locations>{% if property.validation %}

      <validation>

        <type>{{ property.validation }}</type>

      </validation>{% endif %}

    </field>{% endfor %}

  </fields>

</block>

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2021 Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from django.conf.urls import url

from .views import TypeIntervention, TypeInterventionAsBlocks

management_urlpatterns = [

    url(

        r'^(?P<slug>[\w,-]+)/type-intervention/as-blocks/$',

        TypeInterventionAsBlocks.as_view(),

        name='toulouse-smart-type-intervention-as-blocks',

    ),

    url(

        r'^(?P<slug>[\w,-]+)/type-intervention/$',

        TypeIntervention.as_view(),

        name='toulouse-smart-type-intervention',

    ),

]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2021 Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import hashlib

import uuid

import zipfile

from django.http import HttpResponse

from django.template.loader import render_to_string

from django.utils.text import slugify

from django.views.generic import DetailView

from .models import ToulouseSmartResource

class TypeIntervention(DetailView):

    model = ToulouseSmartResource

    template_name = 'toulouse_smart/type-intervention.html'

class TypeInterventionAsBlocks(DetailView):

    model = ToulouseSmartResource

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

        self.object = self.get_object()

        def make_id(s):

            return str(uuid.UUID(bytes=hashlib.md5(s.encode()).digest()))

        # generate file contents

        files = {}

        for intervention_type in self.object.get_intervention_types():

            slug = slugify(intervention_type['name'])

            # only export intervention_type with properties

            if not intervention_type.get('properties'):

                continue

            for prop in intervention_type['properties']:

                # generate a natural id for fields

                prop['id'] = make_id(slug + slugify(prop['name']))

                # adapt types

                prop.setdefault('type', 'string')

                if prop['type'] == 'boolean':

                    prop['type'] = 'bool'

                if prop['type'] == 'int':

                    prop['type'] = 'string'

                    prop['validation'] = 'digits'

            filename = 'block-%s.wcs' % slug

            files[filename] = render_to_string('toulouse_smart/wcs_block.wcs', context=intervention_type)

        # zip it !

        response = HttpResponse(content_type='application/zip')

        response['Content-Disposition'] = 'attachment; filename=blocks.zip'

        with zipfile.ZipFile(response, mode='w') as zip_file:

            for name in files:

                zip_file.writestr(name, files[name])

        return response

    'passerelle.contrib.toulouse_smart',

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2021 Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import functools

import io

import zipfile

import lxml.etree as ET

import httmock

import pytest

import utils

from passerelle.contrib.toulouse_smart.models import ToulouseSmartResource

from passerelle.utils.xml import to_json

from test_manager import login

@pytest.fixture

def smart(db):

    return utils.make_resource(

        ToulouseSmartResource,

        title='Test',

        slug='test',

        description='Test',

        webservice_base_url='https://smart.example.com/',

        basic_auth_username='username',

        basic_auth_password='password',

    )

def mock_response(*path_contents):

    def decorator(func):

        @httmock.urlmatch()

        def error(url, request):

            assert False, 'request to %s' % (url,)

        @functools.wraps(func)

        def wrapper(*args, **kwargs):

            handlers = []

            for row in path_contents:

                path, content = row

                @httmock.urlmatch(path=path)

                def handler(url, request):

                    return content

                handlers.append(handler)

            handlers.append(error)

            with httmock.HTTMock(*handlers):

                return func(*args, **kwargs)

        return wrapper

    return decorator

@mock_response(['/v1/type-intervention', b'<List></List>'])

def test_empty_intervention_types(smart):

    assert smart.get_intervention_types() == []

INTERVENTION_TYPES = b'''<List>

   <item>

       <id>1234</id>

       <name>coin</name>

       <properties>

           <properties>

              <name>FIELD1</name>

              <type>string</type>

              <required>false</required>

           </properties>

           <properties>

              <name>FIELD2</name>

              <type>int</type>

              <required>true</required>

           </properties>

       </properties>

   </item>

</List>'''

@mock_response(['/v1/type-intervention', INTERVENTION_TYPES])

def test_model_intervention_types(smart):

    assert smart.get_intervention_types() == [

        {

            'id': '1234',

            'name': 'coin',

            'order': 1,

            'properties': [

                {'name': 'FIELD1', 'required': False, 'type': 'string'},

                {'name': 'FIELD2', 'required': True, 'type': 'int'},

            ],

        },

    ]

URL = '/toulouse-smart/test/'

@mock_response(['/v1/type-intervention', INTERVENTION_TYPES])

def test_endpoint_intervention_types(app, smart):

    resp = app.get(URL + 'type-intervention')

    assert resp.json == {'data': [{'id': '1234', 'text': 'coin'}], 'err': 0}

@mock_response()

def test_endpoint_intervention_types_unavailable(app, smart):

    resp = app.get(URL + 'type-intervention')

    assert resp.json == {'data': [{'id': '', 'text': 'Service is unavailable', 'disabled': True}], 'err': 0}

@mock_response(['/v1/type-intervention', INTERVENTION_TYPES])

def test_manage_intervention_types(app, smart, admin_user):

    login(app)

    resp = app.get('/manage' + URL + 'type-intervention/')

    assert [[td.text for td in tr.cssselect('td,th')] for tr in resp.pyquery('tr')] == [

        ["Nom du type d'intervention", 'Nom', 'Type', 'Requis', 'Valeur par défaut'],

        ['1 - coin'],

        [None, 'FIELD1', 'string', '✘', None],

        [None, 'FIELD2', 'int', '✔', None],

    ]

    resp = resp.click('Export to blocks')

    with zipfile.ZipFile(io.BytesIO(resp.body)) as zip_file:

        assert zip_file.namelist() == ['block-coin.wcs']

        with zip_file.open('block-coin.wcs') as fd:

            content = ET.tostring(ET.fromstring(fd.read()), pretty_print=True).decode()

            assert (

                content

                == '''<block id="1234">

  <name>coin</name>

  <slug>coin</slug>

  <fields>

    <field>

      <id>038a8c2e-14de-4d4f-752f-496eb7fe90d7</id>

      <label>FIELD1</label>

      <type>string</type>

      <required>False</required>

      <varname>field1</varname>

      <display_locations>

        <display_location>validation</display_location>

        <display_location>summary</display_location>

      </display_locations>

    </field>

    <field>

      <id>e72f251a-5eef-5b78-c35a-94b549510029</id>

      <label>FIELD2</label>

      <type>string</type>

      <required>True</required>

      <varname>field2</varname>

      <display_locations>

        <display_location>validation</display_location>

        <display_location>summary</display_location>

      </display_locations>

      <validation>

        <type>digits</type>

      </validation>

    </field>

  </fields>

</block>

'''

            )