Usa API Win32 per verificare share

src/bakrest/server_check.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import json
 import socket
-import subprocess
+import ctypes
 from dataclasses import asdict, dataclass
 from datetime import datetime, timezone
 from pathlib import Path
@@ -10,6 +10,10 @@ from pathlib import Path
 from .config import BackupConfig
 
 
+INVALID_FILE_ATTRIBUTES = 0xFFFFFFFF
+FILE_ATTRIBUTE_DIRECTORY = 0x00000010
+
+
 @dataclass(frozen=True)
 class ServerStatus:
     reachable: bool
@@ -60,19 +64,15 @@ def check_backup_target(config: BackupConfig) -> ServerStatus:
 
 
 def share_is_reachable(destination: str, timeout_seconds: int) -> bool:
-    probe = destination.rstrip("\\/") + "\\."
+    del timeout_seconds
-    command = ["cmd.exe", "/d", "/c", f'if exist "{probe}" (exit /b 0) else (exit /b 1)']
+    normalized = destination.rstrip("\\/")
     try:
-        completed = subprocess.run(
+        attributes = ctypes.windll.kernel32.GetFileAttributesW(str(normalized))
-            command,
+    except OSError:
-            capture_output=True,
-            text=True,
-            timeout=max(timeout_seconds, 1),
-            check=False,
-        )
-    except (OSError, subprocess.TimeoutExpired):
         return False
-    return completed.returncode == 0
+    if attributes == INVALID_FILE_ATTRIBUTES:
+        return False
+    return bool(attributes & FILE_ATTRIBUTE_DIRECTORY)
 
 
 def write_status(path: Path, status: ServerStatus, pending_count: int) -> None:

tests/test_server_check.py

@@ -1,32 +1,30 @@
 from __future__ import annotations
 
-import subprocess
+from bakrest.server_check import FILE_ATTRIBUTE_DIRECTORY, INVALID_FILE_ATTRIBUTES, share_is_reachable
-
-from bakrest.server_check import share_is_reachable
 
 
-def test_share_is_reachable_returns_true_on_zero_exit(monkeypatch) -> None:
+def test_share_is_reachable_returns_true_for_directory_attributes(monkeypatch) -> None:
-    def fake_run(*args, **kwargs):
+    monkeypatch.setattr(
-        return subprocess.CompletedProcess(args[0], 0)
+        "bakrest.server_check.ctypes.windll.kernel32.GetFileAttributesW",
-
+        lambda path: FILE_ATTRIBUTE_DIRECTORY,
-    monkeypatch.setattr("bakrest.server_check.subprocess.run", fake_run)
+    )
 
     assert share_is_reachable("\\\\server\\share", 3)
 
 
-def test_share_is_reachable_returns_false_on_nonzero_exit(monkeypatch) -> None:
+def test_share_is_reachable_returns_false_for_invalid_attributes(monkeypatch) -> None:
-    def fake_run(*args, **kwargs):
+    monkeypatch.setattr(
-        return subprocess.CompletedProcess(args[0], 1)
+        "bakrest.server_check.ctypes.windll.kernel32.GetFileAttributesW",
-
+        lambda path: INVALID_FILE_ATTRIBUTES,
-    monkeypatch.setattr("bakrest.server_check.subprocess.run", fake_run)
+    )
 
     assert not share_is_reachable("\\\\server\\share", 3)
 
 
-def test_share_is_reachable_returns_false_on_timeout(monkeypatch) -> None:
+def test_share_is_reachable_returns_false_for_non_directory(monkeypatch) -> None:
-    def fake_run(*args, **kwargs):
+    monkeypatch.setattr(
-        raise subprocess.TimeoutExpired(args[0], 3)
+        "bakrest.server_check.ctypes.windll.kernel32.GetFileAttributesW",
-
+        lambda path: 0,
-    monkeypatch.setattr("bakrest.server_check.subprocess.run", fake_run)
+    )
 
     assert not share_is_reachable("\\\\server\\share", 3)