Ajout simplifié d'une convention finalisée - étape 1: sélection de l'opération

conventions/forms/convention_form_add.py

@@ -0,0 +1,5 @@
+from django import forms
+
+
+class ConventionAddForm(forms.Form):
+ pass

conventions/services/add_from_operation.py

@@ -0,0 +1,111 @@
+from dataclasses import dataclass
+
+from django.conf import settings
+from django.contrib.postgres.search import TrigramSimilarity
+from django.db.models import QuerySet, Value
+from django.db.models.functions import Replace
+from django.http import HttpRequest
+
+from conventions.forms.convention_form_add import ConventionAddForm
+from programmes.models import NatureLogement, Programme
+from siap.exceptions import SIAPException
+from siap.siap_client.client import SIAPClient
+
+
+@dataclass
+class Operation:
+ numero: str
+ nom: str
+ bailleur: str
+ nature: str
+ commune: str
+
+ @classmethod
+ def from_siap_payload(cls, payload: dict[str, str]) -> "Operation":
+ return cls(
+ nom=payload["donneesOperation"]["nomOperation"],
+ numero=payload["donneesOperation"]["numeroOperation"],
+ nature=payload["donneesOperation"]["natureLogement"],
+ bailleur=payload["gestionnaire"]["code"],
+ commune=payload["donneesLocalisation"]["adresseComplete"]["commune"],
+ )
+
+ @classmethod
+ def from_apilos_programme(cls, programme: Programme) -> "Operation":
+ return cls(
+ numero=programme.numero_galion,
+ nom=programme.nom,
+ bailleur=programme.bailleur.nom,
+ nature=NatureLogement[programme.nature_logement].label,
+ commune=programme.ville,
+ )
+
+
+class SelectOperationService:
+ request: HttpRequest
+ numero_operation: str
+
+ def __init__(self, request: HttpRequest, numero_operation: str) -> None:
+ self.request = request
+ self.numero_operation = numero_operation
+
+ def fetch_operations(self) -> tuple[bool, list[Operation]]:
+ if self.numero_operation is None:
+ return False, []
+
+ if operation := self._fetch_siap_operation():
+ return True, [operation]
+
+ if operation := self._get_apilos_operation():
+ return True, [operation]
+
+ return False, self._get_nearby_apilos_operations()
+
+ def _fetch_siap_operation(self) -> Operation | None:
+ try:
+ return Operation.from_siap_payload(
+ payload=SIAPClient.get_instance().get_operation(
+ user_login=self.request.user.cerbere_login,
+ habilitation_id=self.request.session["habilitation_id"],
+ operation_identifier=self.numero_operation,
+ )
+ )
+ except SIAPException:
+ return None
+
+ def _user_programmes(self) -> QuerySet[Programme]:
+ return self.request.user.programmes()
+
+ def _get_apilos_operation(self) -> Operation | None:
+ qs = self._user_programmes().filter(numero_galion=self.numero_operation)
+ if qs.count() == 1:
+ return Operation.from_apilos_programme(programme=qs.first())
+ return None
+
+ def _get_nearby_apilos_operations(self) -> list[Operation]:
+ qs = (
+ self._user_programmes()
+ .annotate(
+ numero_operation_trgrm=TrigramSimilarity(
+ Replace(Replace("numero_galion", Value("/")), Value("-")),
+ self.numero_operation.replace("-", "").replace("/", ""),
+ )
+ )
+ .filter(numero_operation_trgrm__gt=settings.TRIGRAM_SIMILARITY_THRESHOLD)
+ .order_by("-numero_operation_trgrm", "-cree_le")
+ )
+ return [Operation.from_apilos_programme(programme=p) for p in qs[:10]]
+
+
+class ConventionAddService:
+ request: HttpRequest
+ form: ConventionAddForm
+
+ def __init__(self, request: HttpRequest) -> None:
+ self.request = request
+ self.form = None
+
+ def get_form(self) -> ConventionAddForm:
+ if self.form is None:
+ self.form = ConventionAddForm()
+ return self.form

conventions/tests/services/test_add_from_operation.py

