0005-toulouse_smart-add-create-intervention-endpoint-5523.patch

Nicolas Roche, 02 août 2021 16:18

Voir les différences: en ligne côte à côte

Subject: [PATCH 5/7] toulouse_smart: add create-intervention endpoint (#55230)

 .../migrations/0002_auto_20210731_0031.py     |  56 ++++
 passerelle/contrib/toulouse_smart/models.py   | 148 ++++++++-
 passerelle/contrib/toulouse_smart/schemas.py  | 126 ++++++++
 .../toulouse_smart/create_intervention.json   |  54 ++++
 tests/test_toulouse_smart.py                  | 289 +++++++++++++++++-
 5 files changed, 671 insertions(+), 2 deletions(-)
 create mode 100644 passerelle/contrib/toulouse_smart/migrations/0002_auto_20210731_0031.py
 create mode 100644 passerelle/contrib/toulouse_smart/schemas.py
 create mode 100644 tests/data/toulouse_smart/create_intervention.json

     # -*- coding: utf-8 -*-
     # Generated by Django 1.11.29 on 2021-07-30 22:31
     from __future__ import unicode_literals
     import uuid
     import django.contrib.postgres.fields.jsonb
     import django.db.models.deletion
     from django.db import migrations, models
     class Migration(migrations.Migration):
         dependencies = [
             ('toulouse_smart', '0001_initial'),
+        ]
         operations = [
             migrations.CreateModel(
                 name='WcsRequest',
                 fields=[
                     ('wcs_form_api_url', models.CharField(max_length=256, primary_key=True, serialize=False)),
                     ('wcs_form_number', models.CharField(max_length=16)),
                     ('uuid', models.UUIDField(default=uuid.uuid4, editable=False, unique=True)),
                     ('payload', django.contrib.postgres.fields.jsonb.JSONField(default=dict)),
                     ('result', django.contrib.postgres.fields.jsonb.JSONField(default=dict)),
+                    (
                         'status',
                         models.CharField(
                             choices=[('registered', 'Registered'), ('sent', 'Sent')],
                             default='registered',
                             max_length=20,
                         ),
                     ),
+                    (
                         'resource',
                         models.ForeignKey(
                             on_delete=django.db.models.deletion.CASCADE,
                             related_name='wcs_requests',
                             to='toulouse_smart.ToulouseSmartResource',
                             verbose_name='WcsRequest',
                         ),
                     ),
                 ],
             ),
             migrations.AlterField(
                 model_name='cache',
                 name='resource',
                 field=models.ForeignKey(
                     on_delete=django.db.models.deletion.CASCADE,
                     related_name='cache_entries',
                     to='toulouse_smart.ToulouseSmartResource',
                     verbose_name='Resource',
                 ),
             ),
+        ]

     # 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 datetime
     from uuid import uuid4
     import lxml.etree as ET
     from django.contrib.postgres.fields import JSONField
     from django.db import models
     from django.db.transaction import atomic
     from django.utils.text import slugify
     from django.utils.timezone import now
     from django.utils.translation import ugettext_lazy as _
     from requests import RequestException
     from passerelle.base.models import BaseResource, HTTPResource
     from passerelle.base.models import BaseResource, HTTPResource, SkipJob
     from passerelle.utils import xml
     from passerelle.utils.api import endpoint
     from passerelle.utils.jsonresponse import APIError
     from . import schemas
     class ToulouseSmartResource(BaseResource, HTTPResource):
         category = _('Business Process Connectors')
         webservice_base_url = models.URLField(_('Webservice Base URL'))
         log_requests_errors = False
-...
                 'id': {'description': _('Intervention identifier')},
             },
