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

Nicolas Roche, 19 juillet 2021 10:22

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

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

 .../migrations/0002_auto_20210718_1258.py     |  63 +++++
 passerelle/contrib/toulouse_smart/models.py   | 113 +++++++++
 passerelle/contrib/toulouse_smart/schemas.py  | 125 ++++++++++
 .../toulouse_smart/create_intervention.json   |  54 +++++
 tests/test_toulouse_smart.py                  | 218 +++++++++++++++++-
 5 files changed, 572 insertions(+), 1 deletion(-)
 create mode 100644 passerelle/contrib/toulouse_smart/migrations/0002_auto_20210718_1258.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-18 10:58
     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_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'),
                                 ('updating', 'Updating'),
                                 ('failed', 'Failed'),
                                 ('updated', 'Updated'),
                             ],
                             default='received',
                             max_length=20,
                         ),
                     ),
                     ('trigger', models.CharField(blank=True, max_length=64, null=True, verbose_name='Trigger')),
+                    (
                         '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.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.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
-...
             },
+        )
         def get_intervention(self, request, id):
             url = self.webservice_base_url + 'v1/intervention/%s' % id
             response = self.request(url)
             doc = ET.fromstring(response.content)
             return {'data': xml.to_json(doc)}
         @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']
             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):
                     data[prop['name']] = cast[prop['type']](block[name])
                 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_url': post_data['form_url'],
                     'wcs_form_number': post_data['external_number'],
+                }
+            )
             if wcs_request.status != 'received':
                 raise APIError(
                     "'%s' intervention already created: '%s' status"
                     % (
                         post_data['external_number'],
                         wcs_request.status,
+                    )
+                )
             notificationUrl = 'update-intervention?uuid=%s' % wcs_request.uuid
             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': notificationUrl,
+            }
             url = self.webservice_base_url + 'v1/intervention'
             response = self.request(url, json=payload)
             try:
                 result = response.json()
             except ValueError:
                 raise APIError('invalid json, got: %s' % response.text)
             wcs_request.payload = payload
             wcs_request.result = result
             wcs_request.status = 'sent'
             wcs_request.save()
             return {'data': result, 'notificationUrl': notificationUrl}
     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_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(_('Status'), default='received', max_length=64)
         status = models.CharField(
             max_length=20,
             default='received',
             choices=(
                 ('registered', _('Registered')),
                 ('sent', _('Sent')),
                 ('updating', _('Updating')),
                 ('failed', _('Failed')),
                 ('updated', _('Updated')),
             ),
+        )
         trigger = models.CharField(_('Trigger'), blank=True, null=True, max_length=64)

     # 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_url': {
                 'description': "L'adresse vers la vue (front-office) du formulaire : {{ form_url }}",
                 'type': 'string',
             },
         },
         'required': [
             'slug',
             'description',
             'lat',
             'lon',
             'interventionCreated',
             'interventionDesired',
             'submitterFirstName',
             'submitterLastName',
             'submitterMail',
             'submitterPhone',
             'submitterAddress',
             'submitterType',
             'external_number',
             'external_status',
             'cityId',
             'address',
         ],
         '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.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,
-...
             @httmock.urlmatch()
             def error(url, request):
                 assert False, 'request to %s' % url.geturl()
             def register(path, payload, content, status_code=200):
                 @httmock.urlmatch(path=path)
                 def handler(url, request):
                     if payload and json.loads(request.body) != payload:
                         import pdb
                         pdb.set_trace()
                         assert False, 'wrong payload sent to request to %s' % url.geturl()
                     return httmock.response(status_code, content)
                 return handler
             @functools.wraps(func)
             def wrapper(*args, **kwargs):
                 handlers = []
-...
         return decorator
     def get_xml_file(filename):
         with open(os.path.join(TEST_BASE_DIR, "%s.xml" % filename), 'rb') 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_url': 'https://wcs.example.com/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']['id'] == '96bc8712-21a6-4fba-a511-740d6f1bd0bc'
         wcs_request = smart.wcs_requests.get(uuid=UUID)
         assert wcs_request.wcs_form_url == 'https://wcs.example.com/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'
         # cannot be done twice
         resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
         assert resp.json['err']
         assert 'already created' in resp.json['err_desc']
         # except for error recovery
         wcs_request.status = 'received'
         wcs_request.save()
         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 == '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_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_error_status(mocked_uuid, app, smart):
         resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
         assert resp.json['err']
         assert 'failed to post' in resp.json['err_desc']
         assert smart.wcs_requests.count() == 1
         wcs_request = smart.wcs_requests.get(uuid=UUID)
         assert wcs_request.status == 'received'
     @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_error_content(mocked_uuid, app, smart):
         resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
         assert resp.json['err']
         assert 'invalid json' in resp.json['err_desc']
         assert smart.wcs_requests.count() == 1
         wcs_request = smart.wcs_requests.get(uuid=UUID)
         assert wcs_request.status == 'received'
         # same uuid is re-used after an error
         resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
         assert smart.wcs_requests.count() == 1
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

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