Produits Entr'ouvert » Passerelle

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

# Generated by Django 1.11.18 on 2020-10-19 13:26

from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0022_auto_20200715_1033'),

    ]

    operations = [

        migrations.CreateModel(

            name='Sigerly',

            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(default=True, verbose_name='TLS verify certificates')),

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

                ('base_url', models.CharField(help_text='example: https://sig.sigerly.fr/syecl_intervention/webservicev2/', max_length=256, verbose_name='Service URL')),

                ('users', models.ManyToManyField(blank=True, related_name='_sigerly_users_+', related_query_name='+', to='base.ApiUser')),

            ],

            options={

                'verbose_name': 'Sigerly',

            },

        ),

    ]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2020  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 urllib.parse import urljoin

from django.db import models

from django.utils.translation import ugettext_lazy as _

from passerelle.base.models import BaseResource, HTTPResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

CREATE_SCHEMA = {

    '$schema': 'http://json-schema.org/draft-04/schema#',

    "type": "object",

    'required': ['demandeur', 'id_typeinterv', 'id_urgence', 'elements'],

    'properties': {

        'demandeur': {

            'description': "Nom du demandeur",

            'type': 'string',

        },

        'id_typeinterv': {

            'description': "Type de l'intervention",

            'type': 'string',

        },

        'id_urgence': {

            'description': 'Urgence',

            'type': 'string',

        },

        'id_qualification': {

            'description': 'Qualification',

            'type': 'string',

        },

        'observations': {

            'description': 'Observations',

            'type': 'string',

        },

        'elements': {

            'description': "Identifiant de l'objet : liste séparée par ':'",

            'type': 'string',

            'pattern': r'^[0-9A-Z :]+$',

        },

    },

}

QUERY_SCHEMA = {

    '$schema': 'http://json-schema.org/draft-04/schema#',

    "type": "object",

    'properties': {

        'id_intervention': {

            'description': 'Rechercher une intervention par son numéro'

            ' (non cumulable avec les autres filtres)',

            'type': 'string',

        },

        'date_debut_demande': {

            'description': 'Recherche toutes les interventions dont la date de demande'

            ' est supérieure ou égale à la date renseignée (YYYY-MM-DD)',

            'type': 'string',

        },

        'date_fin_demande': {

            'description': 'Recherche toutes les interventions dont la date de demande'

            ' est inférieure ou égale à la date renseignée (YYYY-MM-DD)',

            'type': 'string',

        },

        'insee': {

            'description': "Code insee de la commune : liste séparée par ':'",

            'type': 'string',

            'pattern': r'^[0-9A-Z :]+$',

        },

    },

}

class Sigerly(BaseResource, HTTPResource):

    base_url = models.CharField(

        max_length=256,

        blank=False,

        verbose_name='URL du webservice',

        help_text=_('exemple: https://sig.sigerly.fr/syecl_intervention_preprod/webservicev2/'),

    )

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = 'Sigerly'

    def request(self, uri, json):

        url = urljoin(self.base_url, uri)

        headers = {'Accept': 'application/json'}

        response = self.requests.post(url, json=json, headers=headers)

        if response.status_code != 200:

            try:

                json_content = response.json()

            except ValueError:

                json_content = None

            raise APIError(

                'error status:%s %r, content:%r'

                % (response.status_code, response.reason, response.content[:1024]),

                data={'status_code': response.status_code, 'json_content': json_content},

            )

        try:

            json_response = response.json()

        except ValueError:

            raise APIError('invalid JSON content:%r' % response.content[:1024])

        return json_response

    @endpoint(

        perm='can_access',

        methods=['post'],

        description='Versement d’une demande d’éclairage public dans l’extranet du SIGERLY',

        post={'request_body': {'schema': {'application/json': CREATE_SCHEMA}}},

    )

    def create(self, request, post_data):

        post_data['id_typeinterv'] = int(post_data['id_typeinterv'])

        post_data['id_urgence'] = int(post_data['id_urgence'])

        post_data['id_qualification'] = int(post_data['id_qualification'])

        post_data['elements'] = [x.strip() for x in post_data['elements'].split(':') if x.strip()]

        response = self.request('createIntervention.php', json=post_data)

        if not response.get('success', None):

            raise APIError(response.get('message', None))

        if not response.get('message', None):

            raise APIError('No intervention id returned')

        return {'data': response}

    @endpoint(

        perm='can_access',

        methods=['post'],

        description='Interroger le SIGERLY sur l’etat d’avancement d’une demande',

        post={'request_body': {'schema': {'application/json': QUERY_SCHEMA}}},

    )

    def query(self, request, post_data):

        if post_data.get('id_intervention', None):

            post_data['id_intervention'] = int(post_data['id_intervention'])

            post_data.pop('date_debut_demande', None)

            post_data.pop('date_fin_demande', None)

            post_data.pop('insee', None)

        else:

            post_data.pop('id_intervention', None)

            if post_data.get('insee'):

                post_data['insee'] = [x.strip() for x in post_data['insee'].split(':') if x.strip()]

        response = self.request('getIntervention.php', json=post_data)

        for record in response:

            record['id'] = record.get('code_inter')

            record['text'] = '%(code_inter)s: %(libelle_intervention)s' % record

        return {'data': response}

