python-common: fix mypy errors in earmarking.py

Fix various errors found by running mypy with python 3.12 on the
python-common subtree. Uses a Protocol as a stand-in for actual file
system integration objects.

Part of an effort to get ceph tox environments passing on Python 3.12.

Signed-off-by: John Mulligan <jmulligan@redhat.com>
(cherry picked from commit b77829c45e213d3e984789b903d50d9267be5c74)

diff --git a/src/python-common/ceph/fs/earmarking.py b/src/python-common/ceph/fs/earmarking.py
index 3d11da933397fcf8368aad7aaaeeca6879b66b6c..ba0e955f9dd4e6ec3a8d6f311a7d380bfa5bcd32 100644 (file)
--- a/src/python-common/ceph/fs/earmarking.py
+++ b/src/python-common/ceph/fs/earmarking.py
@@ -19,13 +19,25 @@ supported top-level scopes.
 import errno
 import enum
 import logging
-from typing import Optional, Tuple
+from typing import Optional, Tuple, Protocol
 
 log = logging.getLogger(__name__)
 
 XATTR_SUBVOLUME_EARMARK_NAME = 'user.ceph.subvolume.earmark'
 
 
+class FSOperations(Protocol):
+    """Protocol class representing the file system operations earmarking
+    classes will perform.
+    """
+
+    def setxattr(
+        self, path: str, key: str, value: bytes, flags: int
+    ) -> None: ...
+
+    def getxattr(self, path: str, key: str) -> bytes: ...
+
+
 class EarmarkTopScope(enum.Enum):
     NFS = "nfs"
     SMB = "smb"
@@ -44,11 +56,11 @@ class EarmarkException(Exception):
 
 
 class CephFSVolumeEarmarking:
-    def __init__(self, fs, path: str) -> None:
+    def __init__(self, fs: FSOperations, path: str) -> None:
         self.fs = fs
         self.path = path
 
-    def _handle_cephfs_error(self, e: Exception, action: str) -> None:
+    def _handle_cephfs_error(self, e: Exception, action: str) -> Optional[str]:
         if isinstance(e, ValueError):
             raise EarmarkException(errno.EINVAL, f"Invalid earmark specified: {e}") from e
         elif isinstance(e, OSError):
@@ -88,7 +100,7 @@ class CephFSVolumeEarmarking:
             self._handle_cephfs_error(e, "getting")
             return None
 
-    def set_earmark(self, earmark: str):
+    def set_earmark(self, earmark: str) -> None:
         # Validate the earmark before attempting to set it
         if not self._validate_earmark(earmark):
             raise EarmarkException(