From: John Mulligan <jmulligan@redhat.com>
Date: Fri, 1 Nov 2024 15:25:35 +0000 (-0400)
Subject: python-common: fix mypy errors in earmarking.py
X-Git-Tag: v20.0.0~647^2~26
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=b77829c45e213d3e984789b903d50d9267be5c74;p=ceph.git

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>
---

diff --git a/src/python-common/ceph/fs/earmarking.py b/src/python-common/ceph/fs/earmarking.py
index c5d4a59a4d5..f4fd4ddf96c 100644
--- 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 List, NamedTuple, Optional, Tuple
+from typing import List, NamedTuple, 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"
@@ -53,11 +65,11 @@ class EarmarkParseError(ValueError):
 
 
 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):
@@ -135,7 +147,7 @@ class CephFSVolumeEarmarking:
         except Exception as e:
             return self._handle_cephfs_error(e, "getting")
 
-    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(