chrono/api/urls.py

@@ -14,7 +14,7 @@
 # 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/>.
 
-from django.urls import path, re_path
+from django.urls import include, path, re_path
 
 from . import views
 
@@ -150,4 +150,5 @@ urlpatterns = [
     ),
     path('statistics/', views.statistics_list, name='api-statistics-list'),
     path('statistics/bookings/', views.bookings_statistics, name='api-statistics-bookings'),
+    path('ants/', include('chrono.apps.ants_hub.api_urls')),
 ]

chrono/apps/ants_hub/api_urls.py

@@ -0,0 +1,23 @@
+# chrono - agendas system
+# Copyright (C) 2023  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/>.
+
+from django.urls import path
+
+from . import views
+
+urlpatterns = [
+    path('check-duplicate/', views.CheckDuplicateAPI.as_view(), name='api-ants-check-duplicate'),
+]

chrono/apps/ants_hub/hub.py

@@ -16,6 +16,7 @@
 
 import requests
 from django.conf import settings
+from django.utils.translation import gettext as _
 from requests.adapters import HTTPAdapter
 from urllib3.util.retry import Retry
 
@@ -71,3 +72,17 @@ def push_rendez_vous_disponibles(payload):
         return True
     except (TypeError, KeyError, requests.RequestException) as e:
         raise AntsHubException(str(e))
+
+
+def check_duplicate(identifiants_predemande: list):
+    params = [
+        ('identifiant_predemande', identifiant_predemande)
+        for identifiant_predemande in identifiants_predemande
+    ]
+    session = make_http_session()
+    try:
+        response = session.get(make_url('rdv-status/'), params=params)
+        response.raise_for_status()
+        return response.json()
+    except (ValueError, requests.RequestException) as e:
+        return {'err': 1, 'err_desc': f'ANTS hub is unavailable: {e!r}'}

chrono/apps/ants_hub/views.py

@@ -14,14 +14,21 @@
 # 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 re
 import sys
 
 from django import forms
+from django.conf import settings
 from django.contrib import messages
 from django.core.cache import cache
 from django.shortcuts import get_object_or_404, redirect
 from django.utils.translation import gettext_lazy as _
+from django.utils.translation import gettext_noop as N_
 from django.views.generic import CreateView, DeleteView, ListView, TemplateView, UpdateView
+from rest_framework import permissions
+from rest_framework.views import APIView
+
+from chrono.api.utils import APIErrorBadRequest, Response
 
 from . import hub, models
 
@@ -249,3 +256,33 @@ class Synchronize(TemplateView):
             ants_hub_city_push.spool(domain=getattr(tenant, 'domain_url', None))
         else:
             models.City.push()
+
+
+class CheckDuplicateAPI(APIView):
+    permission_classes = (permissions.IsAuthenticated,)
+
+    identifiant_predemande_re = re.compile(r'^[A-Z0-9]{10}$')
+
+    def post(self, request):
+        if not settings.CHRONO_ANTS_HUB_URL:
+            raise APIErrorBadRequest(N_('CHRONO_ANTS_HUB_URL is not configured'))
+
+        data = request.data if isinstance(request.data, dict) else {}
+        identifiant_predemande = data.get('identifiant_predemande', request.GET.get('identifiant_predemande'))
+        identifiants_predemande = identifiant_predemande or []
+
+        if isinstance(identifiants_predemande, str):
+            identifiants_predemande = identifiants_predemande.split(',')
+
+        if not isinstance(identifiants_predemande, list):
+            raise APIErrorBadRequest(
+                N_('identifiant_predemande must be a list of identifiants separated by commas: %s'),
+                repr(identifiants_predemande),
+            )
+
+        identifiants_predemande = list(filter(None, map(str.upper, map(str.strip, identifiants_predemande))))
+
+        if not identifiants_predemande:
+            return Response({'err': 0, 'data': {'accept_rdv': True}})
+
+        return Response(hub.check_duplicate(identifiants_predemande))

tests/ants_hub/test_api.py

