0001-toulouse_smart-allow-to-redo-a-failed-intervention-6.patch

Nicolas Roche, 21 février 2022 12:53

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

Subject: [PATCH] toulouse_smart: allow to redo a failed intervention (#62012)

 passerelle/contrib/toulouse_smart/models.py | 56 ++++++++++-----------
 tests/test_toulouse_smart.py                | 15 ++++--
 2 files changed, 38 insertions(+), 33 deletions(-)

                     except ValueError:
                         raise APIError(
                             "cannot cast '%s' field to %s : '%s'" % (varname, cast[prop['type']], block[varname]),
                             http_status=400,
+                        )
                 elif prop['required']:
                     raise APIError("'%s' field is required on '%s' block" % (varname, slug), http_status=400)
             if self.wcs_requests.filter(wcs_form_api_url=post_data['form_api_url']):
                 raise APIError(
                     "'%s' intervention already created" % post_data['external_number'],
                     http_status=400,
             try:
                 wcs_request = self.wcs_requests.get(wcs_form_api_url=post_data['form_api_url'])
             except WcsRequest.DoesNotExist:
                 wcs_request = self.wcs_requests.create(
                     wcs_form_api_url=post_data['form_api_url'],
                     wcs_form_number=post_data['external_number'],
+                )
             wcs_request = self.wcs_requests.create(
                 wcs_form_api_url=post_data['form_api_url'],
                 wcs_form_number=post_data['external_number'],
+            )
             else:
                 if wcs_request.status != 'failed':
                     return {'data': wcs_request.reply}
                 wcs_request.tries = 0
                 wcs_request.status = 'registered'
                 wcs_request.result = None
             endpoint_url = {}
             for endpoint_name in 'update-intervention', 'add-media':
                 endpoint_url[endpoint_name] = request.build_absolute_uri(
                     reverse(
                         'generic-endpoint',
                         kwargs={'connector': 'toulouse-smart', 'endpoint': endpoint_name, 'slug': self.slug},
+                    )
+                )
-...
                 wcs_request.payload[label] = utils.localtz_to_utc(post_data[label])
             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,
+                )
             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,
                     'tries': wcs_request.tries,
+                }
+            }
             return {'data': wcs_request.reply}
         def create_intervention_job(self, *args, **kwargs):
             wcs_request = self.wcs_requests.get(pk=kwargs['pk'])
             after_timestamp = None
             if not wcs_request.push():
                 if wcs_request.tries < 5:
                     if wcs_request.tries == 3:
                         after_timestamp = datetime.timedelta(hours=1)
                     if wcs_request.tries == 4:
                         after_timestamp = datetime.timedelta(days=1)
                 else:
                     wcs_request.status = 'failed'
                     wcs_request.save()
             payload = {
                 'creation_response': {
                     'wcs_form_api_url': wcs_request.wcs_form_api_url,
                     'wcs_form_number': wcs_request.wcs_form_number,
                     'uuid': str(wcs_request.uuid),
                     'payload': wcs_request.payload,
                     'result': wcs_request.result,
                     'status': wcs_request.status,
                     'tries': wcs_request.tries,
+                }
+            }
             payload = {'creation_response': wcs_request.reply}
             smart_request = wcs_request.smart_requests.create(payload=payload)
             self.add_job(
                 'update_intervention_job',
                 id=smart_request.id,
                 natural_id='smart-request-%s' % smart_request.id,
+            )
             if wcs_request.status == 'registered':
                 raise SkipJob(after_timestamp)
-...
                 self.save()
                 return False
             for label in 'interventionCreated', 'interventionDesired':
                 self.result[label] = utils.utc_to_localtz(self.result[label])
             self.status = 'sent'
             self.save()
             return True
         @property
         def reply(self):
             return {
                 'wcs_form_api_url': self.wcs_form_api_url,
                 'wcs_form_number': self.wcs_form_number,
                 'uuid': str(self.uuid),
                 'payload': self.payload,
                 'result': self.result,
                 'status': self.status,
                 'tries': self.tries,
+            }
     def upload_to(wcs_request_file, filename):
         instance = wcs_request_file.resource.resource
         uuid = wcs_request_file.resource.uuid
         return '%s/%s/%s/%s' % (instance.get_connector_slug(), instance.slug, uuid, filename)
     class WcsRequestFile(models.Model):

     @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']
         assert resp.json['data']['status'] == 'registered'
         assert smart.wcs_requests.count() == 1
         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']
         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):
-...
         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-sent creation query
         resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)
         wcs_request = smart.wcs_requests.get(uuid=UUID)
         assert '400 Client Error' in wcs_request.result
         assert wcs_request.tries == 1
     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',
+        }
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0001-toulouse_smart-allow-to-redo-a-failed-intervention-6.patch