🔥(plugins) remove CommuneCreation plugin

Remove the French communes auto-provisioning plugin that handled
SIRET lookup, DNS setup, and MailDomain creation.

CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to
 
 ## [Unreleased]
 
+- 🔥(plugins) remove CommuneCreation plugin
+
 ## [1.21.0] - 2025-12-05
 
 - ✨(aliases) import existing aliases from dimail

env.d/development/france.dist

@@ -1,2 +1 @@
 INSTALLED_PLUGINS=plugins.la_suite
-DNS_PROVISIONING_TARGET_ZONE=test.collectivite.fr

src/backend/people/settings.py

@@ -581,26 +581,6 @@ class Base(Configuration):
         environ_name="MAIL_CHECK_DOMAIN_INTERVAL",
         environ_prefix=None,
     )
-    DNS_PROVISIONING_TARGET_ZONE = values.Value(
-        default=None,
-        environ_name="DNS_PROVISIONING_TARGET_ZONE",
-        environ_prefix=None,
-    )
-    DNS_PROVISIONING_API_URL = values.Value(
-        default="https://api.scaleway.com",
-        environ_name="DNS_PROVISIONING_API_URL",
-        environ_prefix=None,
-    )
-    DNS_PROVISIONING_RESOURCE_ID = values.Value(
-        default=None,
-        environ_name="DNS_PROVISIONING_RESOURCE_ID",
-        environ_prefix=None,
-    )
-    DNS_PROVISIONING_API_CREDENTIALS = values.Value(
-        default=None,
-        environ_name="DNS_PROVISIONING_API_CREDENTIALS",
-        environ_prefix=None,
-    )
     MATRIX_BASE_HOME_SERVER = values.Value(
         default="https://matrix.agent.dinum.tchap.gouv.fr",
         environ_name="MATRIX_BASE_HOME_SERVER",

src/backend/plugins/la_suite/hooks.py

@@ -11,22 +11,9 @@ from core.plugins.registry import register_hook
 from plugins.la_suite.hooks_utils.all_organizations import (
     get_organization_name_and_metadata_from_siret,
 )
-from plugins.la_suite.hooks_utils.communes import CommuneCreation
 
 
 @register_hook("organization_created")
 def get_organization_name_and_metadata_from_siret_hook(organization):
     """After creating an organization, update the organization name & metadata."""
     get_organization_name_and_metadata_from_siret(organization)
-
-
-@register_hook("organization_created")
-def commune_organization_created(organization):
-    """After creating an organization, update the organization name."""
-    CommuneCreation().run_after_create(organization)
-
-
-@register_hook("organization_access_granted")
-def commune_organization_access_granted(organization_access):
-    """After granting an organization access, check for needed domain access grant."""
-    CommuneCreation().run_after_grant_access(organization_access)

src/backend/plugins/la_suite/hooks_utils/all_organizations.py

@@ -21,25 +21,45 @@ API_URL = "https://recherche-entreprises.api.gouv.fr/search?q={siret}"
 
 def _get_organization_name_and_metadata_from_results(data, siret):
     """Return the organization name and metadata from the results of a SIRET search."""
-    org_metadata = {}
+    # Find matching organization
-    for result in data["results"]:
+    match = next(
-        for organization in result["matching_etablissements"]:
+        (
-            if organization.get("siret") == siret:
+            (res, org)
-                org_metadata["is_public_service"] = result.get("complements", {}).get(
+            for res in data.get("results", [])
-                    "est_service_public", False
+            for org in res.get("matching_etablissements", [])
-                )
+            if org.get("siret") == siret
-                org_metadata["is_commune"] = (
+        ),
-                    str(result.get("nature_juridique", "")) == "7210"
+        None,
     )
 
+    if not match:
+        logger.warning("No organization name found for SIRET %s", siret)
+        return None, {}
+
+    result, organization = match
+
+    # Extract metadata
+    is_commune = str(result.get("nature_juridique", "")) == "7210"
+    metadata = {
+        "is_public_service": result.get("complements", {}).get(
+            "est_service_public", False
+        ),
+        "is_commune": is_commune,
+    }
+
+    # Extract name (priority: commune name > store signs > business name)
+    name = None
+    if is_commune:
+        name = result.get("siege", {}).get("libelle_commune")
+    if not name:  # Fallback for non-communes OR if commune has no libelle_commune
         store_signs = organization.get("liste_enseignes") or []
-                if store_signs:
+        name = store_signs[0] if store_signs else result.get("nom_raison_sociale")
-                    return store_signs[0].title(), org_metadata
+
-                if name := result.get("nom_raison_sociale"):
+    if name:
-                    return name.title(), org_metadata
+        return name.title(), metadata
 
     logger.warning("No organization name found for SIRET %s", siret)
-    return None, org_metadata
+    return None, metadata
 
 
 def get_organization_name_and_metadata_from_siret(organization):

src/backend/plugins/la_suite/hooks_utils/communes.py

@@ -1,243 +0,0 @@
-"""Organization related plugins."""
-
-import logging
-import re
-
-from django.conf import settings
-from django.utils.text import slugify
-
-import requests
-from requests.adapters import HTTPAdapter, Retry
-
-from mailbox_manager.enums import MailDomainRoleChoices
-from mailbox_manager.models import MailDomain, MailDomainAccess
-
-logger = logging.getLogger(__name__)
-
-
-class ApiCall:
-    """Encapsulates a call to an external API"""
-
-    inputs: dict = {}
-    method: str = "GET"
-    base: str = ""
-    url: str = ""
-    params: dict = {}
-    headers: dict = {}
-    response_data = None
-
-    def execute(self):
-        """Call the specified API endpoint with supplied parameters and record response"""
-        if self.method in ("POST", "PATCH"):
-            response = requests.request(
-                method=self.method,
-                url=f"{self.base}/{self.url}",
-                json=self.params,
-                headers=self.headers,
-                timeout=20,
-            )
-        else:
-            response = requests.request(
-                method=self.method,
-                url=f"{self.base}/{self.url}",
-                params=self.params,
-                headers=self.headers,
-                timeout=20,
-            )
-        self.response_data = response.json()
-        logger.info(
-            "API call: %s %s %s %s",
-            self.method,
-            self.url,
-            self.params,
-            self.response_data,
-        )
-
-
-class CommuneCreation:
-    """
-    This plugin handles setup tasks for French communes.
-    """
-
-    _api_url = "https://recherche-entreprises.api.gouv.fr/search?q={siret}"
-
-    def get_organization_name_from_results(self, data, siret):
-        """Return the organization name from the results of a SIRET search."""
-        for result in data["results"]:
-            nature = "nature_juridique"
-            commune = nature in result and result[nature] == "7210"
-            if commune:
-                return result["siege"]["libelle_commune"].title()
-
-        logger.warning("Not a commune: SIRET %s", siret)
-        return None
-
-    def dns_call(self, spec):
-        """Call to add a DNS record"""
-        zone_name = self.zone_name(spec.inputs["name"])
-
-        records = [
-            {
-                "name": item["target"],
-                "type": item["type"].upper(),
-                "data": item["value"],
-                "ttl": 3600,
-            }
-            for item in spec.response_data
-        ]
-        result = ApiCall()
-        result.method = "PATCH"
-        result.base = "https://api.scaleway.com"
-        result.url = f"/domain/v2beta1/dns-zones/{zone_name}/records"
-        result.params = {"changes": [{"add": {"records": records}}]}
-        result.headers = {"X-Auth-Token": settings.DNS_PROVISIONING_API_CREDENTIALS}
-        return result
-
-    def normalize_name(self, name: str) -> str:
-        """Map the name to a standard form"""
-        name = re.sub("'", "-", name)
-        return slugify(name)
-
-    def zone_name(self, name: str) -> str:
-        """Derive the zone name from the commune name"""
-        normalized = self.normalize_name(name)
-        return f"{normalized}.{settings.DNS_PROVISIONING_TARGET_ZONE}"
-
-    def complete_commune_creation(self, name: str) -> ApiCall:
-        """Specify the tasks to be completed after a commune is created."""
-        inputs = {"name": self.normalize_name(name)}
-
-        create_zone = ApiCall()
-        create_zone.method = "POST"
-        create_zone.base = "https://api.scaleway.com"
-        create_zone.url = "/domain/v2beta1/dns-zones"
-        create_zone.params = {
-            "project_id": settings.DNS_PROVISIONING_RESOURCE_ID,
-            "domain": settings.DNS_PROVISIONING_TARGET_ZONE,
-            "subdomain": inputs["name"],
-        }
-        create_zone.headers = {
-            "X-Auth-Token": settings.DNS_PROVISIONING_API_CREDENTIALS
-        }
-
-        zone_name = self.zone_name(inputs["name"])
-
-        create_domain = ApiCall()
-        create_domain.method = "POST"
-        create_domain.base = settings.MAIL_PROVISIONING_API_URL
-        create_domain.url = "/domains/"
-        create_domain.params = {
-            "name": zone_name,
-            "delivery": "virtual",
-            "features": ["webmail", "mailbox"],
-            "context_name": zone_name,
-        }
-        create_domain.headers = {
-            "Authorization": f"Basic {settings.MAIL_PROVISIONING_API_CREDENTIALS}"
-        }
-
-        spec_domain = ApiCall()
-        spec_domain.inputs = inputs
-        spec_domain.base = settings.MAIL_PROVISIONING_API_URL
-        spec_domain.url = f"/domains/{zone_name}/spec"
-        spec_domain.headers = {
-            "Authorization": f"Basic {settings.MAIL_PROVISIONING_API_CREDENTIALS}"
-        }
-
-        return [create_zone, create_domain, spec_domain]
-
-    def complete_zone_creation(self, spec_call):
-        """Specify the tasks to be performed to set up the zone."""
-        return self.dns_call(spec_call)
-
-    def run_after_create(self, organization):
-        """After creating an organization, update the organization name."""
-        logger.info("In CommuneCreation")
-        if not organization.registration_id_list:
-            # No registration ID to convert...
-            return
-
-        # In the nominal case, there is only one registration ID because
-        # the organization has been created from it.
-        try:
-            # Retry logic as the API may be rate limited
-            s = requests.Session()
-            retries = Retry(total=5, backoff_factor=0.1, status_forcelist=[429])
-            s.mount("https://", HTTPAdapter(max_retries=retries))
-
-            siret = organization.registration_id_list[0]
-            response = s.get(self._api_url.format(siret=siret), timeout=10)
-            response.raise_for_status()
-            data = response.json()
-            name = self.get_organization_name_from_results(data, siret)
-            # Not a commune ?
-            if not name:
-                return
-        except requests.RequestException as exc:
-            logger.exception("%s: Unable to fetch organization name from SIRET", exc)
-            return
-
-        organization.name = name
-        organization.save(update_fields=["name", "updated_at"])
-        logger.info("Organization %s name updated to %s", organization, name)
-
-        zone_name = self.zone_name(name)
-        support = "support-regie@numerique.gouv.fr"
-        MailDomain.objects.get_or_create(name=zone_name, support_email=support)
-
-        # Compute and execute the rest of the process
-        tasks = self.complete_commune_creation(name)
-        for task in tasks:
-            task.execute()
-        last_task = self.complete_zone_creation(tasks[-1])
-        last_task.execute()
-
-    def complete_grant_access(self, sub, zone_name):
-        """Specify the tasks to be completed after making a user admin"""
-        create_user = ApiCall()
-        create_user.method = "POST"
-        create_user.base = settings.MAIL_PROVISIONING_API_URL
-        create_user.url = "/users/"
-        create_user.params = {
-            "name": sub,
-            "password": "no",
-            "is_admin": False,
-            "perms": [],
-        }
-        create_user.headers = {
-            "Authorization": f"Basic {settings.MAIL_PROVISIONING_API_CREDENTIALS}"
-        }
-
-        grant_access = ApiCall()
-        grant_access.method = "POST"
-        grant_access.base = settings.MAIL_PROVISIONING_API_URL
-        grant_access.url = "/allows/"
-        grant_access.params = {
-            "user": sub,
-            "domain": zone_name,
-        }
-        grant_access.headers = {
-            "Authorization": f"Basic {settings.MAIL_PROVISIONING_API_CREDENTIALS}"
-        }
-
-        return [create_user, grant_access]
-
-    def run_after_grant_access(self, organization_access):
-        """After granting an organization access, check for needed domain access grant."""
-        orga = organization_access.organization
-        user = organization_access.user
-        zone_name = self.zone_name(orga.name)
-
-        try:
-            domain = MailDomain.objects.get(name=zone_name)
-        except MailDomain.DoesNotExist:
-            domain = None
-
-        if domain:
-            MailDomainAccess.objects.create(
-                domain=domain, user=user, role=MailDomainRoleChoices.OWNER
-            )
-
-            tasks = self.complete_grant_access(user.sub, zone_name)
-            for task in tasks:
-                task.execute()

src/backend/plugins/la_suite/tests/hooks/test_commune_creation.py

@@ -1,231 +0,0 @@
-"""Tests for the CommuneCreation plugin."""
-
-from django.conf import settings
-from django.test.utils import override_settings
-
-import pytest
-import responses
-
-from plugins.la_suite.hooks_utils.communes import ApiCall, CommuneCreation
-
-pytestmark = pytest.mark.django_db
-
-
-def test_extract_name_from_org_data_when_commune():
-    """Test the name is extracted correctly for a French commune."""
-    data = {
-        "results": [
-            {
-                "nom_complet": "COMMUNE DE VARZY",
-                "nom_raison_sociale": "COMMUNE DE VARZY",
-                "siege": {
-                    "libelle_commune": "VARZY",
-                    "liste_enseignes": ["MAIRIE"],
-                    "siret": "21580304000017",
-                },
-                "nature_juridique": "7210",
-                "matching_etablissements": [
-                    {
-                        "siret": "21580304000017",
-                        "libelle_commune": "VARZY",
-                        "liste_enseignes": ["MAIRIE"],
-                    }
-                ],
-            }
-        ]
-    }
-
-    plugin = CommuneCreation()
-    name = plugin.get_organization_name_from_results(data, "21580304000017")
-    assert name == "Varzy"
-
-
-def test_api_call_execution():
-    """Test that calling execute() faithfully executes the API call"""
-    task = ApiCall()
-    task.method = "POST"
-    task.base = "https://some_host"
-    task.url = "some_url"
-    task.params = {"some_key": "some_value"}
-    task.headers = {"Some-Header": "Some-Header-Value"}
-
-    with responses.RequestsMock() as rsps:
-        rsps.add(
-            rsps.POST,
-            url="https://some_host/some_url",
-            body='{"some_key": "some_value"}',
-            content_type="application/json",
-            headers={"Some-Header": "Some-Header-Value"},
-        )
-
-        task.execute()
-
-
-@override_settings(DNS_PROVISIONING_TARGET_ZONE="collectivite.fr")
-def test_tasks_on_commune_creation_include_zone_creation():
-    """Test the first task in commune creation: creating the DNS sub-zone"""
-    plugin = CommuneCreation()
-    name = "Varzy"
-
-    tasks = plugin.complete_commune_creation(name)
-
-    assert tasks[0].base == "https://api.scaleway.com"
-    assert tasks[0].url == "/domain/v2beta1/dns-zones"
-    assert tasks[0].method == "POST"
-    assert tasks[0].params == {
-        "project_id": settings.DNS_PROVISIONING_RESOURCE_ID,
-        "domain": "collectivite.fr",
-        "subdomain": "varzy",
-    }
-    assert tasks[0].headers["X-Auth-Token"] == settings.DNS_PROVISIONING_API_CREDENTIALS
-
-
-@override_settings(DNS_PROVISIONING_TARGET_ZONE="collectivite.fr")
-def test_tasks_on_commune_creation_include_dimail_domain_creation():
-    """Test the second task in commune creation: creating the domain in Dimail"""
-    plugin = CommuneCreation()
-    name = "Merlaut"
-
-    tasks = plugin.complete_commune_creation(name)
-
-    assert tasks[1].base == settings.MAIL_PROVISIONING_API_URL
-    assert tasks[1].url == "/domains/"
-    assert tasks[1].method == "POST"
-    assert tasks[1].params == {
-        "name": "merlaut.collectivite.fr",
-        "delivery": "virtual",
-        "features": ["webmail", "mailbox"],
-        "context_name": "merlaut.collectivite.fr",
-    }
-    assert (
-        tasks[1].headers["Authorization"]
-        == f"Basic {settings.MAIL_PROVISIONING_API_CREDENTIALS}"
-    )
-
-
-@override_settings(DNS_PROVISIONING_TARGET_ZONE="collectivite.fr")
-def test_tasks_on_commune_creation_include_fetching_spec():
-    """Test the third task in commune creation: asking Dimail for the spec"""
-    plugin = CommuneCreation()
-    name = "Loc-Eguiner"
-
-    tasks = plugin.complete_commune_creation(name)
-
-    assert tasks[2].base == settings.MAIL_PROVISIONING_API_URL
-    assert tasks[2].url == "/domains/loc-eguiner.collectivite.fr/spec"
-    assert tasks[2].method == "GET"
-    assert (
-        tasks[2].headers["Authorization"]
-        == f"Basic {settings.MAIL_PROVISIONING_API_CREDENTIALS}"
-    )
-
-
-@override_settings(DNS_PROVISIONING_TARGET_ZONE="collectivite.fr")
-def test_tasks_on_commune_creation_include_dns_records():
-    """Test the next several tasks in commune creation: creating records"""
-    plugin = CommuneCreation()
-    name = "Abidos"
-
-    spec_response = [
-        {"target": "", "type": "mx", "value": "mx.dev.ox.numerique.gouv.fr."},
-        {
-            "target": "dimail._domainkey",
-            "type": "txt",
-            "value": "v=DKIM1; h=sha256; k=rsa; p=MIICIjANB<truncated>AAQ==",
-        },
-        {"target": "imap", "type": "cname", "value": "imap.dev.ox.numerique.gouv.fr."},
-        {"target": "smtp", "type": "cname", "value": "smtp.dev.ox.numerique.gouv.fr."},
-        {
-            "target": "",
-            "type": "txt",
-            "value": "v=spf1 include:_spf.dev.ox.numerique.gouv.fr -all",
-        },
-        {
-            "target": "webmail",
-            "type": "cname",
-            "value": "webmail.dev.ox.numerique.gouv.fr.",
-        },
-    ]
-
-    tasks = plugin.complete_commune_creation(name)
-    tasks[2].response_data = spec_response
-
-    expected = {
-        "changes": [
-            {
-                "add": {
-                    "records": [
-                        {
-                            "name": item["target"],
-                            "type": item["type"].upper(),
-                            "data": item["value"],
-                            "ttl": 3600,
-                        }
-                        for item in spec_response
-                    ]
-                }
-            }
-        ]
-    }
-
-    zone_call = plugin.complete_zone_creation(tasks[2])
-    assert zone_call.params == expected
-    assert zone_call.url == "/domain/v2beta1/dns-zones/abidos.collectivite.fr/records"
-    assert (
-        zone_call.headers["X-Auth-Token"] == settings.DNS_PROVISIONING_API_CREDENTIALS
-    )
-
-
-@override_settings(DNS_PROVISIONING_TARGET_ZONE="collectivite.fr")
-def test_tasks_on_grant_access():
-    """Test the final tasks after making user admin of an org"""
-    plugin = CommuneCreation()
-
-    tasks = plugin.complete_grant_access("some-sub", "mezos.collectivite.fr")
-
-    assert tasks[0].base == settings.MAIL_PROVISIONING_API_URL
-    assert tasks[0].url == "/users/"
-    assert tasks[0].method == "POST"
-    assert tasks[0].params == {
-        "name": "some-sub",
-        "password": "no",
-        "is_admin": False,
-        "perms": [],
-    }
-    assert (
-        tasks[0].headers["Authorization"]
-        == f"Basic {settings.MAIL_PROVISIONING_API_CREDENTIALS}"
-    )
-
-    assert tasks[1].base == settings.MAIL_PROVISIONING_API_URL
-    assert tasks[1].url == "/allows/"
-    assert tasks[1].method == "POST"
-    assert tasks[1].params == {
-        "user": "some-sub",
-        "domain": "mezos.collectivite.fr",
-    }
-    assert (
-        tasks[1].headers["Authorization"]
-        == f"Basic {settings.MAIL_PROVISIONING_API_CREDENTIALS}"
-    )
-
-
-def test_normalize_name():
-    """Test name normalization"""
-    plugin = CommuneCreation()
-    assert plugin.normalize_name("Asnières-sur-Saône") == "asnieres-sur-saone"
-    assert plugin.normalize_name("Bâgé-le-Châtel") == "bage-le-chatel"
-    assert plugin.normalize_name("Courçais") == "courcais"
-    assert plugin.normalize_name("Moÿ-de-l'Aisne") == "moy-de-l-aisne"
-    assert plugin.normalize_name("Salouël") == "salouel"
-    assert (
-        plugin.normalize_name("Bors (Canton de Tude-et-Lavalette)")
-        == "bors-canton-de-tude-et-lavalette"
-    )
-
-
-@override_settings(DNS_PROVISIONING_TARGET_ZONE="collectivite.fr")
-def test_zone_name():
-    """Test transforming a commune name to a sub-zone of collectivite.fr"""
-    plugin = CommuneCreation()
-    assert plugin.zone_name("Bâgé-le-Châtel") == "bage-le-chatel.collectivite.fr"

src/backend/plugins/la_suite/tests/hooks/test_hooks_loaded.py

@@ -26,15 +26,6 @@ def test_hooks_loaded():
     ]
     assert organization_created_hook_names == [
         "get_organization_name_and_metadata_from_siret_hook",
-        "commune_organization_created",
-    ]
-
-    organization_access_granted_hook_names = [
-        callback.__name__
-        for callback in registry.get_callbacks("organization_access_granted")
-    ]
-    assert organization_access_granted_hook_names == [
-        "commune_organization_access_granted"
     ]
 
     # cleanup the hooks

src/frontend/apps/e2e/__tests__/app-desk/domain-provisioning.spec.ts

@@ -1,27 +0,0 @@
-import { expect, test } from '@playwright/test';
-
-import { keyCloakSignIn } from './common';
-
-test.beforeEach(async ({ page, browserName }) => {
-  await page.goto('/');
-  await keyCloakSignIn(page, browserName, 'marie');
-});
-
-test.describe('When a commune, domain is created on first login via ProConnect', () => {
-  test('it checks the domain has been created and is operational', async ({
-    page,
-  }) => {
-    const menu = page.locator('menu').first();
-
-    await menu.getByRole('button', { name: 'Mail Domains button' }).click();
-    await page.waitForURL('http://localhost:3000/mail-domains/**');
-    await expect(
-      page.getByRole('heading', {
-        name: 'Domains of the organization',
-        exact: true,
-      }),
-    ).toBeVisible();
-    await expect(page.getByText('merlaut.test.collectivite.fr')).toHaveCount(1);
-    await expect(page.getByText('No domains exist.')).toHaveCount(0);
-  });
-});