+        )
         def get_intervention(self, request, id):
             url = self.webservice_base_url + 'v1/intervention/%s' % id
             response = self.request(url)
             return {'data': response.json()}
         @atomic
         @endpoint(
             name='create-intervention',
             methods=['post'],
             description=_('Create an intervention'),
             perm='can_access',
             post={'request_body': {'schema': {'application/json': schemas.CREATE_SCHEMA}}},
+        )
         def create_intervention(self, request, post_data):
             slug = post_data['slug']
             wcs_form_number = post_data['external_number']
             try:
                 types = [x for x in self.get_intervention_types() if slugify(x['name']) == slug]
             except KeyError:
                 raise APIError('Service is unavailable')
             if len(types) == 0:
                 raise APIError("unknown '%s' block slug" % slug, http_status=400)
             intervention_type = types[0]
             wcs_block_varname = slugify(intervention_type['name']).replace('-', '_')
             try:
                 block = post_data['fields']['%s_raw' % wcs_block_varname][0]
             except:
                 raise APIError("cannot find '%s' block field content" % slug, http_status=400)
             data = {}
             cast = {'string': str, 'int': int, 'boolean': bool, 'item': str}
             for prop in intervention_type['properties']:
                 name = prop['name'].lower()
                 if block.get(name):
                     try:
                         data[prop['name']] = cast[prop['type']](block[name])
                     except ValueError:
                         raise APIError(
                             "cannot cast '%s' field to %s : '%s'" % (name, cast[prop['type']], block[name]),
                             http_status=400,
+                        )
                 elif prop['required']:
                     raise APIError("'%s' field is required on '%s' block" % (name, slug), http_status=400)
             wcs_request, created = self.wcs_requests.update_or_create(
                 defaults={
                     'wcs_form_api_url': post_data['form_api_url'],
                     'wcs_form_number': post_data['external_number'],
+                }
+            )
             if not created:
                 raise APIError(
                     "'%s' intervention already created: '%s' status"
                     % (
                         post_data['external_number'],
                         wcs_request.status,
                     ),
                     http_status=400,
+                )
             wcs_request = self.wcs_requests.get(wcs_form_number=wcs_form_number)
             wcs_request.payload = {
                 'description': post_data['description'],
                 'cityId': post_data['cityId'],
                 'interventionCreated': post_data['interventionCreated'] + 'Z',
                 'interventionDesired': post_data['interventionDesired'] + 'Z',
                 'submitterFirstName': post_data['submitterFirstName'],
                 'submitterLastName': post_data['submitterLastName'],
                 'submitterMail': post_data['submitterMail'],
                 'submitterPhone': post_data['submitterPhone'],
                 'submitterAddress': post_data['submitterAddress'],
                 'submitterType': post_data['submitterType'],
                 'external_number': post_data['external_number'],
                 'external_status': post_data['external_status'],
                 'address': post_data['address'],
                 'interventionData': data,
                 'geom': {
                     'type': 'Point',
                     'coordinates': [post_data['lon'], post_data['lat']],
                     'crs': 'EPSG:4326',
                 },
                 'interventionTypeId': intervention_type['id'],
                 'notificationUrl': 'update-intervention?uuid=%s' % wcs_request.uuid,
+            }
             wcs_request.save()
             if not wcs_request.push():
                 self.add_job(
                     'create_intervention_job',
                     pk=wcs_request.pk,
                     natural_id='wcs-request-%s' % wcs_request.pk,
+                )
             wcs_request = self.wcs_requests.get(wcs_form_number=wcs_form_number)
             return {
                 'data': {
                     'wcs_form_api_url': wcs_request.wcs_form_api_url,
                     'wcs_form_number': wcs_request.wcs_form_number,
                     'uuid': wcs_request.uuid,
                     'payload': wcs_request.payload,
                     'result': wcs_request.result,
                     'status': wcs_request.status,
+                }
+            }
         def create_intervention_job(self, *args, **kwargs):
             wcs_request = self.wcs_requests.get(pk=kwargs['pk'])
             if not wcs_request.push():
                 raise SkipJob()
     class Cache(models.Model):
         resource = models.ForeignKey(
             verbose_name=_('Resource'),
             to=ToulouseSmartResource,
             on_delete=models.CASCADE,
             related_name='cache_entries',
+        )
         key = models.CharField(_('Key'), max_length=64)
         timestamp = models.DateTimeField(_('Timestamp'), auto_now=True)
         value = JSONField(_('Value'), default=dict)
     class WcsRequest(models.Model):
         resource = models.ForeignKey(
             verbose_name=_('WcsRequest'),
             to=ToulouseSmartResource,
             on_delete=models.CASCADE,
             related_name='wcs_requests',
+        )
         wcs_form_api_url = models.CharField(max_length=256, primary_key=True)
         wcs_form_number = models.CharField(max_length=16)
         uuid = models.UUIDField(default=uuid4, unique=True, editable=False)
         payload = JSONField(default=dict)
         result = JSONField(default=dict)
         status = models.CharField(
             max_length=20,
             default='registered',
             choices=(
                 ('registered', _('Registered')),
                 ('sent', _('Sent')),
             ),
+        )
         def push(self):
             url = self.resource.webservice_base_url + 'v1/intervention'
             try:
                 response = self.resource.request(url, json=self.payload)
             except APIError as e:
                 self.result = str(e)
                 self.save()
                 return False
             try:
                 self.result = response.json()
             except ValueError:
                 err_desc = 'invalid json, got: %s' % response.text
                 self.result = err_desc
                 self.save()
                 return False
             self.status = 'sent'
             self.save()
             return True

     # 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/>.
     CREATE_SCHEMA = {
         '$schema': 'http://json-schema.org/draft-04/schema#',
         'type': 'object',
         'properties': {
             'slug': {
                 'description': "slug du block de champs intervention",
                 'type': 'string',
             },
             'description': {
                 'description': "Description de la demande",
                 'type': 'string',
             },
             'lat': {
                 'description': 'Latitude',
                 'type': 'number',
             },
             'lon': {
                 'description': 'Longitude',
                 'type': 'number',
             },
             'cityId': {
                 'description': 'Code INSEE de la commune',
                 'type': 'string',
             },
             'safeguardRequired': {
                 'description': 'Présence d’un danger ?',
                 'type': 'boolean',
             },
             'interventionCreated': {
                 'description': 'Date de création de la demande',
                 'type': 'string',
                 'pattern': '[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}',
             },
             'interventionDesired': {
                 'description': 'Date d’intervention souhaitée',
                 'type': 'string',
                 'pattern': '[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}',
             },
             'submitterFirstName': {
                 'description': 'Prénom du demandeur',
                 'type': 'string',
             },
             'submitterLastName': {
                 'description': 'Nom du demandeur',
                 'type': 'string',
             },
             'submitterMail': {
                 'description': 'Adresse mail du demandeur',
                 'type': 'string',
             },
             'submitterPhone': {
                 'description': 'Numéro de téléphone du demandeur',
                 'type': 'string',
             },
             'submitterAddress': {
                 'description': 'Adresse du demandeur',
                 'type': 'string',
             },
             'submitterType': {
                 'description': 'Type de demandeur (Commune, élu, usager…)',
                 'type': 'string',
             },
             'onPrivateLand': {
                 'description': 'Intervention sur le domaine privé ?',
                 'type': 'boolean',
             },
             'checkDuplicated': {
                 'description': 'Activation de la détection de doublon (laisser à false pour le moment)',
                 'type': 'boolean',
             },
             'external_number': {
                 'description': 'Numéro externe de la demande (numéro Publik : {{ form_number }})',
                 'type': 'string',
                 'pattern': '[0-9]+[0-9]+',
             },
             'external_status': {
                 'description': 'Statut externe de la demande (statut Publik : {{ form_status }})',
                 'type': 'string',
             },
             'address': {
                 'description': 'Adresse de la demande (démarche dans Publik : {{ form_backoffice_url }})',
                 'type': 'string',
             },
             'form_api_url': {
                 'description': "L'adresse vers la vue API du formulaire : {{ form_api_url }}",
                 'type': 'string',
             },
         },
         'required': [
             'slug',
             'description',
             'lat',
             'lon',
             'cityId',
             'interventionCreated',
             'interventionDesired',
             'submitterFirstName',
             'submitterLastName',
             'submitterMail',
             'submitterPhone',
             'submitterAddress',
             'submitterType',
             'external_number',
             'external_status',
             'address',
             'form_api_url',
         ],
         'merge_extra': True,
+    }

