summaryrefslogtreecommitdiffstats
path: root/src/pybind/mgr/smb/proto.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/pybind/mgr/smb/proto.py')
-rw-r--r--src/pybind/mgr/smb/proto.py47
1 files changed, 13 insertions, 34 deletions
diff --git a/src/pybind/mgr/smb/proto.py b/src/pybind/mgr/smb/proto.py
index 96aed6c4174..ffcc647a48e 100644
--- a/src/pybind/mgr/smb/proto.py
+++ b/src/pybind/mgr/smb/proto.py
@@ -9,7 +9,6 @@ from typing import (
List,
Optional,
Tuple,
- TypeVar,
)
import sys
@@ -18,7 +17,7 @@ from ceph.deployment.service_spec import SMBSpec
# this uses a version check as opposed to a try/except because this
# form makes mypy happy and try/except doesn't.
-if sys.version_info >= (3, 8):
+if sys.version_info >= (3, 8): # pragma: no cover
from typing import Protocol
elif TYPE_CHECKING: # pragma: no cover
# typing_extensions will not be available for the real mgr server
@@ -29,7 +28,7 @@ else: # pragma: no cover
pass
-if sys.version_info >= (3, 11):
+if sys.version_info >= (3, 11): # pragma: no cover
from typing import Self
elif TYPE_CHECKING: # pragma: no cover
# typing_extensions will not be available for the real mgr server
@@ -78,13 +77,8 @@ class ConfigEntry(Protocol):
... # pragma: no cover
-class ConfigStore(Protocol):
- """A protocol for describing a configuration data store capable of
- retaining and tracking configuration entry objects.
- """
-
- def __getitem__(self, key: EntryKey) -> ConfigEntry:
- ... # pragma: no cover
+class ConfigStoreListing(Protocol):
+ """A protocol for describing the content-listing methods of a config store."""
def namespaces(self) -> Collection[str]:
... # pragma: no cover
@@ -95,6 +89,15 @@ class ConfigStore(Protocol):
def __iter__(self) -> Iterator[EntryKey]:
... # pragma: no cover
+
+class ConfigStore(ConfigStoreListing, Protocol):
+ """A protocol for describing a configuration data store capable of
+ retaining and tracking configuration entry objects.
+ """
+
+ def __getitem__(self, key: EntryKey) -> ConfigEntry:
+ ... # pragma: no cover
+
def remove(self, ns: EntryKey) -> bool:
... # pragma: no cover
@@ -151,27 +154,3 @@ class AccessAuthorizer(Protocol):
self, volume: str, entity: str, caps: str = ''
) -> None:
... # pragma: no cover
-
-
-T = TypeVar('T')
-
-
-# TODO: move to a utils.py
-def one(lst: List[T]) -> T:
- if len(lst) != 1:
- raise ValueError("list does not contain exactly one element")
- return lst[0]
-
-
-class IsNoneError(ValueError):
- pass
-
-
-def checked(v: Optional[T]) -> T:
- """Ensures the provided value is not a None or raises a IsNoneError.
- Intended use is similar to an `assert v is not None` but more usable in
- one-liners and list/dict/etc comprehensions.
- """
- if v is None:
- raise IsNoneError('value is None')
- return v