[

   {

      "annule" : null,

      "code_inter" : "DP.20.116.1",

      "date_cr1" : null,

      "date_emission" : "2020-11-19",

      "date_inter_planif1" : null,

      "date_inter_rea_1" : null,

      "date_valid" : "2020-11-19",

      "date_valid_planif_1" : null,

      "date_validfact" : null,

      "elements" : [

         {

            "icon_gmap" : "luminaire.png",

            "id_interv" : 22014,

            "idelement" : 42282,

            "ident" : "LIMW003D",

            "libelle_objetgeo" : "Luminaire",

            "nom_rue" : "Voie d'accès parc des Sports",

            "the_geom" : "0101000020110F000038A23D18A5372041C775D7FE5BF35541",

            "travaux" : []

         },

         {

            "icon_gmap" : "luminaire.png",

            "id_interv" : 22014,

            "idelement" : 42283,

            "ident" : "LIMW003C",

            "libelle_objetgeo" : "Luminaire",

            "nom_rue" : "Voie d'accès parc des Sports",

            "the_geom" : "0101000020110F00006F56132FA53720417E4630FB5BF35541",

            "travaux" : []

         }

      ],

      "flag_mob" : null,

      "garantie" : null,

      "id_commune" : 29,

      "id_demande_web" : 10914,

      "id_entreprise" : 2,

      "id_entreprise_2" : null,

      "id_traitement" : null,

      "id_typeinterv" : 5,

      "id_urgence" : 1,

      "idinterv" : 22014,

      "libcause" : null,

      "libcommune" : "LIMONEST",

      "libelle_intervention" : "DEPANNAGE EP",

      "libelle_urgence" : "Normale",

      "libentreprise" : "EIFFAGE",

      "libentreprise2" : null,

      "libterritoire" : "NORD",

      "num_mandat" : null,

      "ref_technique" : null,

      "valid_entreprise" : null,

      "valid_sydev" : null

   }

]