@@ -0,0 +1,85 @@
+# chrono - agendas system
+# Copyright (C) 2023  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/>.
+
+from unittest import mock
+
+import pytest
+import requests
+import responses
+from django.contrib.auth import get_user_model
+
+from chrono.apps.ants_hub.hub import AntsHubException, check_duplicate, ping, push_rendez_vous_disponibles
+
+User = get_user_model()
+
+
+def test_authorization(app, user):
+    app.post('/api/ants/check-duplicate/', status=401)
+
+
+@pytest.fixture
+def user(db):
+    user = User(username='john.doe', first_name='John', last_name='Doe', email='john.doe@example.net')
+    user.set_password('password')
+    user.save()
+    return user
+
+
+@pytest.fixture
+def auth_app(user, app):
+    app.authorization = ('Basic', ('john.doe', 'password'))
+    return app
+
+
+class TestCheckDuplicateAPI:
+    def test_not_configured(self, auth_app):
+        resp = auth_app.post('/api/ants/check-duplicate/', status=400)
+        assert resp.json == {
+            'err': 1,
+            'err_class': 'CHRONO_ANTS_HUB_URL is not configured',
+            'err_desc': 'CHRONO_ANTS_HUB_URL is not configured',
+            'reason': 'CHRONO_ANTS_HUB_URL is not configured',
+        }
+
+    def test_input_empty(self, hub, auth_app):
+        resp = auth_app.post('/api/ants/check-duplicate/')
+        assert resp.json == {'data': {'accept_rdv': True}, 'err': 0}
+
+    @mock.patch('chrono.apps.ants_hub.hub.check_duplicate')
+    def test_proxy(self, check_duplicate_mock, hub, auth_app):
+        # do not care about output
+        check_duplicate_mock.return_value = {'err': 0, 'data': {'xyz': '1234'}}
+
+        # GET param
+        resp = auth_app.post('/api/ants/check-duplicate/?identifiant_predemande= ABCdE12345, ,1234567890 ')
+        assert resp.json == {'err': 0, 'data': {'xyz': '1234'}}
+        assert check_duplicate_mock.call_args[0][0] == ['ABCDE12345', '1234567890']
+
+        # JSON payload as string
+        resp = auth_app.post_json(
+            '/api/ants/check-duplicate/?identifiant_predemande=XYZ',
+            params={'identifiant_predemande': ' XBCdE12345, ,1234567890 '},
+        )
+        assert resp.json == {'err': 0, 'data': {'xyz': '1234'}}
+        assert check_duplicate_mock.call_args[0][0] == ['XBCDE12345', '1234567890']
+
+        # JSON payload as list
+        resp = auth_app.post_json(
+            '/api/ants/check-duplicate/?identifiant_predemande=XYZ',
+            params={'identifiant_predemande': [' YBCdE12345', ' ', '1234567890 ']},
+        )
+        assert resp.json == {'err': 0, 'data': {'xyz': '1234'}}
+        assert check_duplicate_mock.call_args[0][0] == ['YBCDE12345', '1234567890']

tests/ants_hub/test_hub.py

@@ -18,7 +18,7 @@ import pytest
 import requests
 import responses
 
-from chrono.apps.ants_hub.hub import AntsHubException, ping, push_rendez_vous_disponibles
+from chrono.apps.ants_hub.hub import AntsHubException, check_duplicate, ping, push_rendez_vous_disponibles
 
 
 def test_ping_timeout(hub):
@@ -67,3 +67,37 @@ def test_push_rendez_vous_disponibles_application_error(hub):
     )
     with pytest.raises(AntsHubException, match='overload'):
         push_rendez_vous_disponibles({})
+
+
+class TestCheckDuplicate:
+    def test_status_500(self, hub):
+        hub.add(responses.GET, 'https://toto:@ants-hub.example.com/api/chrono/rdv-status/', status=500)
+        assert check_duplicate(['A' * 10, '1' * 10]) == {
+            'err': 1,
+            'err_desc': "ANTS hub is unavailable: HTTPError('500 Server Error: Internal Server Error for url: https://toto:@ants-hub.example.com/api/chrono/rdv-status/?identifiant_predemande=AAAAAAAAAA&identifiant_predemande=1111111111')",
+        }
+
+    def test_timeout(self, hub):
+        hub.add(
+            responses.GET,
+            'https://toto:@ants-hub.example.com/api/chrono/rdv-status/',
+            body=requests.Timeout('boom!'),
+        )
+        assert check_duplicate(['A' * 10, '1' * 10]) == {
+            'err': 1,
+            'err_desc': "ANTS hub is unavailable: Timeout('boom!')",
+        }
+
+    def test_ok(self, hub):
+        hub.add(
+            responses.GET,
+            'https://toto:@ants-hub.example.com/api/chrono/rdv-status/?identifiant_predemande=AAAAAAAAAA&identifiant_predemande=1111111111',
+            json={
+                'err': 0,
+                'data': {},
+            },
+        )
+        assert check_duplicate(['A' * 10, '1' * 10]) == {
+            'err': 0,
+            'data': {},
+        }