entrouvert
/
ants-hub
15
0
Fork 0
Code Issues Pull Requests 3 Releases Wiki Activity

Implement new rdv-status endpoint (#76412) 

* Use the new /api/status endpoint the "Optimisation des rendez-vous en
  mairies" to check status of some "application_ids" / "identifiants de
  pré-demande".
* Always return "err": 0 if the web-service responds, and return a
  simple "accept_rdv" boolean to test in w.c.s. conditions. It prevents
  useless logged errors,
* When accept_rdv is false, the message key contains the reason to show
  to users and an eventual list of appointments with the concerned
  identifiant_predemande:

    {
      "data": {
         "accept_rdv": false,
         "message": "Prédemande·\"ABCDE12345\"·déjà·liée·à·un·ou·des·rendez-vous.",
         "appointments": [
             {
                "appointment_date": "2023-09-20T09:00:11",
                "identifiant_predemande": "ABCDE12345",
                "management_url": "https://rdvenmairie.fr/gestion/login?ants=83AHERZY8F&appointment_id=64594c435d7bfc0012fa8c87&canceled=true",
                "meeting_point": "Mairie de Luisant",
             }
         ],
    }
This commit is contained in:
Benjamin Dauvergne 2023-08-29 18:37:28 +02:00
parent d5e6f79257
commit 2f9cd366b3
5 changed files with 419 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
src/ants_hub/api/ants.py
View File

@ -29,6 +29,8 @@ def get_api_optimisation_url():
def push_rdv(rdv):
# swagger : https://api-coordination.rendezvouspasseport.ants.gouv.fr/docs#/Appointments%20-%20%C3%A9diteurs/add_appointment_api_appointments_post
auth_token = get_api_optimisation_auth_token()
if auth_token is None:
raise ANTSError('REQUEST_TO_ANTS_V2_AUTH_TOKEN not configured')
base_url = get_api_optimisation_url()
try:
if rdv.canceled is None:
@ -87,3 +89,53 @@ def upload_rdvs():
rdvs.filter(pk__in=[rdv.pk for rdv in pushed]).update(last_upload=start)
for rdv in pushed:
logger.info('pushed rdv(%s) of lieu %s', rdv, rdv.lieu)
def get_status_of_predemandes(identifiant_predemandes: list):
auth_token = get_api_optimisation_auth_token()
base_url = get_api_optimisation_url()
params = [
('application_ids', identifiant_predemande) for identifiant_predemande in identifiant_predemandes
]
try:
response = requests.get(
base_url + 'status',
params=params,
headers={'x-rdv-opt-auth-token': auth_token},
timeout=10,
)
response.raise_for_status()
data = response.json()
if not isinstance(data, dict):
raise ValueError(f'reponse is not a dict: {data!r}')
except (requests.RequestException, ValueError, TypeError) as e:
raise ANTSError(repr(e))
valid = True
msg = []
appointments = []
for identifiant_predemande in identifiant_predemandes:
if identifiant_predemande not in data:
msg.append(f'Prédemande "{identifiant_predemande}" inconnue.')
valid = False
continue
predemande_state = data[identifiant_predemande]
if not isinstance(predemande_state, dict):
raise ANTSError(f'application_id state is not a dict: {data!r}')
if predemande_state.get('status') != 'validated':
msg.append(f'Prédemande "{identifiant_predemande}" inconnue, expirée ou déjà consommée.')
valid = False
continue
if predemande_state.get('appointments', []):
appointments.extend(
[
{'identifiant_predemande': identifiant_predemande, **appointment}
for appointment in predemande_state['appointments']
]
)
msg.append(f'Prédemande "{identifiant_predemande}" déjà liée à un ou des rendez-vous.')
valid = False
continue
return valid, ' '.join(msg), data, appointments

1
src/ants_hub/api/urls.py
View File

@ -16,4 +16,5 @@ urlpatterns += [
path('chrono/ping/', chrono.ping),
path('chrono/rendez-vous-disponibles/', chrono.rendez_vous_disponibles),
path('chrono/predemandes/', chrono.predemandes),
path('chrono/rdv-status/', chrono.rdv_status),
]

78
src/ants_hub/api/views/chrono.py
View File

@ -8,6 +8,7 @@ import functools
import json
import logging
import os
import re
import time
import jsonschema
@ -21,6 +22,7 @@ from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.generic import View
from ants_hub.api.ants import ANTSError, get_status_of_predemandes
from ants_hub.data.models import (
Collectivite,
Config,
@ -505,3 +507,79 @@ class PredemandesView(View):
predemandes = csrf_exempt(authenticate(PredemandesView.as_view()))
class RdvStatus(View):
def get(self, request):
identifiant_predemandes = [
part.strip().upper() for part in request.GET.getlist('identifiant_predemande') if part.strip()
]
if not identifiant_predemandes:
return JsonResponse(
{
'err': 0,
'data': {
'accept_rdv': True,
},
}
)
msg = []
for identifiant_predemande in identifiant_predemandes:
if not re.match(r'^([A-Z0-9]{10}[,;:\-/.\s])*[A-Z0-9]{10}$', identifiant_predemande):
msg.append(
f'Identifiant de pré-demande "{identifiant_predemande}" invalide, il doit faire 10 caractères.'
)
if msg:
logger.info(
'rdv-status for application_ids %s is nok: %s',
','.join(identifiant_predemandes),
' '.join(msg),
)
return JsonResponse(
{
'err': 0,
'data': {
'accept_rdv': False,
'message': ' '.join(msg),
'ants_response': {},
},
}
)
try:
accept_rdv, message, ants_response, appointments = get_status_of_predemandes(
identifiant_predemandes
)
except ANTSError as e:
logger.warning(
'could not check status of rdv for identifiants_predemandes(%s): %s',
identifiant_predemandes,
e,
)
return JsonResponse(
{
'err': 1,
'data': str(e),
}
)
if accept_rdv:
logger.info('rdv-status for application_ids %s is ok.', ','.join(identifiant_predemandes))
else:
logger.info(
'rdv-status for application_ids %s is nok: %s', ','.join(identifiant_predemandes), msg
)
return JsonResponse(
{
'err': 0,
'data': {
'accept_rdv': accept_rdv,
'message': message,
'ants_response': ants_response,
'appointments': appointments,
},
}
)
rdv_status = csrf_exempt(authenticate(RdvStatus.as_view()))

157
tests/api/test_ants.py
View File

@ -8,6 +8,7 @@ import responses
import responses.matchers
from django.core.management import call_command
from ants_hub.api.ants import ANTSError, get_status_of_predemandes
from ants_hub.data.models import Config, Lieu, Plage, RendezVous, TypeDeRdv
from ants_hub.timezone import now
@ -513,3 +514,159 @@ class TestAPIV2Push:
assert post_response.call_count == 1
assert RendezVous.objects.count() == 1
assert RendezVous.objects.filter(last_upload__isnull=True).count() == 1
class TestGetStatusOfPredemandes:
@pytest.fixture(autouse=True)
def setup(self, db, settings, django_app, freezer):
Config.set(Config.REQUEST_TO_ANTS_V2_AUTH_TOKEN, 'abcd')
@responses.activate
def test_valid(self, db):
document = {
'ABCDE12345': {
'status': 'validated',
'appointments': [],
},
'1234567890': {
'status': 'validated',
'appointments': [],
},
}
responses.add(
responses.GET,
'https://api-coordination.rendezvouspasseport.ants.gouv.fr/api/status',
json=document,
status=200,
match=[
responses.matchers.query_string_matcher(
'application_ids=ABCDE12345&application_ids=1234567890'
),
responses.matchers.header_matcher({'x-rdv-opt-auth-token': 'abcd'}),
],
)
valid, msg, data, appointments = get_status_of_predemandes(['ABCDE12345', '1234567890'])
assert valid
assert msg == ''
assert data == document
assert not appointments
@responses.activate
@pytest.mark.parametrize(
'document,expected_message,expected_appointments',
[
(
{
'ABCDE12345': {
'status': 'unknown',
'appointments': [],
},
'1234567890': {
'status': 'validated',
'appointments': [],
},
},
'Prédemande "ABCDE12345" inconnue, expirée ou déjà consommée.',
[],
),
(
{
'1234567890': {
'status': 'validated',
'appointments': [],
},
},
'Prédemande "ABCDE12345" inconnue.',
[],
),
(
{
'ABCDE12345': {
'status': 'validated',
'appointments': [],
},
'1234567890': {
'status': 'validated',
'appointments': [
{
'management_url': 'https://rdvenmairie.fr/gestion/login?ants=83AHERZY8F&appointment_id=64594c435d7bfc0012fa8c87&canceled=true',
'meeting_point': 'Mairie de Luisant',
'appointment_date': '2023-09-20T09:00:11',
}
],
},
},
'Prédemande "1234567890" déjà liée à un ou des rendez-vous.',
[
{
'identifiant_predemande': '1234567890',
'management_url': 'https://rdvenmairie.fr/gestion/login?ants=83AHERZY8F&appointment_id=64594c435d7bfc0012fa8c87&canceled=true',
'meeting_point': 'Mairie de Luisant',
'appointment_date': '2023-09-20T09:00:11',
}
],
),
],
ids=[
'one status is unknown',
'missing application_id',
'appointments exists',
],
)
def test_invalid(self, document, expected_message, expected_appointments, db):
responses.add(
responses.GET,
'https://api-coordination.rendezvouspasseport.ants.gouv.fr/api/status',
json=document,
status=200,
match=[
responses.matchers.query_string_matcher(
'application_ids=ABCDE12345&application_ids=1234567890'
),
responses.matchers.header_matcher({'x-rdv-opt-auth-token': 'abcd'}),
],
)
valid, msg, data, appointments = get_status_of_predemandes(['ABCDE12345', '1234567890'])
assert not valid
assert msg == expected_message
assert data == document
assert appointments == expected_appointments
@responses.activate
@pytest.mark.parametrize(
'response_kwargs',
[
{
'json': None,
'status': 200,
},
{
'json': '',
'status': 200,
},
{
'status': 500,
},
],
ids=[
'JSON is null',
'JSON is string',
'HTTP 500',
],
)
def test_error(self, response_kwargs, db):
responses.add(
responses.GET,
'https://api-coordination.rendezvouspasseport.ants.gouv.fr/api/status',
match=[
responses.matchers.query_string_matcher(
'application_ids=ABCDE12345&application_ids=1234567890'
),
responses.matchers.header_matcher({'x-rdv-opt-auth-token': 'abcd'}),
],
**response_kwargs,
)
with pytest.raises(ANTSError):
get_status_of_predemandes(['ABCDE12345', '1234567890'])

131
tests/api/test_chrono.py
View File

@ -1238,3 +1238,134 @@ def test_rendez_vous_disponibles_database_is_busy(django_app, db, monkeypatch, s
assert call_count == 4
assert resp.json['err'] == 1
assert resp.json['error'] == 'busy'
class TestRdvStatus:
@pytest.fixture(autouse=True)
def setup(self, db, settings, django_app):
Config.set(Config.REQUEST_TO_ANTS_V2_AUTH_TOKEN, 'abcd')
Raccordement.objects.create(name='plateforme', apikey='abcd')
django_app.set_authorization(('Basic', ('abcd', '')))
@pytest.fixture
def document(self):
return {
'ABCDE12345': {
'status': 'validated',
'appointments': [],
},
'1234567890': {
'status': 'validated',
'appointments': [],
},
}
@responses.activate
def test_ok(self, db, django_app, document):
responses.add(
responses.GET,
'https://api-coordination.rendezvouspasseport.ants.gouv.fr/api/status',
json=document,
status=200,
match=[
responses.matchers.query_string_matcher(
'application_ids=ABCDE12345&application_ids=1234567890'
),
responses.matchers.header_matcher({'x-rdv-opt-auth-token': 'abcd'}),
],
)
response = django_app.get(
'/api/chrono/rdv-status/',
params=[('identifiant_predemande', 'ABCDE12345'), ('identifiant_predemande', '1234567890')],
)
assert response.json == {
'err': 0,
'data': {
'accept_rdv': True,
'message': '',
'ants_response': {
'1234567890': {'appointments': [], 'status': 'validated'},
'ABCDE12345': {'appointments': [], 'status': 'validated'},
},
'appointments': [],
},
}
assert (
responses._default_mock.calls[0].request.url
== 'https://api-coordination.rendezvouspasseport.ants.gouv.fr/api/status?application_ids=ABCDE12345&application_ids=1234567890'
)
@responses.activate
def test_nok(self, db, django_app, document):
document['ABCDE12345']['appointments'] = [
{
'management_url': 'https://rdvenmairie.fr/gestion/login?ants=83AHERZY8F&appointment_id=64594c435d7bfc0012fa8c87&canceled=true',
'meeting_point': 'Mairie de Luisant',
'appointment_date': '2023-09-20T09:00:11',
}
]
responses.add(
responses.GET,
'https://api-coordination.rendezvouspasseport.ants.gouv.fr/api/status',
json=document,
status=200,
match=[
responses.matchers.query_string_matcher(
'application_ids=ABCDE12345&application_ids=1234567890'
),
responses.matchers.header_matcher({'x-rdv-opt-auth-token': 'abcd'}),
],
)
response = django_app.get(
'/api/chrono/rdv-status/',
params=[('identifiant_predemande', 'ABCDE12345'), ('identifiant_predemande', '1234567890')],
)
assert response.json == {
'err': 0,
'data': {
'accept_rdv': False,
'ants_response': {
'1234567890': {'appointments': [], 'status': 'validated'},
'ABCDE12345': {
'appointments': [
{
'appointment_date': '2023-09-20T09:00:11',
'management_url': 'https://rdvenmairie.fr/gestion/login'
'?ants=83AHERZY8F&appointment_id=64594c435d7bfc0012fa8c87&canceled=true',
'meeting_point': 'Mairie ' 'de ' 'Luisant',
}
],
'status': 'validated',
},
},
'message': 'Prédemande "ABCDE12345" déjà liée à un ou des rendez-vous.',
'appointments': [
{
'appointment_date': '2023-09-20T09:00:11',
'identifiant_predemande': 'ABCDE12345',
'management_url': 'https://rdvenmairie.fr/gestion/login?ants=83AHERZY8F&appointment_id=64594c435d7bfc0012fa8c87&canceled=true',
'meeting_point': 'Mairie de Luisant',
}
],
},
}
@responses.activate
def test_invalid_format(self, db, django_app, document):
response = django_app.get(
'/api/chrono/rdv-status/',
params=[('identifiant_predemande', 'ABCDE12345X'), ('identifiant_predemande', '1234567890')],
)
assert response.json == {
'err': 0,
'data': {
'accept_rdv': False,
'ants_response': {},
'message': 'Identifiant de pré-demande "ABCDE12345X" invalide, il '
'doit faire 10 caractères.',
},
}