[

   {

      "annule" : null,

      "code_inter" : "VN.20.291.11",

      "date_cr1" : null,

      "date_emission" : "2020-11-19",

      "date_inter_planif1" : null,

      "date_inter_rea_1" : null,

      "date_inter_rea_2" : null,

      "date_valid" : null,

      "date_valid_planif_1" : null,

      "date_valid_planif_2" : null,

      "date_validfact" : null,

      "elements" : [],

      "flag_mob" : null,

      "garantie" : null,

      "id_commune" : 6,

      "id_demande_web" : null,

      "id_entreprise" : 5,

      "id_entreprise_2" : 6,

      "id_suite" : null,

      "id_traitement" : 1,

      "id_typeinterv" : 4,

      "id_urgence" : null,

      "idinterv" : 25080,

      "inseecommune" : "069291",

      "libcause" : null,

      "libcommune" : "ST SYMPHORIEN D OZON",

      "libelle_intervention" : "VISITE DE NUIT",

      "libelle_urgence" : null,

      "libentreprise" : "SERPOLLET",

      "libentreprise2" : "INEO",

      "libterritoire" : "SUD",

      "mail" : "Mikael.BOBROSKY@serpollet.com",

      "num_mandat" : null,

      "ref_technique" : null,

      "toodego" : null,

      "valid_entreprise" : null,

      "valid_sydev" : null

   },

   {

      "annule" : null,

      "code_inter" : "DP.20.291.53",

      "date_cr1" : "2020-11-19",

      "date_emission" : "2020-11-19",

      "date_inter_planif1" : "2020-11-19",

      "date_inter_rea_1" : "2020-11-19",

      "date_inter_rea_2" : "2020-11-19",

      "date_valid" : "2020-11-19",

      "date_valid_planif_1" : "19/11/2020",

      "date_valid_planif_2" : "2020-11-19",

      "date_validfact" : null,

      "elements" : [

         {

            "icon_gmap" : "armoire.png",

            "id_interv" : 27948,

            "idelement" : 51681,

            "ident" : "AB",

            "libelle_objetgeo" : "Armoire",

            "nom_rue" : "Portes de Lyon (avenue des)",

            "the_geom" : "0101000020110F0000C7A520BC497F20415B9A1EFB2AD45541",

            "travaux" : []

         }

      ],

      "flag_mob" : null,

      "garantie" : null,

      "id_commune" : 6,

      "id_demande_web" : 15879,

      "id_entreprise" : 5,

      "id_entreprise_2" : null,

      "id_suite" : 2,

      "id_traitement" : 1,

      "id_typeinterv" : 5,

      "id_urgence" : null,

      "idinterv" : 27948,

      "inseecommune" : "069291",

      "libcause" : null,

      "libcommune" : "ST SYMPHORIEN D OZON",

      "libelle_intervention" : "DEPANNAGE EP",

      "libelle_urgence" : null,

      "libentreprise" : "SERPOLLET",

      "libentreprise2" : null,

      "libterritoire" : "SUD",

      "mail" : "Mikael.BOBROSKY@serpollet.com",

      "num_mandat" : null,

      "ref_technique" : null,

      "toodego" : null,

      "valid_entreprise" : 1,

      "valid_sydev" : 1

   }

]

    'passerelle.contrib.sigerly',

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2020  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 json

import os

import mock

import pytest

import utils

from passerelle.contrib.sigerly.models import Sigerly

from passerelle.utils.jsonresponse import APIError

@pytest.fixture

def connector(db):

    return utils.setup_access_rights(Sigerly.objects.create(slug='test', base_url='https://dummy-server.org'))

TEST_BASE_DIR = os.path.join(os.path.dirname(__file__), 'data', 'sigerly')

def json_get_data(filename):

    with open(os.path.join(TEST_BASE_DIR, "%s.json" % filename)) as fd:

        return json.dumps(json.load(fd))

def response(status_code, content):

    return utils.FakedResponse(content=content, status_code=status_code)

CREATE = response(

    200,

    json.dumps(

        {

            'success': True,

            'message': '7830',  # unrelated id

        }

    ),

)

CREATE_ERROR_1 = response(

    200,

    json.dumps(

        {

            'success': False,

            'message': 'XXX',  # error message

        }

    ),

)

CREATE_ERROR_2 = response(

    200,

    json.dumps(

        {

            'success': True,

            'message': '',

        }

    ),

)

QUERY_1 = response(200, json_get_data('getIntervention_1'))

QUERY_2 = response(200, json_get_data('getIntervention_2'))

def get_endpoint(name):

    return utils.generic_endpoint_url('sigerly', name)

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

@pytest.mark.parametrize(

    'status_code, json_content, a_dict',

    [

        (200, 'not json', None),

        (500, '{"message": "help"}', {'message': 'help'}),

        (500, 'not json', None),

    ],

)

