make the two-pass strategy work with subdirectories

CHANGELOG.md

@@ -1,4 +1,7 @@
 # Changelog
+## 3.1.3
+- fixed 2-pass to copy subdirectories
+
 ## 3.1.2
 - fix types to work with python 3.12
 

collectfasta/__init__.py

@@ -1 +1 @@
-__version__ = "3.1.2"
+__version__ = "3.1.3"

collectfasta/management/commands/collectstatic.py

@@ -1,6 +1,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from typing import Any
 from typing import Dict
+from typing import Generator
 from typing import List
 from typing import Optional
 from typing import Tuple
@@ -21,6 +22,22 @@
 Task = Tuple[str, str, Storage]
 
 
+def collect_from_folder(
+    storage: Storage, path: str = ""
+) -> Generator[tuple[str, str], str, None]:
+    folders, files = storage.listdir(path)
+    for thefile in files:
+        if path:
+            prefixed = f"{path}/{thefile}"
+        else:
+            prefixed = thefile
+        yield prefixed, prefixed
+    for folder in folders:
+        if path:
+            folder = f"{path}/{folder}"
+        yield from collect_from_folder(storage, folder)
+
+
 class Command(collectstatic.Command):
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
@@ -73,8 +90,8 @@ def second_pass(self, stats: Dict[str, List[str]]) -> Dict[str, List[str]]:
             self.storage = second_pass_strategy.wrap_storage(self.storage)
             self.strategy = second_pass_strategy
             self.log(f"Running second pass with {self.strategy.__class__.__name__}...")
-            for f in source_storage.listdir("")[1]:
-                self.maybe_copy_file((f, f, source_storage))
+            for f, prefixed in collect_from_folder(source_storage):
+                self.maybe_copy_file((f, prefixed, source_storage))
             return {
                 "modified": self.copied_files + self.symlinked_files,
                 "unmodified": self.unmodified_files,
@@ -117,7 +134,6 @@ def handle(self, *args: Any, **options: Any) -> Optional[str]:
     def maybe_copy_file(self, args: Task) -> None:
         """Determine if file should be copied or not and handle exceptions."""
         path, prefixed_path, source_storage = self.strategy.copy_args_hook(args)
-
         # Build up found_files to look identical to how it's created in the
         # builtin command's collect() method so that we can run post_process
         # after all parallel uploads finish.

collectfasta/tests/command/test_command.py

@@ -217,7 +217,12 @@ def test_calls_post_copy_hook(_case: TestCase, post_copy_hook: mock.MagicMock) -
     post_copy_hook.assert_has_calls(
         [
             mock.call(mock.ANY, path_one.name, path_one.name, mock.ANY),
-            mock.call(mock.ANY, path_two.name, path_two.name, mock.ANY),
+            mock.call(
+                mock.ANY,
+                f"{path_one.name.replace('.html','')}_folder/{path_two.name}",
+                f"{path_one.name.replace('.html','')}_folder/{path_two.name}",
+                mock.ANY,
+            ),
         ],
         any_order=True,
     )
@@ -226,7 +231,7 @@ def test_calls_post_copy_hook(_case: TestCase, post_copy_hook: mock.MagicMock) -
 @make_test_all_backends
 @live_test
 @mock.patch("collectfasta.strategies.base.Strategy.on_skip_hook", autospec=True)
-def test_calls_on_skip_hook(_case: TestCase, on_skip_hook: mock.MagicMock) -> None:
+def test_calls_on_skip_hook(case: TestCase, on_skip_hook: mock.MagicMock) -> None:
     clean_static_dir()
     (path_one, path_two) = create_two_referenced_static_files()
     cmd = Command()
@@ -237,7 +242,12 @@ def test_calls_on_skip_hook(_case: TestCase, on_skip_hook: mock.MagicMock) -> No
     on_skip_hook.assert_has_calls(
         [
             mock.call(mock.ANY, path_one.name, path_one.name, mock.ANY),
-            mock.call(mock.ANY, path_two.name, path_two.name, mock.ANY),
+            mock.call(
+                mock.ANY,
+                f"{path_one.name.replace('.html','')}_folder/{path_two.name}",
+                f"{path_one.name.replace('.html','')}_folder/{path_two.name}",
+                mock.ANY,
+            ),
         ],
         any_order=True,
     )

collectfasta/tests/utils.py

@@ -112,8 +112,10 @@ def test(func: F) -> Type[unittest.TestCase]:
 def create_two_referenced_static_files() -> tuple[pathlib.Path, pathlib.Path]:
     """Create a static file, then another file with a reference to the file"""
     path = create_static_file()
-    reference_path = static_dir / f"{uuid.uuid4().hex}.html"
-    reference_path.write_text(f"{{% static '{path.name}' %}}")
+    folder_path = static_dir / (path.stem + "_folder")
+    folder_path.mkdir()
+    reference_path = folder_path / f"{uuid.uuid4().hex}.html"
+    reference_path.write_text(f"{{% static '../{path.name}' %}}")
     return (path, reference_path)
 
 
@@ -140,17 +142,27 @@ def create_big_static_file() -> pathlib.Path:
 
 
 def clean_static_dir() -> None:
+    clean_static_dir_recurse(static_dir.as_posix())
+    clean_static_dir_recurse(django_settings.AWS_LOCATION)
+
+
+def clean_static_dir_recurse(location: str) -> None:
     try:
-        for filename in os.listdir(django_settings.AWS_LOCATION):
-            file = pathlib.Path(django_settings.AWS_LOCATION) / filename
-            if file.is_file():
-                file.unlink()
+        for filename in os.listdir(location):
+            file = pathlib.Path(location) / filename
+            # don't accidentally wipe the whole drive if someone puts / as location.
+            if (
+                "collectfasta" in str(file.absolute())
+                and ".." not in str(file.as_posix())
+                and len(list(filter(lambda x: x == "/", str(file.absolute())))) > 2
+            ):
+                if file.is_file():
+                    file.unlink()
+                elif file.is_dir():
+                    clean_static_dir_recurse(file.as_posix())
+                    file.rmdir()
     except FileNotFoundError:
         pass
-    for filename in os.listdir(static_dir.as_posix()):
-        file = static_dir / filename
-        if file.is_file():
-            file.unlink()
 
 
 def override_setting(name: str, value: Any) -> Callable[[F], F]: