From 04effd01a62ab814b936586c15565cc23a35d828 Mon Sep 17 00:00:00 2001 From: John Mulligan Date: Fri, 1 Nov 2024 11:25:35 -0400 Subject: [PATCH] squid: 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 Signed-off-by: Avan Thakkar (cherry picked from commit b77829c45e213d3e984789b903d50d9267be5c74) --- src/python-common/ceph/fs/earmarking.py | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/src/python-common/ceph/fs/earmarking.py b/src/python-common/ceph/fs/earmarking.py index 3d11da933397f..ba0e955f9dd4e 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 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( -- 2.39.5