@@ -0,0 +1,108 @@
+from django.test import TestCase, override_settings
+from django.test.client import RequestFactory
+from unittest_parametrize import ParametrizedTestCase, param, parametrize
+
+from conventions.services.add_from_operation import (
+ ConventionAddService,
+ Operation,
+ SelectOperationService,
+)
+from core.tests.test_utils import PGTrgmTestMixin
+from programmes.models import NatureLogement
+from programmes.tests.factories import ProgrammeFactory
+from users.tests.factories import UserFactory
+
+
+@override_settings(USE_MOCKED_SIAP_CLIENT=True)
+class TestSelectOperationService(PGTrgmTestMixin, ParametrizedTestCase, TestCase):
+ def setUp(self):
+ self.request = RequestFactory()
+ self.request.user = UserFactory(cerbere=True, is_superuser=True)
+ self.request.session = {"habilitation_id": 5}
+
+ ProgrammeFactory(
+ uuid="67062edc-3ee8-4262-965f-98f885d418f4",
+ numero_galion="2017DD01100057",
+ nom="Programme 1",
+ nature_logement=NatureLogement.LOGEMENTSORDINAIRES,
+ ville="Bayonne",
+ bailleur__nom="Bailleur A",
+ )
+ ProgrammeFactory(
+ uuid="7fb89bd6-62f8-4c06-b15a-4fc81bc02995",
+ numero_galion="2017DD01201254",
+ nom="Programme 3",
+ nature_logement=NatureLogement.RESISDENCESOCIALE,
+ ville="L'Isle-sur-la-Sorgue",
+ bailleur__nom="Bailleur B",
+ )
+
+ @parametrize(
+ "numero_operation, expected_matching, expected_operations",
+ [
+ param(
+ "20220600006",
+ True,
+ [
+ Operation(
+ numero="20220600006",
+ nom="Programme 2",
+ bailleur="13055",
+ nature="LOO",
+ commune="Marseille",
+ )
+ ],
+ id="siap_match_exact",
+ ),
+ param(
+ "2017DD01100057",
+ True,
+ [
+ Operation(
+ numero="2017DD01100057",
+ nom="Programme 1",
+ bailleur="Bailleur A",
+ nature="Logements ordinaires",
+ commune="Bayonne",
+ )
+ ],
+ id="apilos_match_exact",
+ ),
+ param(
+ "2017DD01",
+ False,
+ [
+ Operation(
+ numero="2017DD01201254",
+ nom="Programme 3",
+ bailleur="Bailleur B",
+ nature="Résidence sociale",
+ commune="L'Isle-sur-la-Sorgue",
+ ),
+ Operation(
+ numero="2017DD01100057",
+ nom="Programme 1",
+ bailleur="Bailleur A",
+ nature="Logements ordinaires",
+ commune="Bayonne",
+ ),
+ ],
+ id="apilos_search_trgrm",
+ ),
+ ],
+ )
+ def test_fetch_operations(
+ self, numero_operation, expected_matching, expected_operations
+ ):
+ service = SelectOperationService(
+ request=self.request, numero_operation=numero_operation
+ )
+ exact_match, operations = service.fetch_operations()
+ assert exact_match == expected_matching
+ assert operations == expected_operations
+
+
+class TestConventionAddService(TestCase):
+ def basic_test(self):
+ service = ConventionAddService(request=RequestFactory())
+ assert service.get_form() is not None

conventions/tests/services/test_search.py

@@ -1,4 +1,3 @@
-from django.db import connection
 from django.test import TestCase
 from unittest_parametrize import ParametrizedTestCase, param, parametrize
 
@@ -11,22 +10,17 @@
  UserConventionTermineesSearchService,
 )
 from conventions.tests.factories import AvenantFactory, ConventionFactory
+from core.tests.test_utils import PGTrgmTestMixin
 from programmes.models.choices import Financement, NatureLogement
 from programmes.tests.factories import ProgrammeFactory
 from users.tests.factories import UserFactory
 
 
