0001-lille_kimoce-initial-connector-33099.patch

Serghei Mihai, 20 mai 2019 19:30

Voir les différences: en ligne côte à côte

Subject: [PATCH] lille_kimoce: initial connector (#33099)

 passerelle/contrib/lille_kimoce/__init__.py   |   0
 .../lille_kimoce/migrations/0001_initial.py   |  33 ++
 .../lille_kimoce/migrations/__init__.py       |   0
 passerelle/contrib/lille_kimoce/models.py     | 238 ++++++++++++
 passerelle/utils/http_authenticators.py       |  15 +
 tests/settings.py                             |   1 +
 tests/test_lille_kimoce.py                    | 344 ++++++++++++++++++
 7 files changed, 631 insertions(+)
 create mode 100644 passerelle/contrib/lille_kimoce/__init__.py
 create mode 100644 passerelle/contrib/lille_kimoce/migrations/0001_initial.py
 create mode 100644 passerelle/contrib/lille_kimoce/migrations/__init__.py
 create mode 100644 passerelle/contrib/lille_kimoce/models.py
 create mode 100644 tests/test_lille_kimoce.py

     # -*- coding: utf-8 -*-
     # Generated by Django 1.11.20 on 2019-05-13 08:21
     from __future__ import unicode_literals
     from django.db import migrations, models
     class Migration(migrations.Migration):
         initial = True
         dependencies = [
             ('base', '0012_job'),
+        ]
         operations = [
             migrations.CreateModel(
                 name='Kimoce',
                 fields=[
                     ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                     ('title', models.CharField(max_length=50, verbose_name='Title')),
                     ('description', models.TextField(verbose_name='Description')),
                     ('slug', models.SlugField(unique=True, verbose_name='Identifier')),
                     ('base_url', models.URLField(help_text='API base URL', max_length=256, verbose_name='Base URL')),
                     ('username', models.CharField(max_length=128, verbose_name='Username')),
                     ('password', models.CharField(max_length=128, verbose_name='Password')),
                     ('users', models.ManyToManyField(blank=True, to='base.ApiUser')),
                 ],
                 options={
                     'verbose_name': 'Lille Kimoce',
                 },
             ),
+        ]

     # -*- coding: utf-8 -*-
     # Copyright (C) 2019  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.db import models
     from django.utils.translation import ugettext_lazy as _
     from django.utils.six.moves.urllib_parse import urljoin
     from django.core.cache import cache
     from passerelle.base.models import BaseResource
     from passerelle.utils.api import endpoint
     from passerelle.utils.http_authenticators import HttpBearerAuth
     from passerelle.utils.jsonresponse import APIError
     DEMAND_SCHEMA = {
         '$schema': 'http://json-schema.org/draft-03/schema#',
         'title': 'KIMOCE',
         'description': '',
         'type': 'object',
         'properties': {
             'category': {
                 'description': 'demand category',
                 'type': 'string',
                 'required': True
             },
             'type': {
                 'description': 'demand type',
                 'type': 'string',
                 'required': True
             },
             'subtype': {
                 'description': 'demand sub type',
                 'type': 'string',
                 'required': True
             },
             'form_url': {
                 'description': 'form url',
                 'type': 'string',
                 'required': True
             },
             'first_name': {
                 'description': 'user first name',
                 'type': 'string',
                 'required': True
             },
             'last_name': {
                 'description': 'user last name',
                 'type': 'string',
                 'required': True
             },
             'email': {
                 'description': 'user email',
                 'type': 'string',
                 'required': True
             },
             'priorityId': {
                 'description': 'demand priority',
                 'type': 'integer',
                 'required': False
             },
             'city': {
                 'description': 'demand city',
                 'type': 'string',
                 'required': False
             },
             'zipcode': {
                 'description': 'demand zipcode',
                 'type': 'string',
                 'required': False
             },
             'street_number': {
                 'description': 'demand street number',
                 'type': 'string',
                 'required': False
             },
             'street_name': {
                 'description': 'demand street name',
                 'type': 'string',
                 'required': False
             },
             'lat': {
                 'description': 'demand latitude',
                 'type': 'string',
                 'required': False
             },
             'lon': {
                 'description': 'demand longitude',
                 'type': 'string',
                 'required': False
             },
             'picture1': {
                 'description': 'first picture base64 content',
                 'type': 'any',
                 'required': False
             },
             'picture2': {
                 'description': 'second picture base64 content',
                 'type': 'any',
                 'required': False
+            }
+        }
+    }
     class Kimoce(BaseResource):
         base_url = models.URLField(max_length=256, blank=False,
                                    verbose_name=_('Base URL'),
                                    help_text=_('API base URL'))
         username = models.CharField(max_length=128, verbose_name=_('Username'))
         password = models.CharField(max_length=128, verbose_name=_('Password'))
         category = _('Business Process Connectors')
         class Meta:
             verbose_name = 'Lille Kimoce'
         @classmethod
         def get_verbose_name(cls):
             return cls._meta.verbose_name
         def check_status(self):
             url = urljoin(self.base_url, 'login_check')
             response = self.requests.post(url, json={'username': self.username,
                                                      'password': self.password})
             response.raise_for_status()
         def get_token(self, renew=False):
             token_key = 'lille-kimoce-%s-token' % self.id
             if not renew and cache.get(token_key):
                 return cache.get(token_key)
             url = urljoin(self.base_url, 'login_check')
             response = self.requests.post(url, json={'username': self.username,
                                                      'password': self.password})
             if not response.ok:
                 raise APIError(response.content)
             token = response.json()['token']
             cache.set(token_key, token, 3600)
             return token
         def get_referential(self, endpoint, params=None):
             url = urljoin(self.base_url, endpoint)
             data = []
             response = self.requests.get(url, params=params,
                                 auth=HttpBearerAuth(self.get_token()))
             if response.status_code == 401:
                 response = self.requests.get(url, params=params,
                                     auth=HttpBearerAuth(self.get_token(renew=True)))
             if response.ok:
                 for member in response.json()['hydra:member']:
                     data.append({'id': member['@id'],
                                  'text': member['label'],
                                  'number': member['id']})
             return {'data': data}
         @endpoint(perm='can_access', description=_('List categories'))
         def categories(self, request, *args, **kwargs):
             return self.get_referential('categories')
         @endpoint(perm='can_access', description=_('List types'))
         def types(self, request, category_id):
             return self.get_referential('types', {'category.id': category_id})
         @endpoint(perm='can_access', description=_('List subtypes'))
         def subtypes(self, request, type_id):
             return self.get_referential('sub_types', {'types.id': type_id})
         @endpoint(perm='can_access', description=_('List streets'))
         def streets(self, request, **kwargs):
             url = urljoin(self.base_url, 'company_locations')
             data = []
             params = {'road': 'true'}
             if 'id' in kwargs:
                 params['streetAddress'] = kwargs['id']
             if 'q' in kwargs:
                 params['streetAddress'] = kwargs['q']
             # parentId is a flag to filter street names only
             response = self.requests.get(url, params=params, auth=HttpBearerAuth(self.get_token()))
             if response.status_code == 401:
                 response = self.requests.get(url, params={'streetAddress': q, 'parentId': 0},
                                          auth=HttpBearerAuth(self.get_token(renew=True)))
             if response.ok:
                 for member in response.json()['hydra:member']:
                     data.append({'id': member['@id'],
                                  'text': member['streetAddress']})
             return {'data': data}
         @endpoint(perm='can_access', description=_('Create demand'), post={
             'description': _('Create demand into KIMOCE'),
             'request_body': {
                 'schema': {
                     'application/json': DEMAND_SCHEMA
+                }
+            }
         })
         def create_demand(self, request, post_data):
             payload = {'category': post_data['category'],
                        'type': post_data['type'],
                        'subType': post_data['subtype'],
                        'priorityId': post_data.get('priorityId', 3),
                        'companyLocation': {
                            'number': post_data.get('street_number', ''),
                            'road': post_data.get('street_name', ''),
                            'city': post_data.get('city', ''),
                            'zipCode': post_data.get('zipcode', ''),
                        },
                        'sourceContact': {'firstname': post_data['first_name'],
                                          'lastname': post_data['last_name'],
                                          'mail': post_data['email']},
                        'pictures': [],
                        'GRUResponseLink': post_data['form_url']
+            }
             if post_data.get('lat') and post_data.get('lon'):
                 payload['coordinate'] = {'latitude': post_data['lat'],
                                          'longitude': post_data['lon']}
             for param_name in ('picture1', 'picture2'):
                 if post_data.get(param_name):
                     payload['pictures'].append({'content': post_data[param_name]['content']})
             url = urljoin(self.base_url, 'demands')
             result = self.requests.post(url, json=payload, auth=HttpBearerAuth(self.get_token()))
             if result.status_code  == 401:
                 result = self.requests.post(url, json=payload, auth=HttpBearerAuth(self.get_token(renew=True)))
             if result.ok:
                 return {'data': result.json()}
             raise APIError(result.content)

         def __call__(self, r):
             r.headers['Authorization'] = self.get_authorization_header(r)
             return r
     class HttpBearerAuth(AuthBase):
         def __init__(self, token):
             self.token = token
         def __eq__(self, other):
             return self.token == getattr(other, 'token', None)
         def __ne__(self, other):
             return not self == other
         def __call__(self, r):
             r.headers['Authorization'] = 'Bearer ' + self.token
             return r

tests/settings.py
38	38	'passerelle.contrib.teamnet_axel',
39	39	'passerelle.contrib.tlmcom',
40	40	'passerelle.contrib.tcl',
	41	'passerelle.contrib.lille_kimoce',
41	42	)
42	43
43	44	# enable applications that are otherwise disabled

     # -*- coding: utf-8 -*-
     # Passerelle - uniform access to data and services
     # Copyright (C) 2019  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; exclude 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.deepcopy 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 django.core.urlresolvers import reverse
     from passerelle.contrib.lille_kimoce.models import Kimoce
     from passerelle.utils.jsonresponse import APIError
     @pytest.fixture
     def setup(db):
         return utils.setup_access_rights(
             Kimoce.objects.create(slug='test',
                                   base_url='https://kimoce.mairie-lille.fr/api/',
                                   username='test', password='secret'))
     CATEGORIES_RESPONSE = """{
       "@context": "/api/contexts/Category",
       "@id": "/api/categories",
       "@type": "hydra:Collection",
       "hydra:member": [
         {"@id": "/api/categories/2",
          "id": 2,
          "label": "Accessoire répertorié",
          "reference": "ARE"
         },
         {"@id": "/api/categories/20",
          "id": 20,
          "label": "Achats",
          "reference": "PRE"
         },
         {"@id": "/api/categories/70",
          "id": 70,
          "label": "Table surface",
          "reference": "TBS"
         },
         {"@id": "/api/categories/73",
          "id": 73,
          "label": "Tableau Blanc Interactif",
          "reference": "TBI"
         },
         {"@id": "/api/categories/80",
          "id": 80,
          "label": "Propreté-Déchets-Tags",
          "reference": "W05"
         },
         {"@id": "/api/categories/85",
          "id": 85,
          "label": "Gestion de conflits en temps différ",
          "reference": "MD2"
+        }
       ],
       "hydra:totalItems": 6
     }"""
     TYPES_RESPONSE = """{
       "@context": "/api/contexts/Type",
       "@id": "/api/types",
       "@type": "hydra:Collection",
       "hydra:member": [
         {"@id": "/api/types/1825",
          "@type": "Type",
          "id": 1825,
          "label": "Tags"
         },
         {"@id": "/api/types/1826",
          "@type": "Type",
          "id": 1826,
          "label": "Dépôt sauvage"
         },
         {"@id": "/api/types/1827",
          "@type": "Type",
          "id": 1827,
          "label": "Poubelle, Corbeille publique de rue"
         },
         {"@id": "/api/types/1828",
          "@type": "Type",
          "id": 1828,
          "label": "Affichage sauvage"
         },
         {"@id": "/api/types/1829",
          "@type": "Type",
          "id": 1829,
          "label": "Rue,Espace sales,Déchets éparpillés"
         },
         {"@id": "/api/types/1830",
          "@type": "Type",
          "id": 1830,
          "label": "Poubelle de particuliers"
+        }
       ],
       "hydra:totalItems": 5
     }"""
     SUB_TYPES_RESPONSE = """{
       "@context": "/api/contexts/SubType",
       "@id": "/api/sub_types",
       "@type": "hydra:Collection",
       "hydra:member": [
         {"@id": "/api/sub_types/163",
          "@type": "SubType",
          "id": 163,
          "label": "Tag classique"
         },
         {"@id": "/api/sub_types/164",
          "@type": "SubType",
          "id": 164,
          "label": "Tag injurieux, tendancieux"
+        }
       ],
       "hydra:totalItems": 2
     }"""
     STREETS_RESPONSE = """{
       "@context": "/api/contexts/CompanyLocation",
       "@id": "/api/company_locations",
       "@type": "hydra:Collection",
       "hydra:member": [
         {"@id": "/api/company_locations/4368",
          "@type": "CompanyLocation",
          "streetAddress": "PLACE JOSEPH HENTGES, H"
         },
         {"@id": "/api/company_locations/7550",
          "@type": "CompanyLocation",
          "streetAddress": "9 PLACE JOSEPH HENTGES"
         },
         {"@id": "/api/company_locations/7551",
          "@type": "CompanyLocation",
          "streetAddress": "5 PLACE JOSEPH HENTGES"
+        }
       ],
       "hydra:totalItems": 3
     }"""
     DEMAND_CREATION_RESPONSE = """{
         "@context": "/api/contexts/Demand",
         "@id": "/api/demands/166961",
         "@type": "Demand",
         "category": "/api/categories/33",
         "coordinate": {
             "latitude": 3.50895,
             "longitude": 50.340892
         },
         "pictures": [],
         "priorityId": 3,
         "subType": "/api/sub_types/17",
         "type": "/api/types/916",
         "sourceContact": {
             "firstname": "Foo",
             "lastname": "Bar",
             "mail": "foo@example.net"
         },
         "companyLocation": {
             "@id": "/api/company_locations/3656",
             "@type": "CompanyLocation",
             "city": "lille",
             "number": "55 bis",
             "road": "squares du portugal",
             "zipCode": "59000"
+        }
     }"""
     TOKEN_RESPONSE = """{"token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9"}"""
     TOKEN_ERROR_RESPONSE = """{"message": "Bad credentials"}"""
     @mock.patch('passerelle.utils.Request.post')
     def test_get_token(mocked_post, app, setup):
         with pytest.raises(APIError):
             mocked_post.return_value = utils.FakedResponse(content=TOKEN_ERROR_RESPONSE, ok=False)
             setup.get_token()
         mocked_post.return_value = utils.FakedResponse(content=TOKEN_RESPONSE, ok=True)
         setup.get_token()
         assert mocked_post.call_count == 2
         print mocked_post.call_args
         assert "api/login_check" in mocked_post.call_args[0][0]
         assert mocked_post.call_args[1]['json']['username'] == 'test'
         assert mocked_post.call_args[1]['json']['password'] == 'secret'
         # make sure the token from cache is used
         setup.get_token()
         assert mocked_post.call_count == 2
         setup.get_token(True)
         assert mocked_post.call_count == 3
     @mock.patch('passerelle.utils.Request.post')
     @mock.patch('passerelle.utils.Request.get')
     def test_get_categories(mocked_get, mocked_post, app, setup):
         mocked_post.return_value = utils.FakedResponse(content=TOKEN_RESPONSE, ok=True)
         mocked_get.return_value = utils.FakedResponse(ok=True, content=CATEGORIES_RESPONSE)
         endpoint = reverse(
             'generic-endpoint',
             kwargs={
                 'connector': 'lille-kimoce',
                 'slug': setup.slug,
                 'endpoint': 'categories'
+            }
+        )
         response = app.get(endpoint)
         assert 'data' in response.json
         for item in response.json['data']:
             assert 'id' in item
             assert 'text' in item
             assert 'number' in item
     @mock.patch('passerelle.utils.Request.post')
     @mock.patch('passerelle.utils.Request.get')
     def test_get_types(mocked_get, mocked_post, app, setup):
         mocked_post.return_value = utils.FakedResponse(content=TOKEN_RESPONSE, ok=True)
         mocked_get.return_value = utils.FakedResponse(ok=True, content=TYPES_RESPONSE)
         endpoint = reverse('generic-endpoint',
             kwargs={'connector': 'lille-kimoce',
                     'slug': setup.slug,
                     'endpoint': 'types'
+            }
+        )
         response = app.get(endpoint, status=400)
         assert response.json['err']
         assert response.json['err_desc'] == 'missing parameters: \'category_id\'.'
         assert response.json['err_class'] == 'passerelle.views.WrongParameter'
         response = app.get(endpoint, params={'category_id': 80})
         assert 'data' in response.json
         for item in response.json['data']:
             assert 'id' in item
             assert 'text' in item
             assert 'number' in item
     @mock.patch('passerelle.utils.Request.post')
     @mock.patch('passerelle.utils.Request.get')
     def test_get_sub_types(mocked_get, mocked_post, app, setup):
         mocked_post.return_value = utils.FakedResponse(content=TOKEN_RESPONSE, ok=True)
         mocked_get.return_value = utils.FakedResponse(ok=True, content=SUB_TYPES_RESPONSE)
         endpoint = reverse('generic-endpoint',
             kwargs={'connector': 'lille-kimoce',
                     'slug': setup.slug,
                     'endpoint': 'subtypes'
+            }
+        )
         response = app.get(endpoint, status=400)
         assert response.json['err']
         assert response.json['err_desc'] == 'missing parameters: \'type_id\'.'
         assert response.json['err_class'] == 'passerelle.views.WrongParameter'
         response = app.get(endpoint, params={'type_id': 1825})
         assert 'data' in response.json
         for item in response.json['data']:
             assert 'id' in item
             assert 'text' in item
             assert 'number' in item
     @mock.patch('passerelle.utils.Request.post')
     @mock.patch('passerelle.utils.Request.get')
     def test_get_streets(mocked_get, mocked_post, app, setup):
         mocked_post.return_value = utils.FakedResponse(content=TOKEN_RESPONSE, ok=True)
         mocked_get.return_value = utils.FakedResponse(ok=True, content=STREETS_RESPONSE)
         endpoint = reverse('generic-endpoint',
             kwargs={'connector': 'lille-kimoce',
                     'slug': setup.slug,
                     'endpoint': 'streets'
+            }
+        )
         response = app.get(endpoint)
         assert 'data' in response.json
         assert len(response.json['data']) == 3
         response = app.get(endpoint, params={'q': 'PLACE JosEPH'})
         print mocked_get.call_args[1]['params']
         assert mocked_get.call_args[1]['params']['streetAddress'] == 'PLACE JosEPH'
         response = app.get(endpoint, params={'id': 'PLACE JosEPH'})
         assert mocked_get.call_args[1]['params']['streetAddress'] == 'PLACE JosEPH'
         for item in response.json['data']:
             assert 'id' in item
             assert 'text' in item
     @mock.patch('passerelle.utils.Request.post')
     def test_create_demand(mocked_post, app, setup):
         mocked_post.side_effect = [
             utils.FakedResponse(content=TOKEN_RESPONSE, ok=True),
             utils.FakedResponse(ok=True, content=DEMAND_CREATION_RESPONSE)]
         data = {
             'category': '/api/categories/80',
             'type': '/api/types/1825',
             'subtype': '/api/sub_types/164',
             'form_url': 'http://example.com/form/1/',
             'first_name': 'Foo',
             'last_name': 'Bar',
             'email': 'foo@example.net',
+        }
         endpoint = reverse('generic-endpoint',
             kwargs={'connector': 'lille-kimoce',
                     'slug': setup.slug,
                     'endpoint': 'create_demand'
+            }
+        )
         response = app.post_json(endpoint, params=data)
         assert mocked_post.call_args[0][0] == 'https://kimoce.mairie-lille.fr/api/demands'
         assert mocked_post.call_args[1]['json'] == {'GRUResponseLink': 'http://example.com/form/1/',
                         'priorityId': 3, 'type': '/api/types/1825',
                                                    'category': '/api/categories/80',
                         'companyLocation': {'number': '',
                                     'road': '',
                                     'zipCode': '',
                                     'city': ''},
                         'pictures': [],
                         'subType': '/api/sub_types/164',
                         'sourceContact': {'firstname': 'Foo',
                                           'mail': 'foo@example.net',
                                           'lastname': u'Bar'}
+        }
         assert response.json['data'] == json.loads(DEMAND_CREATION_RESPONSE)
         assert mocked_post.call_count == 2
         data['picture1'] = {'content': 'base64encoded_picture'}
         data['lat'] = '48.85438994604021'
         data['lon'] = '2.3497223854064946'
         mocked_post.return_value = utils.FakedResponse(ok=True, content=DEMAND_CREATION_RESPONSE)
         mocked_post.side_effect = None
         response = app.post_json(endpoint, params=data)
         assert mocked_post.call_count == 3
         assert mocked_post.call_args[1]['json']['pictures'][0]['content'] == 'base64encoded_picture'
         assert mocked_post.call_args[1]['json']['coordinate']['latitude'] == '48.85438994604021'
         assert mocked_post.call_args[1]['json']['coordinate']['longitude'] == '2.3497223854064946'
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0001-lille_kimoce-initial-connector-33099.patch