1276 lines
48 KiB
Python
1276 lines
48 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 base64
|
|
import cgi # noqa pylint: disable=deprecated-module
|
|
import functools
|
|
import io
|
|
import json
|
|
import os
|
|
import uuid
|
|
import zipfile
|
|
from copy import deepcopy
|
|
from unittest import mock
|
|
|
|
import httmock
|
|
import lxml.etree as ET
|
|
import pytest
|
|
from django.utils.encoding import force_str
|
|
from requests.exceptions import ConnectionError, ReadTimeout
|
|
|
|
import tests.utils
|
|
from passerelle.base.models import Job
|
|
from passerelle.contrib.toulouse_smart.models import (
|
|
SmartRequest,
|
|
ToulouseSmartResource,
|
|
WcsRequest,
|
|
WcsRequestFile,
|
|
)
|
|
from passerelle.contrib.toulouse_smart.utils import localtz_to_utc, utc_to_localtz
|
|
from tests.test_manager import login
|
|
|
|
TEST_BASE_DIR = os.path.join(os.path.dirname(__file__), 'data', 'toulouse_smart')
|
|
|
|
|
|
@pytest.fixture
|
|
def smart(db, settings):
|
|
settings.TIME_ZONE = 'Europe/Paris'
|
|
settings.USE_TZ = True
|
|
return tests.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:
|
|
ctype, pdict = cgi.parse_header(request.headers['content-type'])
|
|
if ctype == 'multipart/form-data':
|
|
# here payload is an expected multipart contents list
|
|
pdict['boundary'] = bytes(pdict['boundary'], 'utf-8')
|
|
pdict['CONTENT-LENGTH'] = request.headers['Content-Length']
|
|
postvars = cgi.parse_multipart(io.BytesIO(request.body), pdict)
|
|
for i, media_content in enumerate(postvars['media']):
|
|
assert media_content == payload[i], (
|
|
'wrong multipart content sent to %s' % url.geturl()
|
|
)
|
|
else:
|
|
assert json.loads(request.body) == payload, (
|
|
'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()
|
|
|
|
|
|
def get_media_file(filename):
|
|
with open(os.path.join(TEST_BASE_DIR, '%s' % filename), 'rb') as desc:
|
|
return desc.read()
|
|
|
|
|
|
def test_save_daylight_time_change(settings):
|
|
settings.TIME_ZONE = 'Europe/Paris'
|
|
settings.USE_TZ = True
|
|
assert localtz_to_utc('2022-10-30T02:19:48') == '2022-10-30T01:19:48+00:00'
|
|
assert utc_to_localtz('2022-10-30T01:19:48+00:00') == '2022-10-30T02:19:48'
|
|
|
|
|
|
@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>TYPE-OBJET</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>
|
|
<name>IGNORED-FIELD-HAVING-NO-TYPE</name>
|
|
<displayName>Champ 3</displayName>
|
|
</properties>
|
|
<properties>
|
|
<name>IGNORED-FIELD-HAVING-UNKNOWN-TYPE</name>
|
|
<displayName>Champ 3</displayName>
|
|
<type>plop</type>
|
|
</properties>
|
|
</properties>
|
|
</item>
|
|
<item>
|
|
<id>0002</id>
|
|
<name>empty</name>
|
|
</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': 'TYPE-OBJET',
|
|
'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'},
|
|
],
|
|
},
|
|
{
|
|
'id': '0002',
|
|
'name': 'empty',
|
|
'order': 2,
|
|
},
|
|
]
|
|
|
|
|
|
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'},
|
|
{'id': 'empty', 'text': 'empty', 'uuid': '0002'},
|
|
],
|
|
'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, 'TYPE-OBJET', 'item («Candélabre», «Mât», «Ne sait pas»)', '✘', 'Ne sait pas'],
|
|
[None, 'FIELD2', 'int', '✔', None],
|
|
['2 - empty'],
|
|
]
|
|
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>522697a9-de01-b198-9e37-58c35718203a</id>
|
|
<label>Champ 1</label>
|
|
<type>item</type>
|
|
<required>False</required>
|
|
<varname>type_objet</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>
|
|
'''
|
|
)
|
|
|
|
|
|
INTERVENTION_ID = json.loads(get_json_file('create_intervention'))['id']
|
|
|
|
|
|
@mock_response(
|
|
['/v1/intervention', None, get_json_file('create_intervention')],
|
|
)
|
|
def test_get_intervention(app, smart):
|
|
resp = app.get(URL + 'get-intervention?id=%s' % INTERVENTION_ID)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['id'] == INTERVENTION_ID
|
|
assert resp.json['data']['state'] == {
|
|
'id': 'e844e67f-5382-4c0f-94d8-56f618263485',
|
|
'table': None,
|
|
'stateLabel': 'Nouveau',
|
|
'closes': False,
|
|
}
|
|
assert resp.json['data']['interventionCreated'] == '2021-07-07T14:19:31.302000'
|
|
assert resp.json['data']['interventionDesired'] == '2021-06-30T18:08:05'
|
|
|
|
|
|
@mock_response(
|
|
['/v1/intervention', None, None, 500],
|
|
)
|
|
def test_get_intervention_error_status(app, smart):
|
|
resp = app.get(URL + 'get-intervention?id=%s' % INTERVENTION_ID)
|
|
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=%s' % INTERVENTION_ID)
|
|
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-30T18:08:05.500931+02:00',
|
|
'interventionDesired': '2021-06-30T18:08:05.500931+02:00',
|
|
'submitterFirstName': 'John',
|
|
'submitterLastName': 'Doe',
|
|
'submitterMail': 'john.doe@example.com',
|
|
'submitterPhone': '0123456789',
|
|
'submitterAddress': '3 rue des champs de blés',
|
|
'submitterType': 'usager',
|
|
'externalReferences': 'AlloToulouse',
|
|
'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': [
|
|
{
|
|
'type_objet': 'Candélabre',
|
|
'type_objet_raw': 'Candélabre',
|
|
'field2': '42',
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
CREATE_INTERVENTION_PAYLOAD = {
|
|
'fields': FIELDS_PAYLOAD,
|
|
'extra': CREATE_INTERVENTION_PAYLOAD_EXTRA,
|
|
}
|
|
|
|
UUID = uuid.UUID('12345678123456781234567812345678')
|
|
|
|
CREATE_INTERVENTION_QUERY = {
|
|
'add_media_url': 'http://testserver/toulouse-smart/test/add-media?uuid=%s' % str(UUID),
|
|
'description': 'coin coin',
|
|
'cityId': '12345',
|
|
'interventionCreated': '2021-06-30T16:08:05.500931+00:00',
|
|
'interventionDesired': '2021-06-30T16:08:05.500931+00:00',
|
|
'submitterFirstName': 'John',
|
|
'submitterLastName': 'Doe',
|
|
'submitterMail': 'john.doe@example.com',
|
|
'submitterPhone': '0123456789',
|
|
'submitterAddress': '3 rue des champs de bl\u00e9s',
|
|
'submitterType': 'usager',
|
|
'externalReferences': 'AlloToulouse',
|
|
'external_number': '42-2',
|
|
'external_status': 'statut-1-wcs',
|
|
'address': 'https://wcs.example.com/backoffice/management/foo/2/',
|
|
'interventionData': {'TYPE-OBJET': '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),
|
|
'notification_url': 'http://testserver/toulouse-smart/test/update-intervention?uuid=%s' % str(UUID),
|
|
'onPrivateLand': 'true',
|
|
'safeguardRequired': 'true',
|
|
}
|
|
|
|
CREATE_INTERVENTION_QUERY_WITHOUT_PROPERTIES = deepcopy(CREATE_INTERVENTION_QUERY)
|
|
CREATE_INTERVENTION_QUERY_WITHOUT_PROPERTIES['interventionTypeId'] = '0002'
|
|
CREATE_INTERVENTION_QUERY_WITHOUT_PROPERTIES['interventionData'] = {}
|
|
|
|
|
|
@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['notification_url']
|
|
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['id'] == INTERVENTION_ID
|
|
assert wcs_request.result['state'] == {
|
|
'id': 'e844e67f-5382-4c0f-94d8-56f618263485',
|
|
'table': None,
|
|
'stateLabel': 'Nouveau',
|
|
'closes': False,
|
|
}
|
|
assert wcs_request.result['interventionCreated'] == '2021-07-07T14:19:31.302000'
|
|
assert wcs_request.result['interventionDesired'] == '2021-06-30T18:08:05'
|
|
assert wcs_request.status == 'sent'
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY_WITHOUT_PROPERTIES, get_json_file('create_intervention')],
|
|
)
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_create_intervention_without_properties(mocked_uuid4, app, smart):
|
|
payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
|
|
payload['extra']['slug'] = 'empty'
|
|
payload['fields']['coin_raw'] = None
|
|
resp = app.post_json(URL + 'create-intervention/', params=payload)
|
|
assert not resp.json['err']
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY_WITHOUT_PROPERTIES, get_json_file('create_intervention')],
|
|
)
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_create_intervention_providing_empty_block(mocked_uuid4, app, smart):
|
|
payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
|
|
payload['extra']['slug'] = 'empty'
|
|
payload['fields']['coin_raw'] = None
|
|
payload['fields']['empty_raw'] = None
|
|
resp = app.post_json(URL + 'create-intervention/', params=payload)
|
|
assert not resp.json['err']
|
|
|
|
|
|
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 resp.json['err_desc'] == "'field2' field is required on 'coin' block"
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
)
|
|
def test_create_intervention_string_payload(app, smart):
|
|
payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
|
|
payload['fields']['coin_raw'] = 'plop'
|
|
resp = app.post_json(URL + 'create-intervention/', params=payload, status=400)
|
|
assert resp.json['err']
|
|
assert (
|
|
resp.json['err_desc']
|
|
== "cannot retrieve 'coin' block content from post data: got a <class 'str'> where a dict was expected"
|
|
)
|
|
|
|
|
|
@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': [
|
|
{
|
|
'type_objet': 'Candélabre',
|
|
'type_objet_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': [
|
|
{
|
|
'type_objet': 'Candélabre',
|
|
'type_objet_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(mocked_uuid4, app, smart):
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['status'] == 'registered'
|
|
assert smart.wcs_requests.count() == 1
|
|
|
|
# re-create intervention after it success: no error is returned, but no new request is sent
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['status'] == 'registered'
|
|
assert smart.wcs_requests.count() == 1
|
|
|
|
|
|
@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)
|
|
assert not resp.json['err']
|
|
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)
|
|
assert not resp.json['err']
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.status == 'registered'
|
|
assert 'invalid json' in wcs_request.result
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, '400 Client Error', 400],
|
|
)
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_create_intervention_client_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']
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert '400 Client Error' in wcs_request.result
|
|
assert wcs_request.tries == 1
|
|
assert wcs_request.status == 'registered'
|
|
assert wcs_request.smart_requests.count() == 0
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'registered'
|
|
|
|
smart.jobs()
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.tries == 2
|
|
assert wcs_request.status == 'registered'
|
|
assert wcs_request.smart_requests.count() == 1
|
|
smart_request = wcs_request.smart_requests.latest('id')
|
|
assert smart_request.payload['creation_response']['status'] == 'registered'
|
|
|
|
smart.jobs()
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.tries == 3
|
|
assert wcs_request.status == 'registered'
|
|
assert wcs_request.smart_requests.count() == 2
|
|
smart_request = wcs_request.smart_requests.latest('id')
|
|
assert smart_request.payload['creation_response']['status'] == 'registered'
|
|
|
|
freezer.move_to('2021-07-08 01:00:01')
|
|
smart.jobs()
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.tries == 4
|
|
assert wcs_request.status == 'registered'
|
|
assert wcs_request.smart_requests.count() == 3
|
|
smart_request = wcs_request.smart_requests.latest('id')
|
|
assert smart_request.payload['creation_response']['status'] == 'registered'
|
|
|
|
freezer.move_to('2021-07-09 01:00:02')
|
|
smart.jobs()
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert '400 Client Error' in wcs_request.result
|
|
assert wcs_request.tries == 5
|
|
assert wcs_request.status == 'failed'
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'failed'
|
|
assert '400 Client Error' in job.status_details['error_summary']
|
|
assert wcs_request.smart_requests.count() == 4
|
|
smart_request = wcs_request.smart_requests.latest('id')
|
|
assert smart_request.payload['creation_response']['status'] == 'failed'
|
|
|
|
# re-create intervention after it fails: a new request is sent
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['status'] == 'registered'
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert '400 Client Error' in wcs_request.result
|
|
assert wcs_request.tries == 1
|
|
assert wcs_request.status == 'registered'
|
|
|
|
|
|
@mock.patch('passerelle.utils.RequestSession.request')
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_create_intervention_timeout(mocked_uuid, mocked_get, app, freezer, smart):
|
|
from tests.utils import FakedResponse
|
|
|
|
mocked_get.side_effect = [
|
|
FakedResponse(
|
|
headers={'Content-Type': 'application/xml; charset=charset=utf-8'},
|
|
status_code=200,
|
|
content=INTERVENTION_TYPES,
|
|
),
|
|
ReadTimeout('timeout'),
|
|
FakedResponse(
|
|
headers={'Content-Type': 'application/json'},
|
|
status_code=200,
|
|
content=get_json_file('create_intervention'),
|
|
),
|
|
]
|
|
|
|
# synchronous requests
|
|
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']
|
|
assert mocked_get.call_count == 2
|
|
assert mocked_get.call_args_list[0][1]['timeout'] == 10
|
|
assert mocked_get.call_args_list[1][1]['timeout'] == 10
|
|
|
|
# asynchronous request
|
|
freezer.move_to('2021-07-09 01:00:02')
|
|
smart.jobs()
|
|
assert mocked_get.call_count == 3
|
|
assert mocked_get.call_args_list[2][1]['timeout'] == 25
|
|
|
|
|
|
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[
|
|
'notification_url'
|
|
] == 'http://testserver/toulouse-smart/test/update-intervention?uuid=%s' % str(UUID)
|
|
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=0123456789'
|
|
resp = app.post_json(url, params=UPDATE_INTERVENTION_PAYLOAD, status=400)
|
|
assert resp.json['err']
|
|
assert 'is not a valid UUID.' in resp.json['err_desc']
|
|
assert SmartRequest.objects.count() == 0
|
|
|
|
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, caplog):
|
|
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_wcs_error_not_json(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 is 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 is 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,
|
|
None,
|
|
500,
|
|
ConnectionError('No address associated with hostname'),
|
|
],
|
|
)
|
|
@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 is 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 is None
|
|
|
|
|
|
ADD_MEDIA_PAYLOAD = {
|
|
'files/0': {
|
|
'filename': '201x201.jpg',
|
|
'content_type': 'image/jpeg',
|
|
'content': force_str(base64.b64encode(get_media_file('201x201.jpg'))),
|
|
},
|
|
'files/1': None,
|
|
}
|
|
|
|
ADD_MEDIA_QUERY = [get_media_file('201x201.jpg')]
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention/%s/media' % INTERVENTION_ID, ADD_MEDIA_QUERY, 200],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],
|
|
)
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_add_media(mocked_uuid, app, smart):
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
|
|
url = resp.json['data']['payload']['add_media_url']
|
|
url = URL + 'add-media?uuid=%s' % str(UUID)
|
|
resp = app.post_json(url, params=ADD_MEDIA_PAYLOAD)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['uuid'] == str(UUID)
|
|
assert resp.json['data']['nb_registered'] == 1
|
|
assert Job.objects.count() == 1
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'registered'
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
wcs_request_file = wcs_request.files.get(**job.parameters)
|
|
path = wcs_request_file.content.path
|
|
assert os.path.isfile(path)
|
|
with wcs_request_file.content.open('rb') as desc:
|
|
assert desc.read() == get_media_file('201x201.jpg')
|
|
|
|
# smart not responding
|
|
mocked_push = mock.patch(
|
|
'passerelle.contrib.toulouse_smart.models.WcsRequestFile.push',
|
|
return_value=False,
|
|
)
|
|
mocked_push.start()
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'registered'
|
|
|
|
# smart responding
|
|
mocked_push.stop()
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'completed'
|
|
wcs_request_file = wcs_request.files.get(**job.parameters)
|
|
with pytest.raises(ValueError, match='no file associated'):
|
|
assert not wcs_request_file.content.path
|
|
assert not os.path.isfile(path)
|
|
|
|
|
|
def test_add_media_wrong_uuid(app, smart):
|
|
with pytest.raises(WcsRequest.DoesNotExist):
|
|
smart.wcs_requests.get(uuid=UUID)
|
|
|
|
url = URL + 'add-media?uuid=%s' % str(UUID)
|
|
resp = app.post_json(url, params=ADD_MEDIA_PAYLOAD, status=400)
|
|
assert resp.json['err']
|
|
assert 'Cannot find intervention' in resp.json['err_desc']
|
|
assert WcsRequestFile.objects.count() == 0
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention/%s/media' % json.loads(get_json_file('create_intervention'))['id'], None, None, 500],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],
|
|
)
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_add_media_error(mocked_uuid, app, freezer, smart):
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
|
|
freezer.move_to('2021-10-30 00:00:00')
|
|
url = resp.json['data']['payload']['add_media_url']
|
|
resp = app.post_json(url, params=ADD_MEDIA_PAYLOAD)
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'registered'
|
|
|
|
freezer.move_to('2021-10-30 00:00:03')
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'registered'
|
|
assert job.update_timestamp > job.creation_timestamp
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention/%s/media' % INTERVENTION_ID, None, None, None, ReadTimeout('timeout')],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],
|
|
)
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_add_media_timeout_error(mocked_uuid, app, freezer, smart):
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
|
|
freezer.move_to('2021-10-30 00:00:00')
|
|
url = resp.json['data']['payload']['add_media_url']
|
|
resp = app.post_json(url, params=ADD_MEDIA_PAYLOAD)
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'registered'
|
|
|
|
freezer.move_to('2021-10-30 00:00:03')
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'registered'
|
|
assert job.update_timestamp > job.creation_timestamp
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention/%s/media' % INTERVENTION_ID, ADD_MEDIA_QUERY, 200],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, '400 Client Error', 400],
|
|
)
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_add_media_with_create_intervention_failure(mocked_uuid, app, smart):
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
|
|
url = resp.json['data']['payload']['add_media_url']
|
|
url = URL + 'add-media?uuid=%s' % str(UUID)
|
|
resp = app.post_json(url, params=ADD_MEDIA_PAYLOAD)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['uuid'] == str(UUID)
|
|
assert resp.json['data']['nb_registered'] == 1
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'registered'
|
|
|
|
# simulate failure on intervention creation
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
wcs_request.status = 'failed'
|
|
wcs_request.save()
|
|
wcs_request_file = wcs_request.files.get(**job.parameters)
|
|
path = wcs_request_file.content.path
|
|
assert os.path.isfile(path)
|
|
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'failed'
|
|
assert 'related wcs request failed' in job.status_details['error_summary']
|
|
|
|
|
|
UPDATE_INTERVENTION_QUERY_ON_ASYNC_CREATION = {
|
|
'creation_response': {
|
|
'wcs_form_api_url': CREATE_INTERVENTION_PAYLOAD_EXTRA['form_api_url'],
|
|
'wcs_form_number': CREATE_INTERVENTION_PAYLOAD_EXTRA['external_number'],
|
|
'uuid': str(UUID),
|
|
'payload': CREATE_INTERVENTION_QUERY,
|
|
'result': json.loads(get_json_file('create_intervention')),
|
|
'status': 'sent',
|
|
'tries': 1,
|
|
}
|
|
}
|
|
UPDATE_INTERVENTION_QUERY_ON_ASYNC_CREATION['creation_response']['result'][
|
|
'interventionCreated'
|
|
] = '2021-07-07T14:19:31.302000'
|
|
UPDATE_INTERVENTION_QUERY_ON_ASYNC_CREATION['creation_response']['result'][
|
|
'interventionDesired'
|
|
] = '2021-06-30T18:08:05'
|
|
|
|
|
|
@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_ON_ASYNC_CREATION,
|
|
WCS_RESPONSE_SUCCESS,
|
|
],
|
|
)
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_create_intervention_async(mocked_uuid4, app, smart, wcs_service):
|
|
mocked_wcs_request_push = mock.patch(
|
|
'passerelle.contrib.toulouse_smart.models.WcsRequest.push',
|
|
return_value=False,
|
|
)
|
|
mocked_smart_request_push = mock.patch(
|
|
'passerelle.contrib.toulouse_smart.models.SmartRequest.push',
|
|
return_value=False,
|
|
)
|
|
|
|
# smart and wcs are down
|
|
mocked_wcs_request_push.start()
|
|
mocked_smart_request_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['notification_url']
|
|
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'
|
|
|
|
assert Job.objects.count() == 1
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'registered'
|
|
|
|
# smart is up
|
|
mocked_wcs_request_push.stop()
|
|
smart.jobs()
|
|
assert Job.objects.count() == 2
|
|
|
|
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['id'] == INTERVENTION_ID
|
|
assert wcs_request.status == 'sent'
|
|
|
|
job = Job.objects.get(method_name='update_intervention_job')
|
|
assert job.status == 'registered'
|
|
|
|
# wcs is up
|
|
mocked_smart_request_push.stop()
|
|
smart.jobs()
|
|
|
|
job = Job.objects.get(method_name='update_intervention_job')
|
|
assert job.status == 'completed'
|
|
smart_request = wcs_request.smart_requests.get()
|
|
assert smart_request.payload['creation_response']['uuid'] == str(UUID)
|
|
assert smart_request.payload['creation_response']['result']['id'] == INTERVENTION_ID
|
|
assert smart_request.payload['creation_response']['status'] == 'sent'
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention/%s/media' % INTERVENTION_ID, ADD_MEDIA_QUERY, 200],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],
|
|
)
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_add_media_async(mocked_uuid4, app, smart, freezer):
|
|
mocked_wcs_request_push = mock.patch(
|
|
'passerelle.contrib.toulouse_smart.models.WcsRequest.push',
|
|
return_value=False,
|
|
)
|
|
|
|
# smart is down
|
|
freezer.move_to('2021-10-30 00:00:00')
|
|
mocked_wcs_request_push.start()
|
|
resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert not resp.json['err']
|
|
url = resp.json['data']['payload']['add_media_url']
|
|
resp = app.post_json(url, params=ADD_MEDIA_PAYLOAD)
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'registered'
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'registered'
|
|
assert str(job.after_timestamp) == '2021-10-30 00:10:00+00:00'
|
|
|
|
# smart is up
|
|
freezer.move_to('2021-10-30 00:00:03')
|
|
mocked_wcs_request_push.stop()
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='create_intervention_job')
|
|
assert job.status == 'completed'
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'registered'
|
|
|
|
# 10 minutes later
|
|
freezer.move_to('2021-10-30 00:10:03')
|
|
smart.jobs()
|
|
job = Job.objects.get(method_name='add_media_job')
|
|
assert job.status == 'completed'
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', None, get_json_file('create_intervention')],
|
|
)
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_create_intervention_multiple_step(mocked_uuid4, app, smart):
|
|
app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
assert smart.wcs_requests.get(uuid=UUID, wcs_form_step='initial')
|
|
|
|
NEW_UUID = uuid.UUID('a' * 32)
|
|
mocked_uuid4.return_value = NEW_UUID
|
|
app.post_json(
|
|
URL + 'create-intervention/', params=dict(CREATE_INTERVENTION_PAYLOAD, form_step='reclamation-1')
|
|
)
|
|
assert smart.wcs_requests.get(uuid=NEW_UUID, wcs_form_step='reclamation-1')
|
|
|
|
|
|
def test_pk_change_migration(migration):
|
|
old_apps = migration.before(
|
|
[('toulouse_smart', '0005_auto_20220105_1514'), ('base', '0029_auto_20210202_1627')]
|
|
)
|
|
|
|
Job = old_apps.get_model('base', 'Job')
|
|
ContentType = old_apps.get_model('contenttypes', 'ContentType')
|
|
ToulouseSmartResource = old_apps.get_model('toulouse_smart', 'ToulouseSmartResource')
|
|
WcsRequest = old_apps.get_model('toulouse_smart', 'WcsRequest')
|
|
WcsRequestFile = old_apps.get_model('toulouse_smart', 'WcsRequestFile')
|
|
SmartRequest = old_apps.get_model('toulouse_smart', 'SmartRequest')
|
|
|
|
tsr = ToulouseSmartResource.objects.create(slug='a', description='a')
|
|
ct = ContentType.objects.get(app_label='toulouse_smart', model='toulousesmartresource')
|
|
wcs_request = WcsRequest.objects.create(
|
|
resource=tsr, wcs_form_api_url='https://example.com/1/', wcs_form_number='1'
|
|
)
|
|
WcsRequestFile.objects.create(resource=wcs_request, filename='a', content_type='a', content='a/b')
|
|
SmartRequest.objects.create(resource=wcs_request, payload={}, result={})
|
|
job = Job.objects.create(
|
|
resource_type=ct,
|
|
resource_pk=tsr.pk,
|
|
method_name='create_intervention_job',
|
|
parameters={
|
|
'pk': wcs_request.pk,
|
|
},
|
|
)
|
|
|
|
apps = migration.apply([('toulouse_smart', '0012_migrate_jobs')])
|
|
|
|
Job = apps.get_model('base', 'Job')
|
|
ContentType = apps.get_model('contenttypes', 'ContentType')
|
|
ToulouseSmartResource = apps.get_model('toulouse_smart', 'ToulouseSmartResource')
|
|
WcsRequest = apps.get_model('toulouse_smart', 'WcsRequest')
|
|
WcsRequestFile = apps.get_model('toulouse_smart', 'WcsRequestFile')
|
|
|
|
wcs_request = WcsRequest.objects.get()
|
|
job = Job.objects.get()
|
|
assert job.parameters['pk'] == str(wcs_request.uuid)
|
|
assert WcsRequest.objects.get(pk=job.parameters['pk'])
|
|
assert wcs_request.files.get().resource_id == wcs_request.uuid
|
|
assert wcs_request.smart_requests.get().resource_id == wcs_request.uuid
|
|
|
|
|
|
CREATE_INTERNVENTION_WITH_NONE = json.dumps(
|
|
dict(json.loads(get_json_file('create_intervention')), interventionCreated=None)
|
|
)
|
|
|
|
|
|
@mock_response(
|
|
['/v1/type-intervention', None, INTERVENTION_TYPES],
|
|
['/v1/intervention', CREATE_INTERVENTION_QUERY, CREATE_INTERNVENTION_WITH_NONE],
|
|
)
|
|
@mock.patch('django.db.models.fields.UUIDField.get_default', return_value=UUID)
|
|
def test_create_intervention_none_dates(mocked_uuid4, app, smart):
|
|
app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
|
|
wcs_request = smart.wcs_requests.get(uuid=UUID)
|
|
assert wcs_request.result['interventionCreated'] is None
|
|
assert wcs_request.result['interventionDesired'] == '2021-06-30T18:08:05'
|
|
assert wcs_request.status == 'sent'
|