-class SearchServiceTestBase(ParametrizedTestCase, TestCase):
+class SearchServiceTestBase(PGTrgmTestMixin, ParametrizedTestCase, TestCase):
  __test__ = False
 
  service_class: type
 
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
- with connection.cursor() as cursor:
- cursor.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm;")
-
  def setUp(self) -> None:
  self.user = UserFactory(is_staff=True, is_superuser=True)
  ConventionFactory(
@@ -116,13 +110,9 @@ def setUp(self) -> None:
  Convention.objects.all().update(statut=ConventionStatut.RESILIEE.label)
 
 
-class TestUserConventionSmartSearchService(ParametrizedTestCase, TestCase):
- @classmethod
- def setUpClass(cls):
- super().setUpClass()
- with connection.cursor() as cursor:
- cursor.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm;")
-
+class TestUserConventionSmartSearchService(
+ PGTrgmTestMixin, ParametrizedTestCase, TestCase
+):
  def setUp(self) -> None:
  self.user = UserFactory(is_staff=True, is_superuser=True)
 

conventions/urls.py

@@ -65,21 +65,27 @@
  views.journal,
  name="journal",
  ),
- # Pages de troisième niveau : funnel d'instruction et d'action sur les conventions et avenants
+ # Pages pour l'ajout simplifié d'une convention finalisée
  path(
- "new_convention",
- permission_required("convention.add_convention")(
- views.NewConventionView.as_view()
- ),
- name="new_convention",
+ "select_operation",
+ views.SelectOperationView.as_view(),
+ name="select_operation",
  ),
  path(
  "add_convention",
  permission_required("convention.add_convention")(
- views.AddConventionView.as_view()
+ views.AddConventionFromOperationView.as_view()
  ),
  name="add_convention",
  ),
+ # Pages de troisième niveau : funnel d'instruction et d'action sur les conventions et avenants
+ path(
+ "new_convention",
+ permission_required("convention.add_convention")(
+ views.NewConventionView.as_view()
+ ),
+ name="new_convention",
+ ),
  path(
  "bailleur/<convention_uuid>",
  views.ConventionBailleurView.as_view(),

conventions/views/__init__.py

@@ -1,5 +1,5 @@
 from conventions.views.avenants import *
-from conventions.views.convention_form_add import *
+from conventions.views.convention_form_add_from_operation import *
 from conventions.views.convention_form_annexes import *
 from conventions.views.convention_form_bailleur import *
 from conventions.views.convention_form_cadastre import *

conventions/views/convention_form_add_from_operation.py

@@ -0,0 +1,39 @@
+from typing import Any
+
+from django.conf import settings
+from django.contrib.auth.mixins import LoginRequiredMixin
+from django.views.generic import TemplateView
+from waffle.mixins import WaffleFlagMixin
+
+from conventions.services.add_from_operation import (
+ ConventionAddService,
+ SelectOperationService,
+)
+
+
+class SelectOperationView(WaffleFlagMixin, LoginRequiredMixin, TemplateView):
+ waffle_flag = settings.FLAG_ADD_CONVENTION
+ template_name = "conventions/select_operation.html"
+
+ def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
+ numero_operation = self.request.GET.get("numero_operation")
+
+ exact_match, operations = SelectOperationService(
+ request=self.request, numero_operation=numero_operation
+ ).fetch_operations()
+
+ return super().get_context_data(**kwargs) | {
+ "operations": operations,
+ "siap_assistance_url": settings.SIAP_ASSISTANCE_URL,
+ "exact_match": exact_match,
+ }
+
+
+class AddConventionFromOperationView(WaffleFlagMixin, LoginRequiredMixin, TemplateView):
+ waffle_flag = settings.FLAG_ADD_CONVENTION
+ template_name = "conventions/add_from_operation.html"
+
+ def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
+ service = ConventionAddService(self.request)
+ ctx = super().get_context_data(**kwargs)
+ ctx.update({"form": service.get_form()})

core/tests/test_utils.py

@@ -1,6 +1,8 @@
 import unittest
 import uuid
 
+from django.db import connection
+
 from core.utils import get_key_from_json_field, is_valid_uuid, round_half_up
 
 
@@ -91,3 +93,11 @@ def test_get_key_from_json_field(self):
  ),
  {"myokey": "myovalue"},
  )
+
+
+class PGTrgmTestMixin:
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ with connection.cursor() as cursor:
+ cursor.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm;")