fix(list_dir): check permissions and consider workspace filters config

antarest/study/storage/explorer_service.py

@@ -40,8 +40,16 @@ def list_dir(
         workspace = get_workspace_from_config(self.config, workspace_name, default_allowed=False)
         directory_path = get_folder_from_workspace(workspace, workspace_directory_path)
         directories = []
-        for child in directory_path.iterdir():
-            if child.is_dir() and not is_study_folder(child) and not should_ignore_folder_for_scan(child):
+        try:
+            children = list(directory_path.iterdir())
+        except PermissionError:
+            children = []  # we don't want to try to read folders we can't access
+        for child in children:
+            if (
+                child.is_dir()
+                and not is_study_folder(child)
+                and not should_ignore_folder_for_scan(child, workspace.filter_in, workspace.filter_out)
+            ):
                 # we don't want to expose the full absolute path on the server
                 child_rel_path = child.relative_to(workspace.path)
                 directories.append(NonStudyFolder(path=child_rel_path, workspace=workspace_name, name=child.name))

antarest/study/storage/rawstudy/watcher.py

@@ -126,7 +126,7 @@ def _rec_scan(
         max_depth: Optional[int] = None,
     ) -> List[StudyFolder]:
         try:
-            if should_ignore_folder_for_scan(path):
+            if should_ignore_folder_for_scan(path, filter_in, filter_out):
                 return []
 
             if (path / "study.antares").exists():
@@ -144,14 +144,7 @@ def _rec_scan(
                         if max_depth is not None:
                             max_depth = max_depth - 1
                         try:
-                            if (
-                                (child.is_dir())
-                                and any([re.search(regex, child.name) for regex in filter_in])
-                                and not any([re.search(regex, child.name) for regex in filter_out])
-                            ):
-                                folders = folders + self._rec_scan(
-                                    child, workspace, groups, filter_in, filter_out, max_depth
-                                )
+                            folders += self._rec_scan(child, workspace, groups, filter_in, filter_out, max_depth)
                         except Exception as e:
                             logger.error(f"Failed to scan dir {child}", exc_info=e)
                 return folders

antarest/study/storage/utils.py

@@ -14,6 +14,7 @@
 import logging
 import math
 import os
+import re
 import shutil
 import tempfile
 import time
@@ -463,7 +464,7 @@ def is_ts_gen_tmp_dir(path: Path) -> bool:
     return path.name.startswith(TS_GEN_PREFIX) and "".join(path.suffixes[-2:]) == TS_GEN_SUFFIX and path.is_dir()
 
 
-def should_ignore_folder_for_scan(path: Path) -> bool:
+def should_ignore_folder_for_scan(path: Path, filter_in: t.List[str], filter_out: t.List[str]) -> bool:
     if is_aw_no_scan(path):
         logger.info(f"No scan directive file found. Will skip further scan of folder {path}")
         return True
@@ -476,4 +477,8 @@ def should_ignore_folder_for_scan(path: Path) -> bool:
         logger.info(f"TS generation temporary folder found. Will skip further scan of folder {path}")
         return True
 
-    return False
+    return not (
+        path.is_dir()
+        and any([re.search(regex, path.name) for regex in filter_in])
+        and not any([re.search(regex, path.name) for regex in filter_out])
+    )

tests/storage/business/test_explorer_service.py

@@ -11,6 +11,7 @@
 # This file is part of the Antares project.
 
 from pathlib import Path
+from unittest.mock import patch
 
 import pytest
 
@@ -23,17 +24,9 @@ def build_config(root: Path) -> Config:
     return Config(
         storage=StorageConfig(
             workspaces={
-                DEFAULT_WORKSPACE_NAME: WorkspaceConfig(path=root / DEFAULT_WORKSPACE_NAME, groups=["toto"]),
-                "diese": WorkspaceConfig(
-                    path=root / "diese",
-                    groups=["tata"],
-                    filter_out=["to_skip.*"],
-                ),
-                "test": WorkspaceConfig(
-                    path=root / "test",
-                    groups=["toto"],
-                    filter_out=["to_skip.*"],
-                ),
+                DEFAULT_WORKSPACE_NAME: WorkspaceConfig(path=root / DEFAULT_WORKSPACE_NAME),
+                "diese": WorkspaceConfig(path=root / "diese", filter_out=[".git", ".*RECYCLE.BIN"]),
+                "test": WorkspaceConfig(path=root / "test"),
             }
         )
     )
@@ -74,6 +67,14 @@ def config_scenario_a(tmp_path: Path) -> Config:
     (f / "AW_NO_SCAN").touch()
     (f / "study.antares").touch()
 
+    d = diese / ".git"
+    d.mkdir(parents=True)
+    (d / "config.txt").touch()
+
+    d = diese / "$RECYCLE.BIN"
+    d.mkdir(parents=True)
+    (d / "trash").touch()
+
     config = build_config(tmp_path)
 
     return config
@@ -84,8 +85,8 @@ def test_list_dir_empty_string(config_scenario_a: Config):
     explorer = Explorer(config_scenario_a)
     result = explorer.list_dir("diese", "")
 
+    # We don't want to see the .git folder or the $RECYCLE.BIN as they were ignored in the workspace config
     assert len(result) == 1
-    workspace_path = config_scenario_a.get_workspace_path(workspace="diese")
     assert result[0] == NonStudyFolder(path=Path("folder"), workspace="diese", name="folder")
 
 
@@ -95,7 +96,6 @@ def test_list_dir_several_subfolders(config_scenario_a: Config):
     result = explorer.list_dir("diese", "folder")
 
     assert len(result) == 3
-    workspace_path = config_scenario_a.get_workspace_path(workspace="diese")
     folder_path = Path("folder")
     assert NonStudyFolder(path=(folder_path / "subfolder1"), workspace="diese", name="subfolder1") in result
     assert NonStudyFolder(path=(folder_path / "subfolder2"), workspace="diese", name="subfolder2") in result
@@ -110,6 +110,15 @@ def test_list_dir_in_empty_folder(config_scenario_a: Config):
     assert len(result) == 0
 
 
+@pytest.mark.unit_test
+def test_list_dir_with_permission_error(config_scenario_a: Config):
+    explorer = Explorer(config_scenario_a)
+    with patch("os.listdir", side_effect=PermissionError("Permission denied")):
+        # asserts the endpoint doesn't fail but rather returns an empty list
+        result = explorer.list_dir("diese", "folder")
+        assert len(result) == 0
+
+
 @pytest.mark.unit_test
 def test_list_workspaces(tmp_path: Path):
     config = build_config(tmp_path)