def test_request_error(mocked_post, app, connector, status_code, json_content, a_dict):

    mocked_post.side_effect = [response(status_code, json_content)]

    with pytest.raises(APIError) as exc:

        connector.request('some-url', json=None)

    assert exc.value.err

    if status_code == 200:

        assert exc.value.http_status == 200

        assert exc.value.args[0] == "invalid JSON content:'%s'" % json_content

    else:

        assert exc.value.data['status_code'] == status_code

        assert exc.value.data['json_content'] == a_dict

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

def test_create(mocked_post, app, connector):

    mocked_post.side_effect = [CREATE]

    endpoint = get_endpoint('create')

    payload = {

        'demandeur': 'Test webservice',

        'id_typeinterv': '5',

        'id_urgence': '1',

        'id_qualification': '8',

        'observations': 'Test webservice',

        'elements': 'LIMW003D:LIMWW003C',

    }

    resp = app.post_json(endpoint, params=payload)

    assert mocked_post.mock_calls == [

        mock.call(

            'https://dummy-server.org/createIntervention.php',

            headers={'Accept': 'application/json'},

            json={

                'demandeur': 'Test webservice',

                'id_typeinterv': 5,

                'id_urgence': 1,

                'id_qualification': 8,

                'observations': 'Test webservice',

                'elements': ['LIMW003D', 'LIMWW003C'],

            },

        )

    ]

    assert resp.json == {'err': 0, 'data': json.loads(CREATE.content)}

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

@pytest.mark.parametrize(

    'response, desc',

    [

        (CREATE_ERROR_1, 'XXX'),

        (CREATE_ERROR_2, 'No intervention id returned'),

    ],

)

def test_create_error(mocked_post, app, connector, response, desc):

    mocked_post.side_effect = [response]

    endpoint = get_endpoint('create')

    payload = {

        'demandeur': 'Test webservice',

        'id_typeinterv': '5',

        'id_urgence': '1',

        'id_qualification': '8',

        'observations': 'Test webservice',

        'elements': 'LIMW003D:LIMWW003C',

    }

    resp = app.post_json(endpoint, params=payload)

    assert resp.json['err']

    assert resp.json['err_desc'] == desc

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

def test_query_id(mocked_post, app, connector):

    mocked_post.return_value = QUERY_1

    endpoint = get_endpoint('query')

    payload = {

        'id_intervention': '10914',

    }

    resp = app.post_json(endpoint, params=payload)

    assert mocked_post.mock_calls[0] == mock.call(

        'https://dummy-server.org/getIntervention.php',

        headers={'Accept': 'application/json'},

        json={'id_intervention': 10914},

    )

    assert not resp.json['err']

    assert len(resp.json['data']) == 1

    assert resp.json['data'][0]['id_demande_web'] == 10914

    assert resp.json['data'][0]['idinterv'] == 22014

    elements = resp.json['data'][0]['elements']

    assert len(elements) == 2

    assert [x['ident'] for x in elements] == ['LIMW003D', 'LIMW003C']

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

def test_query_filters(mocked_post, app, connector):

    mocked_post.return_value = QUERY_2

    endpoint = get_endpoint('query')

    payload = {'date_debut_demande': '19/11/2020', 'date_fin_demande': '19/11/2020', 'insee': '069291:069283'}

    resp = app.post_json(endpoint, params=payload)

    assert mocked_post.mock_calls[0] == mock.call(

        'https://dummy-server.org/getIntervention.php',

        headers={'Accept': 'application/json'},

        json={

            'date_debut_demande': '19/11/2020',

            'date_fin_demande': '19/11/2020',

            'insee': ['069291', '069283'],

        },

    )

    assert not resp.json['err']

    assert len(resp.json['data']) == 2

    assert [x['date_emission'] for x in resp.json['data']] == ['2020-11-19'] * 2

    assert [x['inseecommune'] for x in resp.json['data']] == ['069291'] * 2