passerelle/tests/test_lille_kimoce.py

346 lines
12 KiB
Python

# 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
from unittest import mock
import pytest
from django.urls import reverse
import tests.utils
from passerelle.contrib.lille_kimoce.models import Kimoce
from passerelle.utils.jsonresponse import APIError
@pytest.fixture
def setup(db):
return tests.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 = tests.utils.FakedResponse(content=TOKEN_ERROR_RESPONSE, status_code=400)
setup.get_token()
mocked_post.return_value = tests.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 = tests.utils.FakedResponse(content=TOKEN_RESPONSE, status_code=200)
mocked_get.return_value = tests.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 = tests.utils.FakedResponse(content=TOKEN_RESPONSE, status_code=200)
mocked_get.return_value = tests.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 = tests.utils.FakedResponse(content=TOKEN_RESPONSE, status_code=200)
mocked_get.return_value = tests.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 = tests.utils.FakedResponse(content=TOKEN_RESPONSE, status_code=200)
mocked_get.return_value = tests.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 = tests.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 = [
tests.utils.FakedResponse(content=TOKEN_RESPONSE, status_code=200),
tests.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': 'Bar'},
}
assert response.json['data'] == json.loads(DEMAND_CREATION_RESPONSE)
assert mocked_post.call_count == 2
data['picture1'] = {'content': 'base64encoded_picture'}
mocked_post.return_value = tests.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 'comment' not in mocked_post.call_args[1]['json']
data['lat'] = '48.85438994604021'
data['lon'] = '2.3497223854064946'
response = app.post_json(endpoint, params=data)
assert mocked_post.call_count == 4
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'
data['comment'] = 'my comment'
response = app.post_json(endpoint, params=data)
assert mocked_post.call_args[1]['json']['comment']['content'] == 'my comment'