+    {
       "id": "96bc8712-21a6-4fba-a511-740d6f1bd0bc",
       "name": "DI-20210707-0031",
       "description": "truc",
       "geom": {
         "type": "Point",
         "coordinates": [
 .8546695709228525,
 .75129501117154
         ],
         "crs": "EPSG:3943"
       },
       "interventionData": {},
       "safeguardRequired": false,
       "media": null,
       "safeguardDone": null,
       "interventionCreated": "2021-07-07T12:19:31.302Z",
       "interventionDesired": "2021-06-30T16:08:05Z",
       "interventionDone": null,
       "submitter": {
         "id": null,
         "lastName": "admin",
         "firstName": "admin12",
         "mail": "admin@localhost",
         "furtherInformation": null,
         "phoneNumber": "0699999999",
         "address": ""
       },
       "submitterType": "\u00e9lu",
       "organizations": [
+        {
           "id": "f1378d8a-12bf-4c14-913f-22624b0ecab8",
           "name": "Direction des P\u00f4les"
         },
+        {
           "id": "8ad4af63-70b5-416f-a75d-c510d83ce1bd",
           "name": "Transport Logistique"
+        }
       ],
       "domain": null,
       "state": {
         "id": "e844e67f-5382-4c0f-94d8-56f618263485",
         "table": null,
         "stateLabel": "Nouveau",
         "closes": false
       },
       "interventionTypeId": "f72d370c-4d25-489f-956e-2a0d48433789",
       "onPrivateLand": false,
       "duplicates": null,
       "external_number": "174-4",
       "external_status": "statut1",
       "cityId": "12345",
       "address": "https://wcs.example.com/backoffice/management/foo/2/"
+    }

     # 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 test_manager import login
     from passerelle.contrib.toulouse_smart.models import ToulouseSmartResource
     from passerelle.base.models import Job
     from passerelle.contrib.toulouse_smart.models import 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,
-...
         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_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>
-...
     @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/',
+    }
     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': 'update-intervention?uuid=%s' % str(UUID),
+    }
     @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, 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
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0005-toulouse_smart-add-create-intervention-endpoint-5523.patch