passerelle/tests/test_grenoble_gru.py

378 lines
14 KiB
Python

# Passerelle - uniform access to data and services
# Copyright (C) 2017 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
from unittest import mock
import pytest
from django.urls import reverse
import tests.utils
from passerelle.contrib.grenoble_gru.models import GrenobleGRU
@pytest.fixture
def setup(db):
return tests.utils.setup_access_rights(
GrenobleGRU.objects.create(slug='test', base_url='https://itvvoirie-test.lametro.fr')
)
BASIC_PAYLOAD = {
'application_id': '12-1',
'applicant_lastname': 'Doe',
'applicant_firstname': 'John',
'applicant_phone': '0102030405',
'applicant_email': 'j.doe@example.org',
'applicant_required_reply': '0',
'applicant_contact_mode': '3',
'applicant_status': '8',
'intervention_address_type': '1',
'intervention_street_number': '168',
'intervention_street_name': 'Du Chateau',
'intervention_address_insee': '38185',
'intervention_sector': '2',
'intervention_number_type': '5',
'intervention_datetime': '2011-02-07T11:34',
'urgent_demand': '1',
'dysfonction_type': '3',
'intervention_reason': '2',
'dem_pav': 'déchetterie',
'comment_description': 'thank you 😘',
}
DEMAND = """<demande IdGRU="1324" idInterne="422">
<demandeur>
<nom>Mihai</nom>
<prenom>Serghei</prenom>
<tel>0102030405</tel>
<mail>smihai@entrouvert.com</mail>
<adresse>
<commune/>
<voie/>
<numero/>
<complement/>
<secteur/>
</adresse>
<reponsedemandee>OUI</reponsedemandee>
<modecontact>Contact direct</modecontact>
<nature>Usager</nature>
</demandeur>
<lieu>
<adresse>
<manuelle>16 rue du Château</manuelle>
</adresse>
</lieu>
<observation>
<dysfonctionnement>AUTRE</dysfonctionnement>
<motif/>
<description>Serghei fait des tests</description>
</observation>
<transmissions>
<transmission>
<service>Secteur Nord-Ouest</service>
<date>2018-01-03 18:58:01</date>
<motif>
Création de la demande via le webService de la commune de Grenoble
</motif>
</transmission>
</transmissions>
<intervention>
<intervenant/>
<miseajour>
<nom/>
<prenom/>
</miseajour>
<prevue/>
<realisee/>
<description id=""/>
<complement/>
</intervention>
<cloture>
<nom/>
<prenom/>
<date/>
<heure/>
</cloture>
<reponse>
<moyen/>
<date/>
</reponse>
</demande>"""
@pytest.fixture(params=['01', '02', '10', '20', '21', '23', '4242'])
def remote_response(request):
text = str(request.param)
return tests.utils.FakedResponse(index=request.param_index, text=text)
def test_contact_mode_typologies_list_with_invalid_xml(app, setup):
with mock.patch('passerelle.utils.Request.post') as request_post:
response = mock.Mock()
types_filename = os.path.join(os.path.dirname(__file__), 'data', 'grenoble_gru_typologies.xml')
with open(types_filename) as fd:
types = fd.read()
response.content = types.replace('Courrier', 'Courrier & autres')
request_post.return_value = response
endpoint = reverse(
'generic-endpoint',
kwargs={'connector': 'grenoble-gru', 'slug': setup.slug, 'endpoint': 'contact-modes'},
)
response = app.get(endpoint)
request_post.assert_called_once()
assert response.json['err']
assert 'Invalid XML returned:' in response.json['err_desc']
assert response.json['data'] is None
def test_contact_mode_typologies_list(app, setup):
with mock.patch('passerelle.utils.Request.post') as request_post:
response = mock.Mock()
types_filename = os.path.join(os.path.dirname(__file__), 'data', 'grenoble_gru_typologies.xml')
with open(types_filename) as fd:
response.content = fd.read()
request_post.return_value = response
endpoint = reverse(
'generic-endpoint',
kwargs={'connector': 'grenoble-gru', 'slug': setup.slug, 'endpoint': 'contact-modes'},
)
response = app.get(endpoint)
# check that data is in cache
response = app.get(endpoint)
assert request_post.call_count == 1
response = app.get(endpoint, params={'dem_comp': 'Voirie'})
assert request_post.call_count == 2
assert request_post.call_args[1]['data']['dem_comp'] == 'Voirie'
assert len(response.json['data']) == 6
for contact_mode in response.json['data']:
assert 'id' in contact_mode
assert 'text' in contact_mode
# check that data is in cache
response = app.get(endpoint, params={'dem_comp': 'Voirie'})
assert request_post.call_count == 2
def get_typo_response():
types_filename = os.path.join(os.path.dirname(__file__), 'data', 'grenoble_gru_typologies.xml')
with open(types_filename) as fd:
types = fd.read()
typo_response = mock.Mock()
typo_response.content = types
return typo_response
def test_demand_creation_return_codes(app, setup, remote_response):
with mock.patch('passerelle.utils.Request.post') as request_post:
request_post.side_effect = [get_typo_response(), remote_response]
response = app.post_json(
reverse(
'generic-endpoint',
kwargs={'connector': 'grenoble-gru', 'slug': setup.slug, 'endpoint': 'create-demand'},
),
params=BASIC_PAYLOAD,
status=200,
)
index = remote_response.index
if index == 0:
assert response.json['data'] == 'Demand successfully created'
elif index == 1:
assert response.json['err_desc'] == 'Remote service error'
elif index == 2:
assert response.json['err_desc'] == 'Authentication failed'
elif index == 3:
assert response.json['err_desc'] == 'Invalid input format'
elif index == 4:
assert response.json['err_desc'] == 'Required field not provided'
elif index == 5:
assert response.json['err_desc'] == 'Demand already exists'
else:
assert response.json['err_desc'] == 'Unknown error code (4242)'
def test_demand_creation_with_invalid_params(app, setup):
with mock.patch('passerelle.utils.Request.post') as request_post:
request_post.return_value = get_typo_response()
invalid_payload = BASIC_PAYLOAD.copy()
invalid_payload['applicant_status'] = 42
response = app.post_json(
reverse(
'generic-endpoint',
kwargs={'connector': 'grenoble-gru', 'slug': setup.slug, 'endpoint': 'create-demand'},
),
params=invalid_payload,
)
assert response.json['err'] == 1
assert 'applicant_status must be one of' in response.json['err_desc']
assert response.json['data'] is None
def test_demand_creation_params(app, setup):
with mock.patch('passerelle.utils.Request.post') as request_post:
request_post.return_value = get_typo_response()
payload = BASIC_PAYLOAD.copy()
payload['intervention_free_address'] = '169, rue du Château'
payload['applicant_free_address'] = '1, rue de l\'Est'
payload['applicant_phone'] = '01-02.03 04 05'
app.post_json(
reverse(
'generic-endpoint',
kwargs={'connector': 'grenoble-gru', 'slug': setup.slug, 'endpoint': 'create-demand'},
),
params=payload,
)
assert request_post.call_args[1]['data']['int_adresse_manuelle'] == '169, rue du Château'
assert request_post.call_args[1]['data']['dem_adresse_manuelle'] == '1, rue de l\'Est'
assert request_post.call_args[1]['data']['obs_demande_urgente'] == 1
assert request_post.call_args[1]['data']['dem_comp'] == 'Voirie'
assert request_post.call_args[1]['data']['dem_pav'] == 'déchetterie'
assert request_post.call_args[1]['data']['dem_reponse'] == 0
assert request_post.call_args[1]['data']['obs_description_probleme'] == 'thank you '
assert request_post.call_args[1]['data']['dem_tel'] == '0102030405'
payload['applicant_requires_reply'] = True
payload['urgent_demand'] = 'True'
app.post_json(
reverse(
'generic-endpoint',
kwargs={'connector': 'grenoble-gru', 'slug': setup.slug, 'endpoint': 'create-demand'},
),
params=payload,
status=200,
)
assert request_post.call_args[1]['data']['obs_demande_urgente'] == 1
assert request_post.call_args[1]['data']['dem_reponse'] == 1
# make sure empty dem_pav value is not sent
payload['dem_pav'] = ''
app.post_json(
reverse(
'generic-endpoint',
kwargs={'connector': 'grenoble-gru', 'slug': setup.slug, 'endpoint': 'create-demand'},
),
params=payload,
)
assert 'dem_pav' not in request_post.call_args[1]['data']
def test_add_attachment_to_demand(app, setup):
with mock.patch('passerelle.utils.Request.post') as request_post:
response = mock.Mock()
response.content = '01'
request_post.return_value = response
endpoint = reverse(
'generic-endpoint',
kwargs={
'connector': 'grenoble-gru',
'slug': setup.slug,
'endpoint': 'demand',
'rest': '42-42/add-attachment/',
},
)
payload = {'file': 'test.jpg'}
response = app.post_json(endpoint, params=payload, status=200)
assert response.json['data'] is None
assert response.json['err'] == 1
assert response.json['err_desc'] == 'file should be a dict'
file_b64_content = (
'/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAAUDBAQEAwUEBAQFBQUGBwwIBwcHBw8LCwkMEQ8SEhEPERETFh'
)
payload['file'] = {'content_type': 'image/jpg', 'filename': 'photo.jpg'}
response = app.post_json(endpoint, params=payload, status=400)
assert response.json['data'] is None
assert response.json['err'] == 1
assert response.json['err_desc'] == 'missing parameters: \'file[content]\'.'
payload['file']['content'] = file_b64_content
response = app.post_json(endpoint, params=payload)
assert request_post.call_args[1]['data']['dem_tiers_id'] == '42-42'
assert 'filename=photo.jpg' in request_post.call_args[1]['data']['piece_jointe']
assert 'filetype=image%2Fjpg' in request_post.call_args[1]['data']['piece_jointe']
assert (
'filecontent=%2F9j%2F4AAQSkZJRgABAQAAAQABAAD%2F2wBDAAUDBAQEAwUEBAQFBQUGBwwIBwcHBw8LCwkMEQ8SEhEPERETFh'
in request_post.call_args[1]['data']['piece_jointe']
)
def test_get_demand(app, setup):
with mock.patch('passerelle.utils.Request.post') as request_post:
response = mock.Mock()
response.content = DEMAND.replace('\n', '')
request_post.return_value = response
response = app.get(
reverse(
'generic-endpoint',
kwargs={
'connector': 'grenoble-gru',
'slug': setup.slug,
'endpoint': 'demand',
'rest': '42-42/',
},
)
)
assert request_post.call_args[1]['data']['dem_tiers_id'] == '42-42'
demand = response.json['data']
assert 'IdGRU' in demand
assert 'idInterne' in demand
assert isinstance(demand['demandeur'], dict)
assert isinstance(demand['lieu'], dict)
def test_get_demand_with_invalid_xml(app, setup):
with mock.patch('passerelle.utils.Request.post') as request_post:
response = mock.Mock()
response.content = DEMAND.replace('\n', '').replace('de Grenoble', '& Grenoble')
request_post.return_value = response
response = app.get(
reverse(
'generic-endpoint',
kwargs={
'connector': 'grenoble-gru',
'slug': setup.slug,
'endpoint': 'demand',
'rest': '42-42/',
},
)
)
assert request_post.call_args[1]['data']['dem_tiers_id'] == '42-42'
assert response.json['err']
assert 'Invalid XML returned:' in response.json['err_desc']
assert response.json['data'] is None
def test_get_pavs(app, setup):
with mock.patch('passerelle.utils.Request.post') as request_post:
response = mock.Mock()
json_response = mock.Mock()
with open(os.path.join(os.path.dirname(__file__), 'data', 'grenoble_gru_pavs.json')) as fd:
json_response.return_value = json.load(fd)
response.json = json_response
request_post.return_value = response
response = app.get(
reverse(
'generic-endpoint',
kwargs={'connector': 'grenoble-gru', 'slug': setup.slug, 'endpoint': 'pavs'},
)
)
assert response.json['err'] == 0
assert len(response.json['data']) == 7
for item in response.json['data']:
assert isinstance(item['id'], str)
assert item['text']