♻️(plugins) rewrite plugin system as django app
This allow more flexibility around the installed plugins, this will allow to add models in plugins if needed.
This commit is contained in:
Quentin BEY
committed by
Sabrina Demagny
Sabrina Demagny
parent
4ced342062
commit
28fdee868d
1
src/backend/plugins/la_suite/__init__.py
Normal file
1
src/backend/plugins/la_suite/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Plugin module for La Suite numérique."""
|
||||
10
src/backend/plugins/la_suite/apps.py
Normal file
10
src/backend/plugins/la_suite/apps.py
Normal file
@@ -0,0 +1,10 @@
|
||||
"""La Suite plugin application configuration."""
|
||||
|
||||
from core.plugins.base import BasePluginAppConfig
|
||||
|
||||
|
||||
class LaSuitePluginConfig(BasePluginAppConfig):
|
||||
"""Configuration for the La Suite plugin application."""
|
||||
|
||||
name = "plugins.la_suite"
|
||||
verbose_name = "La Suite Plugin"
|
||||
32
src/backend/plugins/la_suite/hooks.py
Normal file
32
src/backend/plugins/la_suite/hooks.py
Normal file
@@ -0,0 +1,32 @@
|
||||
"""
|
||||
Hooks registration for the la_suite plugin.
|
||||
|
||||
This module is automagically loaded by the plugin system.
|
||||
Putting hooks registration here allows to test the "utils"
|
||||
function without registering the hook unwillingly.
|
||||
"""
|
||||
|
||||
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)
|
||||
1
src/backend/plugins/la_suite/hooks_utils/__init__.py
Normal file
1
src/backend/plugins/la_suite/hooks_utils/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Hook modules for La Suite"""
|
||||
@@ -0,0 +1,79 @@
|
||||
"""
|
||||
This hook is used to convert the organization registration ID
|
||||
to the proper name. For French organization the registration ID
|
||||
is the SIRET.
|
||||
|
||||
This is a very specific plugin for French organizations and this
|
||||
first implementation is very basic. It surely needs to be improved
|
||||
later.
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
import requests
|
||||
from requests.adapters import HTTPAdapter, Retry
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
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 = {}
|
||||
for result in data["results"]:
|
||||
for organization in result["matching_etablissements"]:
|
||||
if organization.get("siret") == siret:
|
||||
org_metadata["is_public_service"] = result.get("complements", {}).get(
|
||||
"est_service_public", False
|
||||
)
|
||||
org_metadata["is_commune"] = (
|
||||
str(result.get("nature_juridique", "")) == "7210"
|
||||
)
|
||||
|
||||
store_signs = organization.get("liste_enseignes") or []
|
||||
if store_signs:
|
||||
return store_signs[0].title(), org_metadata
|
||||
if name := result.get("nom_raison_sociale"):
|
||||
return name.title(), org_metadata
|
||||
|
||||
logger.warning("No organization name found for SIRET %s", siret)
|
||||
return None, org_metadata
|
||||
|
||||
|
||||
def get_organization_name_and_metadata_from_siret(organization):
|
||||
"""After creating an organization, update the organization name."""
|
||||
if not organization.registration_id_list:
|
||||
# No registration ID to convert...
|
||||
return
|
||||
|
||||
if organization.name not in organization.registration_id_list:
|
||||
# The name has probably already been customized
|
||||
return
|
||||
|
||||
# In the nominal case, there is only one registration ID because
|
||||
# the organization as 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(API_URL.format(siret=siret), timeout=10)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
except requests.RequestException as exc:
|
||||
logger.exception("%s: Unable to fetch organization name from SIRET", exc)
|
||||
return
|
||||
|
||||
name, metadata = _get_organization_name_and_metadata_from_results(data, siret)
|
||||
if not name: # don't consider metadata either
|
||||
return
|
||||
|
||||
organization.name = name
|
||||
organization.metadata = (organization.metadata or {}) | metadata
|
||||
|
||||
organization.save(update_fields=["name", "metadata", "updated_at"])
|
||||
logger.info("Organization %s name updated to %s", organization, name)
|
||||
243
src/backend/plugins/la_suite/hooks_utils/communes.py
Normal file
243
src/backend/plugins/la_suite/hooks_utils/communes.py
Normal file
@@ -0,0 +1,243 @@
|
||||
"""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()
|
||||
1
src/backend/plugins/la_suite/tests/__init__.py
Normal file
1
src/backend/plugins/la_suite/tests/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Tests for the plugins module."""
|
||||
1
src/backend/plugins/la_suite/tests/hooks/__init__.py
Normal file
1
src/backend/plugins/la_suite/tests/hooks/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Test for the Organization plugins module."""
|
||||
@@ -0,0 +1,231 @@
|
||||
"""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"
|
||||
@@ -0,0 +1,211 @@
|
||||
"""Tests for the NameFromSiretOrganizationPlugin plugin."""
|
||||
|
||||
import pytest
|
||||
import responses
|
||||
|
||||
from core.models import Organization, get_organization_metadata_schema
|
||||
from core.plugins.registry import registry
|
||||
|
||||
from plugins.la_suite.hooks_utils.all_organizations import (
|
||||
get_organization_name_and_metadata_from_siret,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
# disable unused-argument for because organization_plugins_settings
|
||||
# is used to set the settings not to be used in the test
|
||||
# pylint: disable=unused-argument
|
||||
|
||||
|
||||
@pytest.fixture(name="hook_settings")
|
||||
def hook_settings_fixture(settings):
|
||||
"""
|
||||
Fixture to set the organization plugins settings and
|
||||
leave the initial state after the test.
|
||||
"""
|
||||
_original_hooks = dict(registry._hooks.items()) # pylint: disable=protected-access
|
||||
registry.register_hook(
|
||||
"organization_created", get_organization_name_and_metadata_from_siret
|
||||
)
|
||||
|
||||
settings.ORGANIZATION_METADATA_SCHEMA = "fr/organization_metadata.json"
|
||||
|
||||
# Reset the model validation cache
|
||||
get_organization_metadata_schema.cache_clear()
|
||||
get_organization_metadata_schema()
|
||||
|
||||
yield
|
||||
|
||||
# reset the hooks
|
||||
registry._hooks = _original_hooks # pylint: disable=protected-access
|
||||
|
||||
settings.ORGANIZATION_METADATA_SCHEMA = None
|
||||
|
||||
# Reset the model validation cache
|
||||
get_organization_metadata_schema.cache_clear()
|
||||
get_organization_metadata_schema()
|
||||
|
||||
|
||||
@responses.activate
|
||||
@pytest.mark.parametrize(
|
||||
"nature_juridique,is_commune,is_public_service",
|
||||
[
|
||||
("123", False, False),
|
||||
("7210", True, False),
|
||||
("123", False, True),
|
||||
("7210", True, True),
|
||||
],
|
||||
)
|
||||
def test_organization_plugins_run_after_create(
|
||||
hook_settings, nature_juridique, is_commune, is_public_service
|
||||
):
|
||||
"""Test the run_after_create method of the organization plugins for nominal case."""
|
||||
responses.add(
|
||||
responses.GET,
|
||||
"https://recherche-entreprises.api.gouv.fr/search?q=12345678901234",
|
||||
json={
|
||||
"results": [
|
||||
{
|
||||
# skipping some fields
|
||||
"matching_etablissements": [
|
||||
# skipping some fields
|
||||
{
|
||||
"liste_enseignes": ["AMAZING ORGANIZATION"],
|
||||
"siret": "12345678901234",
|
||||
}
|
||||
],
|
||||
"nature_juridique": nature_juridique,
|
||||
"complements": {
|
||||
"est_service_public": is_public_service,
|
||||
},
|
||||
}
|
||||
],
|
||||
"total_results": 1,
|
||||
"page": 1,
|
||||
"per_page": 10,
|
||||
"total_pages": 1,
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
organization = Organization.objects.create(
|
||||
name="12345678901234", registration_id_list=["12345678901234"]
|
||||
)
|
||||
assert organization.name == "Amazing Organization"
|
||||
assert organization.metadata["is_commune"] == is_commune
|
||||
assert organization.metadata["is_public_service"] == is_public_service
|
||||
|
||||
# Check that the organization has been updated in the database also
|
||||
organization.refresh_from_db()
|
||||
assert organization.name == "Amazing Organization"
|
||||
assert organization.metadata["is_commune"] == is_commune
|
||||
assert organization.metadata["is_public_service"] == is_public_service
|
||||
|
||||
|
||||
@responses.activate
|
||||
def test_organization_plugins_run_after_create_api_fail(hook_settings):
|
||||
"""Test the plugin when the API call fails."""
|
||||
responses.add(
|
||||
responses.GET,
|
||||
"https://recherche-entreprises.api.gouv.fr/search?q=12345678901234",
|
||||
json={"error": "Internal Server Error"},
|
||||
status=500,
|
||||
)
|
||||
|
||||
organization = Organization.objects.create(
|
||||
name="12345678901234", registration_id_list=["12345678901234"]
|
||||
)
|
||||
assert organization.name == "12345678901234"
|
||||
|
||||
|
||||
@responses.activate
|
||||
@pytest.mark.parametrize(
|
||||
"results",
|
||||
[
|
||||
{"results": []},
|
||||
{"results": [{"matching_etablissements": []}]},
|
||||
{"results": [{"matching_etablissements": [{"siret": "12345678901234"}]}]},
|
||||
{
|
||||
"results": [
|
||||
{
|
||||
"matching_etablissements": [
|
||||
{"siret": "12345678901234", "liste_enseignes": []}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
],
|
||||
)
|
||||
def test_organization_plugins_run_after_create_missing_data(hook_settings, results):
|
||||
"""Test the plugin when the API call returns missing data."""
|
||||
responses.add(
|
||||
responses.GET,
|
||||
"https://recherche-entreprises.api.gouv.fr/search?q=12345678901234",
|
||||
json=results,
|
||||
status=200,
|
||||
)
|
||||
|
||||
organization = Organization.objects.create(
|
||||
name="12345678901234", registration_id_list=["12345678901234"]
|
||||
)
|
||||
assert organization.name == "12345678901234"
|
||||
|
||||
|
||||
@responses.activate
|
||||
def test_organization_plugins_run_after_create_name_already_set(
|
||||
hook_settings,
|
||||
):
|
||||
"""Test the plugin does nothing when the name already differs from the registration ID."""
|
||||
organization = Organization.objects.create(
|
||||
name="Magic WOW", registration_id_list=["12345678901234"]
|
||||
)
|
||||
assert organization.name == "Magic WOW"
|
||||
|
||||
|
||||
@responses.activate
|
||||
def test_organization_plugins_run_after_create_no_list_enseignes(
|
||||
hook_settings,
|
||||
):
|
||||
"""Test the run_after_create method of the organization plugins for nominal case."""
|
||||
responses.add(
|
||||
responses.GET,
|
||||
"https://recherche-entreprises.api.gouv.fr/search?q=12345678901234",
|
||||
json={
|
||||
"results": [
|
||||
{
|
||||
"nom_raison_sociale": "AMAZING ORGANIZATION",
|
||||
# skipping some fields
|
||||
"matching_etablissements": [
|
||||
# skipping some fields
|
||||
{
|
||||
"liste_enseignes": None,
|
||||
"siret": "12345678901234",
|
||||
}
|
||||
],
|
||||
"nature_juridique": "123",
|
||||
"complements": {
|
||||
"est_service_public": True,
|
||||
},
|
||||
}
|
||||
],
|
||||
"total_results": 1,
|
||||
"page": 1,
|
||||
"per_page": 10,
|
||||
"total_pages": 1,
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
organization = Organization.objects.create(
|
||||
name="12345678901234", registration_id_list=["12345678901234"]
|
||||
)
|
||||
assert organization.name == "Amazing Organization"
|
||||
assert organization.metadata["is_commune"] is False
|
||||
assert organization.metadata["is_public_service"] is True
|
||||
|
||||
# Check that the organization has been updated in the database also
|
||||
organization.refresh_from_db()
|
||||
assert organization.name == "Amazing Organization"
|
||||
assert organization.metadata["is_commune"] is False
|
||||
assert organization.metadata["is_public_service"] is True
|
||||
@@ -0,0 +1,42 @@
|
||||
"""Test module to check all application hooks are loaded."""
|
||||
|
||||
from core.plugins.registry import registry
|
||||
|
||||
from plugins.la_suite.apps import LaSuitePluginConfig
|
||||
|
||||
|
||||
def test_hooks_loaded():
|
||||
"""Test to check all application hooks are loaded."""
|
||||
_original_hooks = dict(registry._hooks.items()) # pylint: disable=protected-access
|
||||
_original_registered_apps = set(registry._registered_apps) # pylint: disable=protected-access
|
||||
|
||||
registry.reset()
|
||||
|
||||
assert registry.get_callbacks("organization_created") == []
|
||||
assert registry.get_callbacks("organization_access_granted") == []
|
||||
|
||||
# Force the application to run "ready" method
|
||||
LaSuitePluginConfig(
|
||||
app_name="plugins.la_suite", app_module=__import__("plugins.la_suite")
|
||||
).ready()
|
||||
|
||||
# Check that the hooks are loaded
|
||||
organization_created_hook_names = [
|
||||
callback.__name__ for callback in registry.get_callbacks("organization_created")
|
||||
]
|
||||
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
|
||||
registry._hooks = _original_hooks # pylint: disable=protected-access
|
||||
registry._registered_apps = _original_registered_apps # pylint: disable=protected-access
|
||||
Reference in New Issue
Block a user