Feat/add tests

backend/geonature/core/command/create_gn_module.py

@@ -1,35 +1,56 @@
+import importlib
 import os
-import sys
-import subprocess
 import site
-import importlib
+import subprocess
+import sys
 from pathlib import Path
 
 import click
+import geonature.utils.config
 from click import ClickException
-
-from geonature.utils.env import ROOT_DIR
-from geonature.utils.module import iter_modules_dist, get_dist_from_code, module_db_upgrade
-
 from geonature.core.command.main import main
-import geonature.utils.config
-from geonature.utils.config import config
 from geonature.utils.command import (
-    install_frontend_dependencies,
-    create_frontend_module_config,
     build_frontend,
+    create_frontend_module_config,
+    install_frontend_dependencies,
 )
+from geonature.utils.config import config
+from geonature.utils.env import ROOT_DIR
+from geonature.utils.module import get_dist_from_code, iter_modules_dist, module_db_upgrade
 
 
 @main.command()
 @click.option(
     "-x", "--x-arg", multiple=True, help="Additional arguments consumed by custom env.py scripts"
 )
-@click.argument("module_path", type=click.Path(exists=True, file_okay=False, path_type=Path))
+@click.argument("module_path", type=click.Path(path_type=Path))
 @click.argument("module_code", required=False)
 @click.option("--build", type=bool, required=False, default=True)
 @click.option("--upgrade-db", type=bool, required=False, default=True)
 def install_gn_module(x_arg, module_path, module_code, build, upgrade_db):
+    """
+    Command definition to install a GeoNature module
+
+    Parameters
+    ----------
+    x_arg : list
+        additional arguments
+    module_path : str
+        path of the module directory
+    module_code : str
+        code of the module, deprecated in future release
+    build : boolean
+        is the frontend rebuild
+    upgrade_db : boolean
+        migrate the revision associated with the module
+
+    Raises
+    ------
+    ClickException
+        No module found with the given module code
+    ClickException
+        No module code was detected in the code
+    """
     click.echo("Installation du backend…")
     subprocess.run(f"pip install -e '{module_path}'", shell=True, check=True)
 
@@ -40,7 +61,7 @@ def install_gn_module(x_arg, module_path, module_code, build, upgrade_db):
     if module_code:
         # load python package
         module_dist = get_dist_from_code(module_code)
-        if not module_dist:
+        if not module_dist:  # FIXME : technically can't go there...
             raise ClickException(f"Aucun module ayant pour code {module_code} n’a été trouvé")
     else:
         for module_dist in iter_modules_dist():
@@ -56,7 +77,6 @@ def install_gn_module(x_arg, module_path, module_code, build, upgrade_db):
             raise ClickException(
                 f"Impossible de détecter le code du module, essayez de le spécifier."
             )
-
     # symlink module in exernal module directory
     module_frontend_path = (module_path / "frontend").resolve()
     module_symlink = ROOT_DIR / "frontend" / "external_modules" / module_code.lower()
@@ -68,7 +88,6 @@ def install_gn_module(x_arg, module_path, module_code, build, upgrade_db):
     else:
         click.echo(f"Création du lien symbolique {module_symlink} → {module_frontend_path}")
         os.symlink(module_frontend_path, module_symlink)
-
     if (Path(module_path) / "frontend" / "package-lock.json").is_file():
         click.echo("Installation des dépendances frontend…")
         install_frontend_dependencies(module_frontend_path)
@@ -80,7 +99,6 @@ def install_gn_module(x_arg, module_path, module_code, build, upgrade_db):
         click.echo("Rebuild du frontend …")
         build_frontend()
         click.secho("Rebuild du frontend terminé.", fg="green")
-
     if upgrade_db:
         click.echo("Installation / mise à jour de la base de données…")
         if not module_db_upgrade(module_dist, x_arg=x_arg):

backend/geonature/core/gn_commons/validation/routes.py

@@ -1,4 +1,5 @@
 import logging
+import uuid
 
 from werkzeug.exceptions import BadRequest
 
@@ -9,20 +10,31 @@
 from geonature.core.gn_commons.models import TValidations
 from geonature.core.gn_permissions import decorators as permissions
 from geonature.utils.env import DB
-from geonature.utils.utilssqlalchemy import test_is_uuid
 
 
 from ..routes import routes
 
 log = logging.getLogger()
 
 
