Testing force_new

Author: forman

tests/test_levels.py

@@ -129,6 +129,26 @@ def test_args_validation(self):
                     use_saved_levels=True,
                 )
 
+    def test_force_new(self):
+        source_path = "memory://source.zarr"
+        make_test_dataset(
+            uri=source_path,
+            dims=("time", "lat", "lon"),
+            shape=(3, 1024, 2048),
+            chunks=(1, 128, 256),
+        )
+
+        target_dir = FileObj("memory://target.levels")
+        self.assertFalse(target_dir.exists())
+        target_dir.mkdir()
+        (target_dir / "0.zarr").mkdir()
+        (target_dir / "0.zarr" / ".zgroup").write("{}")
+        self.assertTrue(target_dir.exists())
+
+        write_levels(source_path=source_path, target_path=target_dir.uri, force_new=True)
+
+        self.assertTrue(target_dir.exists())
+
     def test_default_x_y_with_crs(self):
         source_path = "memory://source.zarr"
         make_test_dataset(

zappend/levels.py

@@ -172,14 +172,19 @@ def write_levels(
         variables=zappend_config.pop("variables", None),
     )
 
-    if target_fs.exists(target_root):
-        if target_fs.exists(target_root):
-            logger.warning(f"Permanently deleting {target_dir}")
+    target_exists = target_fs.exists(target_root)
+    if target_exists:
+        logger.info(f"Target directory {target_path} exists")
+        if force_new:
+            logger.warning(f"Permanently deleting {target_path} (no rollback)")
             if not dry_run:
                 target_fs.rm(target_root, recursive=True)
+    else:
+        logger.info(f"Creating target directory {target_path}")
+        if not dry_run:
+            target_fs.mkdirs(target_root, exist_ok=True)
 
     if not dry_run:
-        target_fs.mkdirs(target_root, exist_ok=True)
         with target_fs.open(f"{target_root}/.zlevels", "wt") as fp:
             levels_data: dict[str, Any] = dict(
                 version="1.0",
@@ -189,7 +194,7 @@ def write_levels(
             )
             json.dump(levels_data, fp, indent=2)
 
-    if link_level_zero and not dry_run:
+    if (not dry_run) and link_level_zero:
         path_class = get_fs_path_class(target_fs)
         rel_source_path = (
             "../"