691 lines
25 KiB
Python
691 lines
25 KiB
Python
# passerelle - uniform access to multiple data sources and services
|
|
# Copyright (C) 2021 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 functools
|
|
import io
|
|
import json
|
|
import os
|
|
import uuid
|
|
import zipfile
|
|
from copy import deepcopy
|
|
|
|
import httmock
|
|
import lxml.etree as ET
|
|
import mock
|
|
import pytest
|
|
import utils
|
|
from requests.exceptions import ReadTimeout
|
|
from test_manager import login
|
|
|
|
from passerelle.base.models import Job
|
|
from passerelle.contrib.toulouse_smart.models import SmartRequest, ToulouseSmartResource, WcsRequest
|
|
|
|
TEST_BASE_DIR = os.path.join(os.path.dirname(__file__), 'data', 'toulouse_smart')
|
|
|
|
|
|
@pytest.fixture
|
|
def smart(db):
|
|
return utils.make_resource(
|
|
ToulouseSmartResource,
|
|
title='Test',
|
|
slug='test',
|
|
description='Test',
|
|
webservice_base_url='https://smart.example.com/',
|
|
basic_auth_username='username',
|
|
basic_auth_password='password',
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def wcs_service(settings):
|
|
wcs_service = {
|
|
'default': {
|
|
'title': 'test',
|
|
'url': 'https://wcs.example.com',
|
|
'secret': 'xxx',
|
|
'orig': 'passerelle',
|
|
},
|
|
}
|
|
settings.KNOWN_SERVICES = {'wcs': wcs_service}
|
|
return wcs_service
|
|
|
|
|
|
def mock_response(*path_contents):
|
|
def decorator(func):
|
|
@httmock.urlmatch()
|
|
def error(url, request):
|
|
assert False, 'request to %s' % url.geturl()
|
|
|
|
def register(path, payload, content, status_code=200, exception=None):
|
|
@httmock.urlmatch(path=path)
|
|
def handler(url, request):
|
|
if payload and json.loads(request.body) != payload:
|
|
assert False, 'wrong payload sent to request to %s' % url.geturl()
|
|
if exception:
|
|
raise exception
|
|
return httmock.response(status_code, content)
|
|
|
|
return handler
|
|
|
|
@functools.wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
handlers = []
|
|
for row in path_contents:
|
|
handlers.append(register(*row))
|
|
handlers.append(error)
|
|
|
|
with httmock.HTTMock(*handlers):
|
|
return func(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
def get_json_file(filename):
|
|
with open(os.path.join(TEST_BASE_DIR, "%s.json" % filename)) as desc:
|
|
return desc.read()
|
|
|
|
|
|
@mock_response(['/v1/type-intervention', None, b'<List></List>'])
|
|
def test_empty_intervention_types(smart):
|
|
assert smart.get_intervention_types() == []
|
|
|
|
|
|
INTERVENTION_TYPES = '''<List>
|
|
<item>
|
|
<id>1234</id>
|
|
<name>coin</name>
|
|
<properties>
|
|
<properties>
|
|
<name>FIELD1</name>
|
|
<displayName>Champ 1</displayName>
|
|
<type>string</type>
|
|
<required>false</required>
|
|
<defaultValue>Ne sait pas</defaultValue>
|
|
<restrictedValues>
|
|
<restrictedValues>Candélabre</restrictedValues>
|
|
<restrictedValues>Mât</restrictedValues>
|
|
<restrictedValues>Ne sait pas</restrictedValues>
|
|
</restrictedValues>
|
|
</properties>
|
|
<properties>
|
|
<name>FIELD2</name>
|
|
<displayName>Champ 2</displayName>
|
|
<type>int</type>
|
|
<required>true</required>
|
|
</properties>
|
|
</properties>
|
|
</item>
|
|
</List>'''.encode()
|
|
|
|
|
|
@mock_response(['/v1/type-intervention', None, INTERVENTION_TYPES])
|
|
def test_model_intervention_types(smart):
|
|
assert smart.get_intervention_types() == [
|
|
{
|
|
'id': '1234',
|
|
'name': 'coin',
|
|
'order': 1,
|
|
'properties': [
|
|
{
|
|
'name': 'FIELD1',
|
|
'displayName': 'Champ 1',
|
|
'required': False,
|
|
'type': 'item',
|
|
'defaultValue': 'Ne sait pas',
|
|
'restrictedValues': ['Candélabre', 'Mât', 'Ne sait pas'],
|
|
},
|
|
{'name': 'FIELD2', 'displayName': 'Champ 2', 'required': True, 'type': 'int'},
|
|
],
|
|
},
|
|
]
|
|
|
|
|
|
URL = '/toulouse-smart/test/'
|
|
|
|
|
|
@mock_response(['/v1/type-intervention', None, INTERVENTION_TYPES])
|
|
def test_endpoint_intervention_types(app, smart):
|
|
resp = app.get(URL + 'type-intervention')
|
|
assert resp.json == {'data': [{'id': 'coin', 'text': 'coin', 'uuid': '1234'}], 'err': 0}
|
|
|
|
|
|
@mock_response()
|
|
def test_endpoint_intervention_types_unavailable(app, smart):
|
|
resp = app.get(URL + 'type-intervention')
|
|
assert resp.json == {'data': [{'id': '', 'text': 'Service is unavailable', 'disabled': True}], 'err': 0}
|
|
|
|
|
|
@mock_response(['/v1/type-intervention', None, INTERVENTION_TYPES])
|
|
def test_manage_intervention_types(app, smart, admin_user):
|
|
login(app)
|
|
resp = app.get('/manage' + URL + 'type-intervention/')
|
|
assert [[td.text for td in tr.cssselect('td,th')] for tr in resp.pyquery('tr')] == [
|
|
["Nom du type d'intervention", 'Nom', 'Type', 'Requis', 'Valeur par défaut'],
|
|
['1 - coin'],
|
|
[None, 'FIELD1', 'item («Candélabre», «Mât», «Ne sait pas»)', '✘', 'Ne sait pas'],
|
|
[None, 'FIELD2', 'int', '✔', None],
|
|
]
|
|
resp = resp.click('Export to blocks')
|
|
with zipfile.ZipFile(io.BytesIO(resp.body)) as zip_file:
|
|
assert zip_file.namelist() == ['block-coin.wcs']
|
|
with zip_file.open('block-coin.wcs') as fd:
|
|
content = ET.tostring(ET.fromstring(fd.read()), pretty_print=True).decode()
|
|
assert (
|
|
content
|
|
== '''<block id="1234">
|
|
<name>coin</name>
|
|
<slug>coin</slug>
|
|
<fields>
|
|
<field>
|
|
<id>038a8c2e-14de-4d4f-752f-496eb7fe90d7</id>
|
|
<label>Champ 1</label>
|
|
<type>item</type>
|
|
<required>False</required>
|
|
<varname>field1</varname>
|
|
<display_locations>
|
|
<display_location>validation</display_location>
|
|
<display_location>summary</display_location>
|
|
</display_locations>
|
|
<items>
|
|
<item>Candélabre</item>
|
|
<item>Mât</item>
|
|
<item>Ne sait pas</item>
|
|
</items>
|
|
</field>
|
|
<field>
|
|
<id>e72f251a-5eef-5b78-c35a-94b549510029</id>
|
|
<label>Champ 2</label>
|
|
<type>string</type>
|
|
<required>True</required>
|
|
<varname>field2</varname>
|
|
<display_locations>
|
|
<display_location>validation</display_location>
|
|
<display_location>summary</display_location>
|
|
</display_locations>
|
|
<validation>
|
|
<type>digits</type>
|
|
</validation>
|
|
</field>
|
|
</fields>
|
|
</block>
|
|
'''
|
|
)
|
|
|
|
|
|
@mock_response(
|
|
['/v1/intervention', None, get_json_file('get_intervention')],
|
|
)
|
|
def test_get_intervention(app, smart):
|
|
resp = app.get(URL + 'get-intervention?id=3f0558bd-7d85-49a8-97e4-d07bc7f8dc9b')
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['id'] == '3f0558bd-7d85-49a8-97e4-d07bc7f8dc9b'
|
|
assert resp.json['data']['state'] == {
|
|
'id': 'e844e67f-5382-4c0f-94d8-56f618263485',
|
|
'table': None,
|
|
'stateLabel': 'Nouveau',
|
|
'closes': False,
|
|
}
|
|
|
|
|
|
@mock_response(
|
|
['/v1/intervention', None, None, 500],
|
|
)
|
|
def test_get_intervention_error_status(app, smart):
|
|
resp = app.get(URL + 'get-intervention?id=3f0558bd-7d85-49a8-97e4-d07bc7f8dc9b')
|
|
assert resp.json['err']
|
|
assert 'failed to get' in resp.json['err_desc']
|
|
|
|
|
|
@mock_response(
|
|
['/v1/intervention', None, None, 404],
|
|
)
|
|
def test_get_intervention_wrond_id(app, smart):
|
|
resp = app.get(URL + 'get-intervention?id=3f0558bd-7d85-49a8-97e4-d07bc7f8dc9b')
|
|
assert resp.json['err']
|
|
assert 'failed to get' in resp.json['err_desc']
|
|
assert '404' in resp.json['err_desc']
|
|
|
|
|
|
CREATE_INTERVENTION_PAYLOAD_EXTRA = {
|
|
'slug': 'coin',
|
|
'description': 'coin coin',
|
|
'lat': 48.833708,
|
|
'lon': 2.323349,
|
|
'cityId': '12345',
|
|
'interventionCreated': '2021-06-30T16:08:05',
|
|
'interventionDesired': '2021-06-30T16:08:05',
|
|
'submitterFirstName': 'John',
|
|
'submitterLastName': 'Doe',
|
|
'submitterMail': 'john.doe@example.com',
|
|
'submitterPhone': '0123456789',
|
|
'submitterAddress': '3 rue des champs de blés',
|
|
'submitterType': 'usager',
|
|
'external_number': '42-2',
|
|
'external_status': 'statut-1-wcs',
|
|
'address': 'https://wcs.example.com/backoffice/management/foo/2/',
|
|
'form_api_url': 'https://wcs.example.com/api/forms/foo/2/',
|
|
'checkDuplicated': 'False',
|
|
'onPrivateLand': 'True',
|
|
'safeguardRequired': True,
|
|
}
|
|
|
|
|
|
FIELDS_PAYLOAD = {
|
|
'coin_raw': [
|
|
{
|
|
'field1': 'Candélabre',
|
|
'field1_raw': 'Candélabre',
|
|
'field2': '42',
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
CREATE_INTERVENTION_PAYLOAD = {
|
|
'fields': FIELDS_PAYLOAD,
|
|
'extra': CREATE_INTERVENTION_PAYLOAD_EXTRA,
|
|
}
|
|
|
|
UUID = uuid.UUID('12345678123456781234567812345678')
|
|
|
|
CREATE_INTERVENTION_QUERY = {
|
|
'description': 'coin coin',
|
|
'cityId': '12345',
|
|
'interventionCreated': '2021-06-30T16:08:05Z',
|
|
'interventionDesired': '2021-06-30T16:08:05Z',
|
|
'submitterFirstName': 'John',
|
|
'submitterLastName': 'Doe',
|
|
'submitterMail': 'john.doe@example.com',
|
|
'submitterPhone': '0123456789',
|
|
'submitterAddress': '3 rue des champs de bl\u00e9s',
|
|
'submitterType': 'usager',
|
|
'external_number': '42-2',
|
|
'external_status': 'statut-1-wcs',
|
|
'address': 'https://wcs.example.com/backoffice/management/foo/2/',
|
|
'interventionData': {'FIELD1': 'Candélabre', 'FIELD2': 42},
|
|
'geom': {'type': 'Point', 'coordinates': [2.323349, 48.833708], 'crs': 'EPSG:4326'},
|
|
'interventionTypeId': '1234',
|
|
'notificationUrl': 'http://testserver/toulouse-smart/test/update-intervention?uuid=%s' % str(UUID),
|
|
'onPrivateLand': 'true',
|
|
'safeguardRequired': 'true',
|
|
}
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],
|
|
)
|
|
@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)
|
|
def test_create_intervention(mocked_uuid4, app, smart):
|
|
with pytest.raises(WcsRequest.DoesNotExist):
|
|
smart.wcs_requests.get(uuid=UUID)
|
|
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert str(UUID) in CREATE_INTERVENTION_QUERY['notificationUrl']
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['uuid'] == str(UUID)
|
|
assert resp.json['data']['wcs_form_api_url'] == 'https://wcs.example.com/api/forms/foo/2/'
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.wcs_form_api_url == 'https://wcs.example.com/api/forms/foo/2/'
|
|
assert wcs_request.wcs_form_number == '42-2'
|
|
assert wcs_request.payload == CREATE_INTERVENTION_QUERY
|
|
assert wcs_request.result == json.loads(get_json_file('create_intervention'))
|
|
assert wcs_request.status == 'sent'
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],
|
|
)
|
|
@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)
|
|
def test_create_intervention_async(mocked_uuid4, app, smart):
|
|
mocked_push = mock.patch(
|
|
"passerelle.contrib.toulouse_smart.models.WcsRequest.push",
|
|
return_value=False,
|
|
)
|
|
mocked_push.start()
|
|
|
|
assert Job.objects.count() == 0
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert str(UUID) in CREATE_INTERVENTION_QUERY['notificationUrl']
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['uuid'] == str(UUID)
|
|
assert resp.json['data']['wcs_form_api_url'] == 'https://wcs.example.com/api/forms/foo/2/'
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.wcs_form_api_url == 'https://wcs.example.com/api/forms/foo/2/'
|
|
assert wcs_request.wcs_form_number == '42-2'
|
|
assert wcs_request.payload == CREATE_INTERVENTION_QUERY
|
|
assert wcs_request.status == 'registered'
|
|
|
|
mocked_push.stop()
|
|
assert Job.objects.count() == 1
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'registered'
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'completed'
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.result == json.loads(get_json_file('create_intervention'))
|
|
assert wcs_request.status == 'sent'
|
|
|
|
|
|
def test_create_intervention_wrong_payload(app, smart):
|
|
payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
|
|
del payload['extra']['slug']
|
|
resp = app.post_json(URL + 'create-intervention/', params=payload, status=400)
|
|
assert resp.json['err']
|
|
assert "'slug' is a required property" in resp.json['err_desc']
|
|
|
|
|
|
@mock_response()
|
|
def test_create_intervention_types_unavailable(app, smart):
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert resp.json['err']
|
|
assert 'Service is unavailable' in resp.json['err_desc']
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
)
|
|
def test_create_intervention_wrong_block_slug(app, smart):
|
|
payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
|
|
payload['extra']['slug'] = 'coin-coin'
|
|
resp = app.post_json(URL + 'create-intervention/', params=payload, status=400)
|
|
assert resp.json['err']
|
|
assert "unknown 'coin-coin' block slug" in resp.json['err_desc']
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
)
|
|
def test_create_intervention_no_block(app, smart):
|
|
payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
|
|
del payload['fields']['coin_raw']
|
|
resp = app.post_json(URL + 'create-intervention/', params=payload, status=400)
|
|
assert resp.json['err']
|
|
assert "cannot find 'coin' block field content" in resp.json['err_desc']
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
)
|
|
def test_create_intervention_cast_error(app, smart):
|
|
payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
|
|
payload['fields']['coin_raw'][0]['field2'] = 'not-an-integer'
|
|
resp = app.post_json(URL + 'create-intervention/', params=payload, status=400)
|
|
assert resp.json['err']
|
|
assert "cannot cast 'field2' field to <class 'int'>" in resp.json['err_desc']
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
)
|
|
def test_create_intervention_missing_value(app, smart):
|
|
field_payload = {
|
|
'coin_raw': [
|
|
{
|
|
'field1': 'Candélabre',
|
|
'field1_raw': 'Candélabre',
|
|
'field2': None,
|
|
},
|
|
],
|
|
}
|
|
payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
|
|
payload['fields'] = field_payload
|
|
resp = app.post_json(URL + 'create-intervention/', params=payload, status=400)
|
|
assert resp.json['err']
|
|
assert "field is required on 'coin' block" in resp.json['err_desc']
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
)
|
|
def test_create_intervention_missing_field(app, smart):
|
|
field_payload = {
|
|
'coin_raw': [
|
|
{
|
|
'field1': 'Candélabre',
|
|
'field1_raw': 'Candélabre',
|
|
},
|
|
],
|
|
}
|
|
payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
|
|
payload['fields'] = field_payload
|
|
resp = app.post_json(URL + 'create-intervention/', params=payload, status=400)
|
|
assert resp.json['err']
|
|
assert "field is required on 'coin' block" in resp.json['err_desc']
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, None, 500],
|
|
)
|
|
@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)
|
|
def test_create_intervention_twice_error(mocked_uuid4, app, smart):
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD, status=400)
|
|
assert resp.json['err']
|
|
assert 'already created' in resp.json['err_desc']
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, None, 500],
|
|
)
|
|
@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)
|
|
def test_create_intervention_transport_error(mocked_uuid, app, freezer, smart):
|
|
freezer.move_to('2021-07-08 00:00:00')
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'registered'
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.status == 'registered'
|
|
assert 'failed to post' in wcs_request.result
|
|
|
|
freezer.move_to('2021-07-08 00:00:03')
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'registered'
|
|
assert job.update_timestamp > job.creation_timestamp
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.status == 'registered'
|
|
assert 'failed to post' in wcs_request.result
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, None, None, ReadTimeout('timeout')],
|
|
)
|
|
@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)
|
|
def test_create_intervention_timeout_error(mocked_uuid, app, smart):
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'registered'
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.status == 'registered'
|
|
assert 'failed to post' in wcs_request.result
|
|
assert 'timeout' in wcs_request.result
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, None, 500],
|
|
)
|
|
@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)
|
|
def test_create_intervention_inconsistency_id_error(mocked_uuid4, app, freezer, smart):
|
|
freezer.move_to('2021-07-08 00:00:00')
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.status == 'registered'
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'registered'
|
|
|
|
freezer.move_to('2021-07-08 00:00:03')
|
|
wcs_request.delete()
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'failed'
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, 'not json content'],
|
|
)
|
|
@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)
|
|
def test_create_intervention_content_error(mocked_uuid, app, freezer, smart):
|
|
freezer.move_to('2021-07-08 00:00:00')
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.status == 'registered'
|
|
assert 'invalid json' in wcs_request.result
|
|
|
|
|
|
UPDATE_INTERVENTION_PAYLOAD = {
|
|
'data': {
|
|
'status': 'close manque info',
|
|
'type_retour_cloture': 'Smart non Fait',
|
|
'libelle_cloture': "rien à l'adresse indiquée",
|
|
'commentaire_cloture': 'le commentaire',
|
|
}
|
|
}
|
|
UPDATE_INTERVENTION_QUERY = UPDATE_INTERVENTION_PAYLOAD
|
|
WCS_RESPONSE_SUCCESS = '{"err": 0, "url": null}'
|
|
WCS_RESPONSE_ERROR = '{"err": 1, "err_class": "Access denied", "err_desc": null}'
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],
|
|
['/api/forms/foo/2/hooks/update_intervention/', UPDATE_INTERVENTION_QUERY, WCS_RESPONSE_SUCCESS],
|
|
)
|
|
@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)
|
|
def test_update_intervention(mocked_uuid, app, smart, wcs_service):
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
assert CREATE_INTERVENTION_QUERY[
|
|
'notificationUrl'
|
|
] == 'http://testserver/toulouse-smart/test/update-intervention?uuid=%s' % str(UUID)
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
|
|
mocked_push = mock.patch(
|
|
"passerelle.contrib.toulouse_smart.models.SmartRequest.push",
|
|
return_value=False,
|
|
)
|
|
mocked_push.start()
|
|
|
|
assert Job.objects.count() == 0
|
|
url = URL + 'update-intervention?uuid=%s' % str(UUID)
|
|
resp = app.post_json(url, params=UPDATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['uuid'] == str(UUID)
|
|
assert resp.json['data']['payload']['data']['type_retour_cloture'] == 'Smart non Fait'
|
|
smart_request = smart.wcs_requests.get(uuid=UUID).smart_requests.get()
|
|
|
|
mocked_push.stop()
|
|
assert Job.objects.count() == 1
|
|
job = Job.objects.get(method_name='update_intervention_job')
|
|
assert job.status == 'registered'
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='update_intervention_job')
|
|
assert job.status == 'completed'
|
|
smart_request = smart.wcs_requests.get(uuid=UUID).smart_requests.get()
|
|
assert smart_request.result == {'err': 0, 'url': None}
|
|
|
|
|
|
def test_update_intervention_wrong_uuid(app, smart):
|
|
with pytest.raises(WcsRequest.DoesNotExist):
|
|
smart.wcs_requests.get(uuid=UUID)
|
|
|
|
url = URL + 'update-intervention?uuid=%s' % str(UUID)
|
|
resp = app.post_json(url, params=UPDATE_INTERVENTION_PAYLOAD, status=400)
|
|
assert resp.json['err']
|
|
assert 'Cannot find intervention' in resp.json['err_desc']
|
|
assert SmartRequest.objects.count() == 0
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],
|
|
)
|
|
@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)
|
|
def test_update_intervention_job_wrong_service(mocked_uuid, app, smart, wcs_service):
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
|
|
wcs_service['default']['url'] = 'http://wrong.example.com'
|
|
url = URL + 'update-intervention?uuid=%s' % str(UUID)
|
|
resp = app.post_json(url, params=UPDATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='update_intervention_job')
|
|
assert job.status == 'completed'
|
|
smart_request = smart.wcs_requests.get(uuid=UUID).smart_requests.get()
|
|
assert 'Cannot find wcs service' in smart_request.result
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],
|
|
['/api/forms/foo/2/hooks/update_intervention/', UPDATE_INTERVENTION_QUERY, WCS_RESPONSE_ERROR, 403],
|
|
)
|
|
@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)
|
|
def test_update_intervention_job_wcs_error(mocked_uuid, app, smart, wcs_service):
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
|
|
url = URL + 'update-intervention?uuid=%s' % str(UUID)
|
|
resp = app.post_json(url, params=UPDATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='update_intervention_job')
|
|
assert job.status == 'completed'
|
|
smart_request = smart.wcs_requests.get(uuid=UUID).smart_requests.get()
|
|
assert smart_request.result == {'err': 1, 'err_class': 'Access denied', 'err_desc': None}
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],
|
|
['/api/forms/foo/2/hooks/update_intervention/', UPDATE_INTERVENTION_QUERY, 'bla', 500],
|
|
)
|
|
@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)
|
|
def test_update_intervention_job_transport_error(mocked_uuid, app, freezer, smart, wcs_service):
|
|
freezer.move_to('2021-07-08 00:00:00')
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
|
|
url = URL + 'update-intervention?uuid=%s' % str(UUID)
|
|
resp = app.post_json(url, params=UPDATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
job = Job.objects.get(method_name='update_intervention_job')
|
|
assert job.status == 'registered'
|
|
smart_request = smart.wcs_requests.get(uuid=UUID).smart_requests.get()
|
|
assert smart_request.result == None
|
|
|
|
freezer.move_to('2021-07-08 00:00:03')
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='update_intervention_job')
|
|
assert job.status == 'registered'
|
|
assert job.update_timestamp > job.creation_timestamp
|
|
smart_request = smart.wcs_requests.get(uuid=UUID).smart_requests.get()
|
|
assert smart_request.result == None
|