entrouvert
/
passerelle
15
0
Fork 0
Code Issues Pull Requests 15 Releases Activity

lille_kimoce: initial connector (#33099)

This commit is contained in:
Serghei Mihai 2019-05-14 09:47:44 +02:00
parent a7e7814002
commit 1cb079e4ba
6 changed files with 628 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
passerelle/contrib/lille_kimoce/__init__.py Normal file
View File

33
passerelle/contrib/lille_kimoce/migrations/0001_initial.py Normal file
View File

@ -0,0 +1,33 @@
# -*- 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',
},
),
]

0
passerelle/contrib/lille_kimoce/migrations/__init__.py Normal file
View File

241
passerelle/contrib/lille_kimoce/models.py Normal file
View File

@ -0,0 +1,241 @@
# -*- 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 data',
'type': 'any',
'required': False
},
'picture2': {
'description': 'second picture data',
'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.status_code // 100 == 2:
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.status_code // 100 == 2:
for member in response.json()['hydra:member']:
member['number'] = member['id']
member['text'] = member['label']
member['id'] = member['@id']
data.append(member)
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=params,
auth=HttpBearerAuth(self.get_token(renew=True)))
if response.status_code // 100 == 2:
for street in response.json()['hydra:member']:
street['number'] = street['@id']
street['id'] = street['@id']
street['text'] = street['streetAddress']
data.append(street)
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) and isinstance(post_data[param_name], dict) and post_data[param_name].get('content'):
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.status_code // 100 == 2:
return {'data': result.json()}
raise APIError(result.content)

1
tests/settings.py
View File

@ -38,6 +38,7 @@ INSTALLED_APPS += (
'passerelle.contrib.teamnet_axel',
'passerelle.contrib.tlmcom',
'passerelle.contrib.tcl',
'passerelle.contrib.lille_kimoce',
)
# enable applications that are otherwise disabled

353
tests/test_lille_kimoce.py Normal file
View File

@ -0,0 +1,353 @@
# -*- 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, status_code=400)
setup.get_token()
mocked_post.return_value = utils.FakedResponse(content=TOKEN_RESPONSE, status_code=200)
setup.get_token()
assert mocked_post.call_count == 2
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, status_code=200)
mocked_get.return_value = utils.FakedResponse(content=CATEGORIES_RESPONSE, status_code=200)
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
assert '@id' in item
assert 'label' in item
assert 'reference' 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, status_code=200)
mocked_get.return_value = utils.FakedResponse(content=TYPES_RESPONSE, status_code=200)
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
assert '@type' in item
assert 'label' 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, status_code=200)
mocked_get.return_value = utils.FakedResponse(content=SUB_TYPES_RESPONSE, status_code=200)
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
assert '@type' in item
assert 'label' 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, status_code=200)
mocked_get.return_value = utils.FakedResponse(content=STREETS_RESPONSE, status_code=200)
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'})
assert mocked_get.call_args[1]['params']['streetAddress'] == 'PLACE JosEPH'
mocked_get.return_value = utils.FakedResponse(content=STREETS_RESPONSE, status_code=401)
response = app.get(endpoint, params={'id': 'RUE de PAris'})
assert mocked_get.call_args[1]['params']['streetAddress'] == 'RUE de PAris'
for item in response.json['data']:
assert 'id' in item
assert 'text' in item
assert '@id' in item
assert '@type' in item
assert 'streetAddress' 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, status_code=200),
utils.FakedResponse(content=DEMAND_CREATION_RESPONSE, status_code=200)]
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(content=DEMAND_CREATION_RESPONSE, status_code=200)
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'