feat: files copy method

core/fields.py

@@ -16,6 +16,7 @@ class Field:
     choices: tuple[str, ...] = ()
     help: str | None = None
     validator: Callable[[Any], Any] | None = None
+    when: Callable[[dict[str, Any]], bool] | None = None
 
     @property
     def flag(self) -> str:

engines/files.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import posixpath
+import re
 from typing import Any
 
 from core.errors import ValidationError
@@ -10,6 +11,7 @@
 from services.ports import PortAllocator
 
 DATA_DIR = "/data"
+SIZE_PATTERN = re.compile(r"\d+(\.\d+)?[bkmgtp]?", re.IGNORECASE | re.ASCII)
 
 
 def host_dir(value: str) -> str:
@@ -36,14 +38,38 @@ def no_braces(patterns: list[str]) -> list[str]:
     return patterns
 
 
+def file_count(value: str) -> int | str:
+    count = value.strip()
+    if not count:
+        return ""
+    if not count.isdecimal():
+        raise ValidationError(f"Must be a whole number >= 0, got {value!r}")
+    return int(count)
+
+
+def file_size(value: str) -> str:
+    size = value.strip()
+    if size and not SIZE_PATTERN.fullmatch(size):
+        raise ValidationError(
+            f"Invalid size {value!r}",
+            hint="e.g. 500M or 10G (B, K, M, G, T or P); no suffix means KiB.",
+        )
+    return size
+
+
+def sync_only(answers: dict[str, Any]) -> bool:
+    return answers.get("method") == "sync"
+
+
 class FilesEngine(DbEngine):
     key, display = "files", "Files"
     template = None
     has_modes = False
     label_default = "Files"
     warning = (
         f"The directory is mounted read-write under {DATA_DIR}: a restore replaces "
-        "its content (excluded paths are kept). Requires Agent 1.24.0+."
+        "its content (excluded paths are kept). Requires Agent 1.24.0+; the copy "
+        "method and the sync delete limits require Agent 1.25.0+."
     )
 
     def fields_existing(self) -> list[Field]:
@@ -66,12 +92,15 @@ def option_fields(self) -> list[Field]:
                 "Backup method",
                 "choice",
                 default="archive",
-                choices=("archive", "snapshot", "sync"),
+                choices=("archive", "snapshot", "sync", "copy"),
                 help=(
                     "archive: a tar.gz archive per run. snapshot: incremental, "
                     "deduplicated restic snapshots. sync: mirrors the directory onto "
                     "each storage; deletions propagate and it cannot be restored from "
-                    "Portabase. Cannot be changed once the source exists."
+                    "Portabase. copy: like sync, but files removed from the source are "
+                    "kept on the storage; changed files are overwritten, with no "
+                    "history, and it cannot be restored from Portabase. Cannot be "
+                    "changed once the source exists."
                 ),
             ),
             Field(
@@ -93,6 +122,31 @@ def option_fields(self) -> list[Field]:
                 default=False,
                 help="Skip mount points found inside the directory.",
             ),