+def is_uuid(uuid_string):
+    try:
+        # Si uuid_string est un code hex valide mais pas un uuid valid,
+        # UUID() va quand même le convertir en uuid valide. Pour se prémunir
+        # de ce problème, on check la version original (sans les tirets) avec
+        # le code hex généré qui doivent être les mêmes.
+        uid = uuid.UUID(uuid_string)
+        return uid.hex == uuid_string.replace("-", "")
+    except ValueError:
+        return False
+
+
 @routes.route("/history/<uuid_attached_row>", methods=["GET"])
 @permissions.check_cruved_scope("R", module_code="SYNTHESE")
 @json_resp
 def get_hist(uuid_attached_row):
     # Test if uuid_attached_row is uuid
-    if not test_is_uuid(uuid_attached_row):
+    if not is_uuid(uuid_attached_row):
         raise BadRequest("Value error uuid_attached_row is not valid")
     """
     Here we use execute() instead of scalars() because

backend/geonature/core/gn_meta/repositories.py

@@ -1,6 +1,6 @@
 import logging
 
-from sqlalchemy import or_, String, Date, and_
+from sqlalchemy import or_, String, Date, and_, func
 from sqlalchemy.inspection import inspect
 from sqlalchemy.orm import joinedload, contains_eager, aliased
 from sqlalchemy.orm.exc import NoResultFound
@@ -32,13 +32,16 @@
 
 
 def cruved_ds_filter(model, role, scope):
+    # TODO check if not used elsewhere (not found in major module of Geonature)
     if scope not in (1, 2, 3):
         raise Unauthorized("Not a valid cruved value")
     elif scope == 3:
         return True
     elif scope in (1, 2):
-        sub_q = DB.select(TDatasets).join(
-            CorDatasetActor, TDatasets.id_dataset == CorDatasetActor.id_dataset
+        sub_q = (
+            DB.select(func.count("*"))
+            .select_from(TDatasets)
+            .join(CorDatasetActor, TDatasets.id_dataset == CorDatasetActor.id_dataset)
         )
 
         or_filter = [
@@ -49,10 +52,8 @@ def cruved_ds_filter(model, role, scope):
         # if organism is None => do not filter on id_organism even if level = 2
         if scope == 2 and role.id_organisme is not None:
             or_filter.append(CorDatasetActor.id_organism == role.id_organisme)
-        sub_q = sub_q.filter(and_(or_(*or_filter), model.id_dataset == TDatasets.id_dataset))
-        return sub_q.exists()
-
-    return True
+        sub_q = sub_q.where(and_(or_(*or_filter), model.id_dataset == TDatasets.id_dataset))
+        return DB.session.execute(sub_q).scalar_one() > 0
 
 
 def cruved_af_filter(model, role, scope):
@@ -61,10 +62,14 @@ def cruved_af_filter(model, role, scope):
     elif scope == 3:
         return True
     elif scope in (1, 2):
-        sub_q = DB.select(TAcquisitionFramework).join(
-            CorAcquisitionFrameworkActor,
-            TAcquisitionFramework.id_acquisition_framework
-            == CorAcquisitionFrameworkActor.id_acquisition_framework,
+        sub_q = (
+            DB.select(func.count("*"))
+            .select_from(TAcquisitionFramework)
+            .join(
+                CorAcquisitionFrameworkActor,
+                TAcquisitionFramework.id_acquisition_framework
+                == CorAcquisitionFrameworkActor.id_acquisition_framework,
+            )
         )
 
         or_filter = [
@@ -81,33 +86,23 @@ def cruved_af_filter(model, role, scope):
                 model.id_acquisition_framework == TAcquisitionFramework.id_acquisition_framework,
             )
         )
-        return sub_q.exists()
+        return DB.session.execute(sub_q).scalar_one() > 0
 
 
 def get_metadata_list(role, scope, args, exclude_cols):
-    num = args.get("num")
-    uuid = args.get("uuid")
-    name = args.get("name")
-    date = args.get("date")
-    organisme = args.get("organism")
-    person = args.get("person")
+    id_acquisition_framework = args.get("num")
+    unique_acquisition_framework_id = args.get("uuid")
+    acquisition_framework_name = args.get("name")
+    meta_create_date = args.get("date")
+    id_organism = args.get("organism")
+    id_role = args.get("person")
     selector = args.get("selector")
     is_parent = args.get("is_parent")
+    order_by = args.get("orderby", None)
 
-    # @TODO : replace by select
-    query = DB.session.query(TAcquisitionFramework)
-
-    if is_parent is not None:
-        query = query.where(TAcquisitionFramework.is_parent)
-
-    if selector == "af" and set(["organism", "person"]).intersection(args):
-        query = query.join(
-            CorAcquisitionFrameworkActor,
-            TAcquisitionFramework.id_acquisition_framework
-            == CorAcquisitionFrameworkActor.id_acquisition_framework,
-        )
-        # remove cor_af_actor from joined load because already joined
-        exclude_cols.append("cor_af_actor")
+    query = DB.select(TAcquisitionFramework).where_if(
+        is_parent is not None, TAcquisitionFramework.is_parent
+    )
 
     if selector == "ds":
         query = query.join(
@@ -132,44 +127,74 @@ def get_metadata_list(role, scope, args, exclude_cols):
             cruved_ds_filter(TDatasets, role, scope),
         )
     )
-    if args.get("selector") == "af":
+    if selector == "af":
+        if set(["organism", "person"]).intersection(args):
+            query = query.join(
+                CorAcquisitionFrameworkActor,
+                TAcquisitionFramework.id_acquisition_framework
+                == CorAcquisitionFrameworkActor.id_acquisition_framework,
+            )
+            # remove cor_af_actor from joined load because already joined
+            exclude_cols.append("cor_af_actor")
         query = (
-            query.where(TAcquisitionFramework.id_acquisition_framework == num if num else True)
+            query.where(
+                TAcquisitionFramework.id_acquisition_framework == id_acquisition_framework
+                if id_acquisition_framework
+                else True
+            )
             .where(
                 cast(TAcquisitionFramework.unique_acquisition_framework_id, String).ilike(
-                    f"%{uuid.strip()}%"
+                    f"%{unique_acquisition_framework_id.strip()}%"
                 )
-                if uuid
+                if unique_acquisition_framework_id
                 else True
             )
             .where(
-                TAcquisitionFramework.acquisition_framework_name.ilike(f"%{name}%")
-                if name
+                TAcquisitionFramework.acquisition_framework_name.ilike(
+                    f"%{acquisition_framework_name}%"
+                )
+                if acquisition_framework_name
                 else True
             )
-            .where(CorAcquisitionFrameworkActor.id_organism == organisme if organisme else True)
-            .where(CorAcquisitionFrameworkActor.id_role == person if person else True)
+            .where(
+                CorAcquisitionFrameworkActor.id_organism == id_organism if id_organism else True
+            )
+            .where(CorAcquisitionFrameworkActor.id_role == id_role if id_role else True)
         )
 
-    elif args.get("selector") == "ds":
+    elif selector == "ds":
         query = (
-            query.where(TDatasets.id_dataset == num if num else True)
+            query.where(
+                TDatasets.id_dataset == id_acquisition_framework
+                if id_acquisition_framework
+                else True
+            )
+            .where(
+                cast(TDatasets.unique_dataset_id, String).ilike(
+                    f"%{unique_acquisition_framework_id.strip()}%"
+                )
+                if unique_acquisition_framework_id
+                else True
+            )
+            .where(
+                TAcquisitionFramework.datasets.any(dataset_name=acquisition_framework_name)
+                if acquisition_framework_name
+                else True
+            )
             .where(
-                cast(TDatasets.unique_dataset_id, String).ilike(f"%{uuid.strip()}%")
-                if uuid
+                cast(TDatasets.meta_create_date, Date) == meta_create_date
+                if meta_create_date
                 else True
             )
-            .where(TAcquisitionFramework.datasets.any(dataset_name=name) if name else True)
-            .where(cast(TDatasets.meta_create_date, Date) == date if date else True)
-            .where(CorDatasetActor.id_organism == organisme if organisme else True)
-            .where(CorDatasetActor.id_role == person if person else True)
+            .where(CorDatasetActor.id_organism == id_organism if id_organism else True)
+            .where(CorDatasetActor.id_role == id_role if id_role else True)
         )
 
-    if args.get("orderby", None):
+    if order_by:
         try:
-            query = query.order_by(getattr(TAcquisitionFramework, args.get("orderby")).asc())
+            query = query.order_by(getattr(TAcquisitionFramework, order_by).asc())
         except:
-            query = query.order_by(getattr(TDatasets, args.get("orderby")).asc())
+            query = query.order_by(getattr(TDatasets, order_by).asc())
         finally:
             pass
     return query