Add no-auth session manager to romi_run_task CLI

src/romitask/cli/print_task_info.py

@@ -288,14 +288,14 @@ def main():
     md_path = os.path.join(args.db_path, 'metadata')
     json_list = [f for f in os.listdir(md_path) if f.startswith(args.task) and f.endswith('.json')]
     if json_list == []:
-        raise IOError("Could not find the JSON metadata file associated to task '{}' in dataset '{}'!".format(args.task,
+        raise IOError("Could not find the JSON metadata file associated with task '{}' in dataset '{}'!".format(args.task,
                                                                                                               args.db_path))
     elif len(json_list) == 1:
         md_json = json_list[0]
-        print("Found a JSON metadata file associated to task '{}' in dataset '{}'!".format(args.task, args.db_path))
+        print("Found a JSON metadata file associated with task '{}' in dataset '{}'!".format(args.task, args.db_path))
         md_json = os.path.join(md_path, md_json)
     else:
-        print("Found more than one JSON metadata file associated to task '{}' in dataset '{}':".format(args.task,
+        print("Found more than one JSON metadata file associated with task '{}' in dataset '{}':".format(args.task,
                                                                                                        args.db_path))
         [print(" - {}".format(json_f)) for json_f in json_list]
         md_json = max([os.path.join(md_path, json_f) for json_f in json_list], key=os.path.getctime)

src/romitask/cli/romi_run_task.py

@@ -64,6 +64,8 @@
 from pathlib import Path
 
 import toml
+from dotenv import dotenv_values
+
 from romitask import PIPE_TOML
 from romitask import SCAN_TOML
 from romitask.log import LOG_LEVELS
@@ -115,6 +117,8 @@ def parsing():
                         help="Username for FSDB login.")
     parser.add_argument('--db-password', dest='db_password', type=str, default=None,
                         help="Password for FSDB login.")
+    parser.add_argument('--no-auth', dest='no_auth', action="store_true",
+                        help="Use a database with automatic 'admin' user log in, for testing purposes only.")
 
     # Luigi related arguments:
     luigi = parser.add_argument_group("luigi options")
@@ -533,10 +537,23 @@ def run_task(dataset_path, task, config, **kwargs):
         with open(logging_file_path, 'w') as f:
             f.write(logging_config)
 
-        # - Define environment variables to provide the logging TOML file path to `luigi`:
-        env = {"LUIGI_CONFIG_PARSER": "toml", "LUIGI_CONFIG_PATH": file_path}
-        env.update({'PYOPENCL_CTX': '0'})  # default choice
-
+        # - Define a custom environment variables dictionary
+        # Get any value defined :
+        env = dotenv_values(dataset_path / ".env")
+        # Set the logging TOML file path to `luigi`:
+        env.update({"LUIGI_CONFIG_PARSER": "toml", "LUIGI_CONFIG_PATH": file_path, "MPLBACKEND": "Agg"})
+        # Set the default choice for PyOpenCL context:
+        env.update({'PYOPENCL_CTX': '0'})
+        # Set the database in "no authentication" mode
+        if kwargs.get('no_auth'):
+            # Enable the "no authentication" session manager if not forbidden
+            # 1. by a 'ROMI_DB_NOAUTH' defined in as environment variable
+            # 2. by a 'ROMI_DB_NOAUTH' defined in the dotenv file at the root of the DB
+            env['ROMI_DB_NOAUTH'] = (os.getenv('ROMI_DB_NOAUTH') or env.get('ROMI_DB_NOAUTH', None)) or "1"
+            if env['ROMI_DB_NOAUTH'] == "0":
+                logger.error(f"Disabling authentication on this database if forbidden.")
+                logger.info(f"Use `--db-user` and `--db-password` to pass your credentials to the CLI.")
+                raise ValueError("Credentials required")
         # Add database credentials to environment if provided
         if kwargs.get('db_user') and kwargs.get('db_password'):
             env['ROMI_DB_USER'] = kwargs['db_user']
@@ -551,12 +568,13 @@ def run_task(dataset_path, task, config, **kwargs):
         if kwargs.get('local_scheduler', False):
             cmd.append("--local-scheduler")
 
+        # - Print or Start the configured pipeline:
         if kwargs.get('dry_run', False):
             logger.info(f"Luigi command to call is:\n{cmd}")
         else:
             t_start = time.time()
-            # - Start the configured pipeline:
-            p = subprocess.run(cmd, env={**os.environ, **env}, check=True)
+            # System‑wide variables (`os.environ`) overwrite any duplicate keys from the custom `env` dict
+            p = subprocess.run(cmd, env={**env, **os.environ}, check=True)
             delta = timedelta(seconds=time.time() - t_start)
             delta = str(delta).split('.')[0]  # to get HH:MM:SS
             if p.returncode == 0:
@@ -638,15 +656,15 @@ def _dataset_path_error(args):
                 run_task(dataset_path, args.task, args.config,
                          log_level=args.log_level, luigicmd=args.luigicmd, module=args.module,
                          local_scheduler=args.ls, dry_run=args.dry_run,
-                         db_user=args.db_user, db_password=args.db_password)
+                         no_auth=args.no_auth, db_user=args.db_user, db_password=args.db_password)
             except Exception as e:
                 print(e)
     else:
         ## For the folder:
         run_task(folders, args.task, args.config,
                  log_level=args.log_level, luigicmd=args.luigicmd, module=args.module,
                  local_scheduler=args.ls, dry_run=args.dry_run,
-                 db_user=args.db_user, db_password=args.db_password)
+                 no_auth=args.no_auth, db_user=args.db_user, db_password=args.db_password)
 
 
 if __name__ == '__main__':

src/romitask/task.py

@@ -193,9 +193,11 @@ def parse(self, scan_path):
         scan_id = path[-1]
         # create & connect to `db` if not defined:
         if db is None:  # TODO: cannot change DB during run...
-            db = FSDB(db_path)
+            no_auth = os.getenv("ROMI_DB_NOAUTH")=='1' or False
+            db = FSDB(db_path, no_auth=no_auth)
             db.connect()
-            db.login(username=os.getenv("ROMI_DB_USER"), password=os.getenv("ROMI_DB_PASSWORD"))
+            if not no_auth:
+                db.login(username=os.getenv("ROMI_DB_USER"), password=os.getenv("ROMI_DB_PASSWORD"))
         # Get the scan dataset object or create one & return it
         if db.scan_exists(scan_id):
             scan = db.get_scan(scan_id)
@@ -234,7 +236,7 @@ class FSDBConfiguration(luigi.Config):
     Examples
     --------
     >>> from romitask.task import FSDBConfiguration
-    >>> from plantdb.commons.fsdb import dummy_db
+    >>> from plantdb.commons.test_database import dummy_db
     >>> # - First, let's create a dummy FSDB database to play with:
     >>> db = dummy_db()
     >>> db.connect()
@@ -256,7 +258,7 @@ class ScanConfiguration(luigi.Config):
     Examples
     --------
     >>> from romitask.task import ScanConfiguration
-    >>> from plantdb.commons.fsdb import dummy_db
+    >>> from plantdb.commons.test_database import dummy_db
     >>> # - First, let's create a dummy FSDB database to play with:
     >>> db = dummy_db()
     >>> db.connect()
@@ -292,7 +294,7 @@ class FilesetTarget(luigi.Target):
     --------
     >>> from romitask.task import FilesetTarget
     >>> from plantdb.commons.fsdb import FSDB
-    >>> from plantdb.commons.fsdb import dummy_db
+    >>> from plantdb.commons.test_database import dummy_db
     >>> # - First, let's create a dummy FSDB database to play with:
     >>> db = dummy_db()
     >>> db.connect()