+            Field(
+                "max_delete",
+                "Max deletions per run",
+                "text",
+                default="",
+                validator=file_count,
+                when=sync_only,
+                help=(
+                    "Sync only. A run that would delete more files, or more data, "
+                    "than these limits fails before deleting anything. Leave empty "
+                    "for no limit."
+                ),
+            ),
+            Field(
+                "max_delete_size",
+                "Max deleted size per run",
+                "text",
+                default="",
+                validator=file_size,
+                when=sync_only,
+                help=(
+                    "Sync only. A size such as 500M or 10G (B, K, M, G, T or P); a "
+                    "number without suffix is in KiB. Leave empty for no limit."
+                ),
+            ),
         ]
 
     def generate(

tests/core/fields.py

@@ -7,13 +7,14 @@
 
 def defaults():
     field = Field("name", "Prompt")
-    assert (field.kind, field.default, field.choices, field.help, field.validator) == (
-        "text",
-        None,
-        (),
-        None,
-        None,
-    )
+    assert (
+        field.kind,
+        field.default,
+        field.choices,
+        field.help,
+        field.validator,
+        field.when,
+    ) == ("text", None, (), None, None, None)
 
 
 @pytest.mark.parametrize(

tests/engines/files.py

@@ -1,13 +1,18 @@
+from typing import cast
+
 import pytest
 
 from core.errors import ValidationError
 from core.specs import DatabaseSpec
 from engines import registry
 from engines.files import FilesEngine
 from tests.support import agent_service, field_specs
+from ui.components.prompt import Prompt
+from ui.form import Form
 
 FILES = registry.get("files")
 MOUNT = "/srv/uploads:/data/srv/uploads"
+FORM = Form(cast(Prompt, None), non_interactive=True)
 
 
 def attributes():
@@ -16,6 +21,7 @@ def attributes():
     assert FILES.template is None
     assert (FILES.auth_variants, FILES.has_modes) == (False, False)
     assert "restore" in (FILES.warning or "")
+    assert "1.25.0" in (FILES.warning or "")
 
 
 def generate(ports):
@@ -46,6 +52,14 @@ def agent_entry():
         "method": "snapshot",
         "exclude": ["*.tmp"],
     }
+    limited = bare.with_options(
+        {"method": "sync", "max_delete": 100, "max_delete_size": "10G"}
+    )
+    assert FILES.agent_entry(limited)["options"] == {
+        "method": "sync",
+        "max_delete": 100,
+        "max_delete_size": "10G",
+    }
 
 
 def fields():
@@ -55,8 +69,10 @@ def fields():
         ("method", "choice", "archive"),
         ("exclude", "list", []),
         ("one_file_system", "bool", False),
+        ("max_delete", "text", ""),
+        ("max_delete_size", "text", ""),
     ]
-    assert FILES.option_fields()[0].choices == ("archive", "snapshot", "sync")
+    assert FILES.option_fields()[0].choices == ("archive", "snapshot", "sync", "copy")
 
 
 def from_existing():
@@ -112,3 +128,40 @@ def exclude_rejects_brace_alternation():
 def mount_for(path, mount):
     spec = DatabaseSpec(id="1", engine="files", name="n", path=path)
     assert FilesEngine.mount_for(spec) == mount
+
+
+def delete_limits_are_written_only_when_set():
+    options = FORM.collect(
+        FILES.option_fields(),
+        {"method": "sync", "max_delete": " 0 ", "max_delete_size": "1.5g"},
+    )
+    assert (options["max_delete"], options["max_delete_size"]) == (0, "1.5g")
+
+    unset = FORM.collect(FILES.option_fields(), {"method": "sync"})
+    spec = FILES.from_existing({"path": "/srv"}).with_options(unset)
+    assert FILES.agent_entry(spec)["options"] == {"method": "sync"}
+
+
+@pytest.mark.parametrize("method", ["archive", "snapshot", "copy"])
+def delete_limits_only_apply_to_sync(method):
+    options = FORM.collect(FILES.option_fields(), {"method": method})
+    assert "max_delete" not in options
+    assert "max_delete_size" not in options
+    with pytest.raises(ValidationError, match="--max-delete does not apply"):
+        FORM.collect(FILES.option_fields(), {"method": method, "max_delete": "5"})
+
+
+@pytest.mark.parametrize(
+    ("option", "value"),
+    [
+        ("max_delete", "-1"),
+        ("max_delete", "1.5"),
+        ("max_delete", "ten"),
+        ("max_delete_size", "10GB"),
+        ("max_delete_size", "G"),
+        ("max_delete_size", "1,5G"),
+    ],
+)
+def delete_limits_reject_invalid_values(option, value):
+    with pytest.raises(ValidationError):
+        FORM.collect(FILES.option_fields(), {"method": "sync", option: value})

ui/form.py

@@ -48,7 +48,16 @@ def ask(self, field: Field, value: Any | None = None) -> Any:
     def collect(
         self, fields: Sequence[Field], values: dict[str, Any]
     ) -> dict[str, Any]:
-        return {field.name: self.ask(field, values.get(field.name)) for field in fields}
+        answers: dict[str, Any] = {}
+        for field in fields:
+            value = values.get(field.name)
+            if field.when is None or field.when(answers):
+                answers[field.name] = self.ask(field, value)
+            elif value is not None:
+                raise ValidationError(
+                    f"{field.flag} does not apply here.", hint=field.help
+                )
+        return answers
 
     def text(
         self, prompt: str, *, value=None, default=None, validator=None, name="value"