New methods in FileManager class:

- get_instance
- get_file_info
- list_tree_os_scandir_bfs
- get_files_and_stats

2 tests for list_tree_os_scandir_bfs

file_manager.py

@@ -2,6 +2,8 @@
 import shutil
 import os
 import logging
+import concurrent.futures
+from collections import deque
 
 logger = logging.getLogger(__name__)
 
@@ -27,6 +29,13 @@ def __new__(cls, run_mode, *args, **kwargs):
             cls._instance.run_mode = run_mode
         return cls._instance
 
+    @classmethod
+    def get_instance(cls):
+        if cls._instance is None:
+            # create a new instance if it doesn't exist - run_mode is False by default. Should not happen in real life
+            cls._instance = cls(False)
+        return cls._instance
+
     def add_protected_dir(self, dir_path):
         protected_dir = Path(dir_path).resolve()
 
@@ -129,12 +138,53 @@ def rmdir(self, dir_path):
             logger.info(f"Would have deleted directory {dir_path}")
         return True
 
+    @staticmethod
+    def get_file_info(file_path):
+        stats = os.stat(file_path)
+        return {
+            'path': file_path,
+            'name': os.path.basename(file_path),
+            'size': stats.st_size,
+            'modified_time': stats.st_mtime,
+            'created_time': stats.st_ctime
+        }
+
+    @staticmethod
+    def list_tree_os_scandir_bfs(directory, raise_on_permission_error=False):
+        queue = deque([directory])
+        while queue:
+            current_dir = queue.popleft()
+            try:
+                with os.scandir(current_dir) as it:
+                    for entry in it:
+                        if entry.is_dir(follow_symlinks=False):
+                            queue.append(entry.path)
+                        else:
+                            yield entry.path
+            except PermissionError:
+                if raise_on_permission_error:
+                    raise
+                else:
+                    continue
+
+    @staticmethod
+    def get_files_and_stats(directory):
+        files_stats = []
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            futures = []
+            for file_path in FileManager.list_tree_os_scandir_bfs(directory):
+                futures.append(executor.submit(FileManager.get_file_info, file_path))
+
+            for future in concurrent.futures.as_completed(futures):
+                files_stats.append(future.result())
+
+        return files_stats
+
     def reset_all(self):
         self.protected_dirs = set()
         self.allowed_dirs = set()
         return self
 
-    # new version for reset_file_manager that gets the protected and allowed directories as arguments
     @staticmethod
     def reset_file_manager(protected_dirs, allowed_dirs, run_mode):
         FileManager._instance = None

tests/test_file_manager.py

@@ -1,5 +1,6 @@
 from tests.helpers_testing import *
 from pathlib import Path
+from file_manager import FileManager
 
 # FileManager suppose to protect some directories from being moved, copied or deleted.
 
@@ -205,3 +206,52 @@ def test_add_protected_dir():
     assert len(fm.protected_dirs) == 2
     assert Path("C:\\").resolve() in fm.protected_dirs
     assert Path("D:\\").resolve() in fm.protected_dirs
+
+
+def get_folder_files_as_set(folder):
+    ret = set()
+    for root, dirs, files in os.walk(folder):
+        for file in files:
+            ret.add(os.path.join(root, file))
+    return ret
+
+
+def test_list_tree_os_scandir_bfs_simple(setup_teardown):
+    source_dir, target_dir, move_to_dir, common_args = setup_teardown
+    setup_test_files(range(1, 6), [2, 3])
+    fm = file_manager.FileManager.get_instance()
+
+    source_files = get_folder_files_as_set(source_dir)
+    source_tree = fm.list_tree_os_scandir_bfs(source_dir)  # result is in the form of full path
+    assert set(source_tree) == source_files
+
+    target_files = get_folder_files_as_set(target_dir)
+    target_tree = fm.list_tree_os_scandir_bfs(target_dir)  # result is in the form of full path
+    assert set(target_tree) == target_files
+
+
+# files only in source - root folder has 3 files and 2 sub folders. each sub folder has some files and 2 sub folders.
+# goes 3 levels deep.
+def test_list_tree_os_scandir_bfs_tree_with_many_subfolders(setup_teardown):
+    source_dir, target_dir, move_to_dir, common_args = setup_teardown
+
+    os.makedirs(os.path.join(source_dir, "sub1"))
+    os.makedirs(os.path.join(source_dir, "sub2"))
+    os.makedirs(os.path.join(source_dir, "sub1", "sub1"))
+    os.makedirs(os.path.join(source_dir, "sub1", "sub2"))
+    os.makedirs(os.path.join(source_dir, "sub2", "sub1"))
+    os.makedirs(os.path.join(source_dir, "sub2", "sub2"))
+
+    copy_files(range(1, 4), source_dir)
+    copy_files(range(1, 3), os.path.join(source_dir, "sub1"))
+    copy_files(range(1, 3), os.path.join(source_dir, "sub2"))
+    copy_files(range(2, 4), os.path.join(source_dir, "sub1", "sub1"))
+    copy_files(range(3, 6), os.path.join(source_dir, "sub1", "sub2"))
+    copy_files(range(2, 5), os.path.join(source_dir, "sub2", "sub1"))
+    copy_files(range(1, 5), os.path.join(source_dir, "sub2", "sub2"))
+
+    fm = file_manager.FileManager.get_instance()
+    source_files = get_folder_files_as_set(source_dir)
+    source_tree = fm.list_tree_os_scandir_bfs(source_dir)  # result is in the form of full path
+    assert set(source_tree) == source_files
+