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

Nicolas Roche, 09 juillet 2021 22:00

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

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

 passerelle/contrib/toulouse_smart/models.py   |  56 ++++++
 passerelle/contrib/toulouse_smart/schemas.py  | 124 +++++++++++++
 .../toulouse_smart/create_intervention.json   |  54 ++++++
 tests/test_toulouse_smart.py                  | 172 ++++++++++++++++++
 4 files changed, 406 insertions(+)
 create mode 100644 passerelle/contrib/toulouse_smart/schemas.py
 create mode 100644 tests/data/toulouse_smart/create_intervention.json

     # 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
     import lxml.etree as ET
     from django.contrib.postgres.fields import JSONField
     from django.db import models
     from django.urls import reverse
     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, utils
     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):
             form_url = post_data.pop('form_url')
             slug = post_data.pop('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]
             try:
                 block = post_data['data']['data'][0]
             except:
                 raise APIError("cannot find '%s' block field content" % slug, http_status=400)
             del post_data['data']
             data = {}
             cast = {'string': str, 'int': int, 'boolean': bool, 'item': str}
             for prop in intervention_type['properties']:
                 id = utils.make_id(slug + slugify(prop['name']))
                 if block.get(id):
                     data[prop['name']] = cast[prop['type']](block[id])
                 elif prop['required']:
                     raise APIError("'%s' field is required on '%s' block" % (id, slug), http_status=400)
             post_data['interventionData'] = data
             lon = post_data.pop('lon')
             lat = post_data.pop('lat')
             post_data['geom'] = {"type": "Point", "coordinates": [lon, lat], "crs": "EPSG:4326"}
             post_data['interventionCreated'] += 'Z'
             post_data['interventionDesired'] += 'Z'
             post_data['interventionTypeId'] = intervention_type['id']
             post_data['notificationUrl'] = reverse(
                 'generic-endpoint',
                 kwargs={'connector': self, 'slug': self.slug, 'endpoint': 'update-intervention'},
+            )
             url = self.webservice_base_url + 'v1/intervention'
             response = self.request(url, json=post_data)
             try:
                 result = response.json()
             except ValueError:
                 raise APIError('invalid json, got: %s' % response.text)
             self.set('intervention-%s' % result['id'], form_url)
             return {'data': result}
     class Cache(models.Model):
         resource = models.ForeignKey(
             verbose_name=_('Resource'),
             to=ToulouseSmartResource,
             on_delete=models.CASCADE,
             related_name='cache_entries',
+        )

     # 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',
         ],
+    }

+    {
       "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 zipfile
     from copy import deepcopy
     import httmock
     import lxml.etree as ET
     import pytest
     import utils
     from test_manager import login
     from passerelle.contrib.toulouse_smart.models import ToulouseSmartResource
-...
         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']
     BLOCK_FIELD_PAYLOAD = {
         'data': [
+            {
                 '038a8c2e-14de-4d4f-752f-496eb7fe90d7': 'Candélabre',
                 '038a8c2e-14de-4d4f-752f-496eb7fe90d7_display': 'Candélabre',
                 'e72f251a-5eef-5b78-c35a-94b549510029': '42',
+            }
         ],
         'schema': {
             '038a8c2e-14de-4d4f-752f-496eb7fe90d7': 'item',
             'e72f251a-5eef-5b78-c35a-94b549510029': 'string',
         },
+    }
     CREATE_INTERVENTION_PAYLOAD = {
         'slug': 'coin',
         'description': 'coin coin',
         'data': BLOCK_FIELD_PAYLOAD,
         '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/',
+    }
     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': '/Test/test/update-intervention',
+    }
     @mock_response(
         ['/v1/type-intervention', None, INTERVENTION_TYPES],
         ['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],
+    )
     def test_create_intervention(app, smart):
         resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
         assert not resp.json['err']
         assert resp.json['data']['id'] == '96bc8712-21a6-4fba-a511-740d6f1bd0bc'
         assert smart.get('intervention-%s' % resp.json['data']['id']) == 'https://wcs.example.com/foo/2/'
     def test_create_intervention_wrong_payload(app, smart):
         payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
         del payload['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['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['data']
         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):
         block_field_payload = {
             'data': [
+                {
                     '038a8c2e-14de-4d4f-752f-496eb7fe90d7': 'Candélabre',
                     '038a8c2e-14de-4d4f-752f-496eb7fe90d7_display': 'Candélabre',
                     'e72f251a-5eef-5b78-c35a-94b549510029': None,
+                }
             ],
             'schema': {
                 '038a8c2e-14de-4d4f-752f-496eb7fe90d7': 'item',
                 'e72f251a-5eef-5b78-c35a-94b549510029': 'string',
             },
+        }
         payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
         block_field_payload = deepcopy(BLOCK_FIELD_PAYLOAD)
         block_field_payload['data'][0]['e72f251a-5eef-5b78-c35a-94b549510029'] = None
         payload['data'] = block_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):
         payload = deepcopy(CREATE_INTERVENTION_PAYLOAD)
         block_field_payload = deepcopy(BLOCK_FIELD_PAYLOAD)
         del block_field_payload['data'][0]['e72f251a-5eef-5b78-c35a-94b549510029']
         payload['data'] = block_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],
+    )
     def test_create_intervention_error_status(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']
     @mock_response(
         ['/v1/type-intervention', None, INTERVENTION_TYPES],
         ['/v1/intervention', CREATE_INTERVENTION_QUERY, 'not json content'],
+    )
     def test_create_intervention_error_content(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']
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

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