Fix app API typing and repair confirmation ordering

Author: jamesyc

src/timecapsulesmb/app/ops/deploy.py

@@ -43,6 +43,7 @@
     verify_managed_runtime,
 )
 from timecapsulesmb.device.compat import (
+    DeviceCompatibility,
     is_netbsd4_payload_family,
     payload_family_description,
     render_compatibility_message,
@@ -74,14 +75,14 @@
     payload_verification_error,
     render_flash_runtime_config,
 )
-from timecapsulesmb.services.runtime import load_env_config, resolve_validated_managed_target
+from timecapsulesmb.services.runtime import ManagedTargetState, load_env_config, resolve_validated_managed_target
 from timecapsulesmb.transport.ssh import SshCommandTimeout, SshConnection, SshError
 
 
 ACP_REBOOT_REQUEST_TIMEOUT_SECONDS = 10
 
 
-def require_supported_payload(target, *, allow_unsupported: bool) -> object:
+def require_supported_payload(target: ManagedTargetState, *, allow_unsupported: bool) -> DeviceCompatibility:
     probe_state = target.probe_state
     if probe_state is None:
         raise AppOperationError("Failed to determine remote device OS compatibility.", code="remote_error")
@@ -103,7 +104,7 @@ def load_config_and_target(
     *,
     profile: str,
     include_probe: bool,
-) -> tuple[AppConfig, object]:
+) -> tuple[AppConfig, ManagedTargetState]:
     sink.stage(operation, "load_config")
     config = overlay_request_credentials(load_env_config(env_path=config_path(params)), params)
     sink.stage(operation, "resolve_managed_target")

src/timecapsulesmb/app/ops/maintenance.py

@@ -344,15 +344,15 @@ def observe_reboot_cycle(
 
 def repair_xattrs_operation(params: dict[str, object], sink: EventSink) -> OperationResult:
     operation = "repair-xattrs"
-    dry_run = bool_param(params, "dry_run")
-    sink.stage(operation, "platform_check")
-    if sys.platform != "darwin":
-        raise AppOperationError(
-            "repair-xattrs must be run on macOS because it uses xattr/chflags on the mounted SMB share.",
-            code="validation_failed",
-        )
     sink.stage(operation, "validate_params")
+    dry_run = bool_param(params, "dry_run")
     path = required_path_param(params, "path")
+    recursive = bool_param(params, "recursive", True)
+    max_depth = optional_int_param(params, "max_depth")
+    include_hidden = bool_param(params, "include_hidden")
+    include_time_machine = bool_param(params, "include_time_machine")
+    fix_permissions = bool_param(params, "fix_permissions")
+    verbose = bool_param(params, "verbose")
     if not dry_run:
         require_confirmation(
             params,
@@ -368,17 +368,23 @@ def repair_xattrs_operation(params: dict[str, object], sink: EventSink) -> Opera
             ),
             legacy_names=("confirm_repair",),
         )
+    sink.stage(operation, "platform_check")
+    if sys.platform != "darwin":
+        raise AppOperationError(
+            "repair-xattrs must be run on macOS because it uses xattr/chflags on the mounted SMB share.",
+            code="validation_failed",
+        )
     config = load_optional_env_config(env_path=config_path(params))
     args = argparse.Namespace(
         path=path,
         dry_run=dry_run,
         yes=not dry_run,
-        recursive=bool_param(params, "recursive", True),
-        max_depth=optional_int_param(params, "max_depth"),
-        include_hidden=bool_param(params, "include_hidden"),
-        include_time_machine=bool_param(params, "include_time_machine"),
-        fix_permissions=bool_param(params, "fix_permissions"),
-        verbose=bool_param(params, "verbose"),
+        recursive=recursive,
+        max_depth=max_depth,
+        include_hidden=include_hidden,
+        include_time_machine=include_time_machine,
+        fix_permissions=fix_permissions,
+        verbose=verbose,
     )
     context = RepairExecutionContext(lambda stage: sink.stage(operation, stage))
     stdout_capture = LineLogCapture(lambda message: sink.log(operation, message, level="info"))

src/timecapsulesmb/services/runtime.py

@@ -8,7 +8,7 @@
 from timecapsulesmb.core.config import DEFAULTS, AppConfig, ConfigError, load_app_config, require_valid_app_config
 from timecapsulesmb.core.net import extract_host, ipv4_literal, is_link_local_ipv4, resolve_host_ipv4s
 from timecapsulesmb.core.paths import resolve_app_paths
-from timecapsulesmb.device.compat import require_compatibility
+from timecapsulesmb.device.compat import DeviceCompatibility, require_compatibility
 from timecapsulesmb.device.probe import (
     ProbedDeviceState,
     RemoteInterfaceProbeResult,
@@ -134,7 +134,7 @@ def resolve_validated_managed_target(
     return ManagedTargetState(connection=connection, interface_probe=None, probe_state=probe_state)
 
 
-def require_connection_compatibility(connection: SshConnection):
+def require_connection_compatibility(connection: SshConnection) -> DeviceCompatibility:
     state = probe_connection_state(connection)
     return require_compatibility(
         state.compatibility,

tests/test_app_api.py

@@ -1358,21 +1358,43 @@ def test_repair_xattrs_passes_valid_max_depth_as_int(self) -> None:
     def test_repair_xattrs_requires_confirmation_for_non_dry_run(self) -> None:
         collector = CollectingSink()
 
-        with mock.patch("timecapsulesmb.app.ops.maintenance.repair_xattrs_service.run_repair_structured") as runner:
-            rc = service.run_api_request(
-                {
-                    "operation": "repair-xattrs",
-                    "params": {"path": "/Volumes/Data", "dry_run": False},
-                },
-                collector.sink,
-            )
+        with mock.patch("timecapsulesmb.app.ops.maintenance.sys.platform", "linux"):
+            with mock.patch("timecapsulesmb.app.ops.maintenance.repair_xattrs_service.run_repair_structured") as runner:
+                rc = service.run_api_request(
+                    {
+                        "operation": "repair-xattrs",
+                        "params": {"path": "/Volumes/Data", "dry_run": False},
+                    },
+                    collector.sink,
+                )
 
         self.assertEqual(rc, 1)
         error = self.assert_single_terminal_event(collector, "error")
         self.assertEqual(error["code"], "confirmation_required")
         self.assertEqual(error["recovery"]["title"], "Repair confirmation required")
         runner.assert_not_called()
 
+    def test_repair_xattrs_checks_platform_after_confirmation(self) -> None:
+        collector = CollectingSink()
+
+        with mock.patch("timecapsulesmb.app.ops.maintenance.sys.platform", "linux"):
+            with mock.patch("timecapsulesmb.app.ops.maintenance.load_optional_env_config") as load_config:
+                with mock.patch("timecapsulesmb.app.ops.maintenance.repair_xattrs_service.run_repair_structured") as runner:
+                    rc = service.run_api_request(
+                        {
+                            "operation": "repair-xattrs",
+                            "params": {"path": "/Volumes/Data", "dry_run": False, "confirm_repair": True},
+                        },
+                        collector.sink,
+                    )
+
+        self.assertEqual(rc, 1)
+        error = self.assert_single_terminal_event(collector, "error")
+        self.assertEqual(error["code"], "validation_failed")
+        self.assertEqual(error["recovery"]["title"], "repair-xattrs requires macOS")
+        load_config.assert_not_called()
+        runner.assert_not_called()
+
     def test_helper_reads_request_and_writes_ndjson(self) -> None:
         output = io.StringIO()
         fake_stdin = io.StringIO('{"operation":"paths","params":{}}')