Use abstract properties for child classes if necessary

drivers/blktap2.py

@@ -18,7 +18,9 @@
 # blktap2: blktap/tapdisk management layer
 #
 
-from typing import Any, Callable, ClassVar
+from typing import Any, Callable, override
+
+from abc import ABC, ABCMeta, abstractmethod
 
 import grp
 import os
@@ -514,17 +516,21 @@ def mkdirs(path, mode=0o777):
                 raise
 
 
-class KObject(object):
-
-    SYSFS_CLASSTYPE: ClassVar[str]
+class KObject(ABC):
+    @abstractmethod
+    @property
+    def SYSFS_CLASSTYPE(self) -> str:
+        pass
 
     def sysfs_devname(self):
         raise NotImplementedError("sysfs_devname is undefined")
 
 
-class Attribute(object):
-
-    SYSFS_NODENAME: ClassVar[str]
+class Attribute(ABC):
+    @abstractmethod
+    @property
+    def SYSFS_NODENAME(self) -> str:
+        pass
 
     def __init__(self, path):
         self.path = path
@@ -576,7 +582,10 @@ class Blktap(ClassDevice):
 
     DEV_BASEDIR = '/dev/xen/blktap-2'
 
-    SYSFS_CLASSTYPE = "blktap2"
+    @override
+    @property
+    def SYSFS_CLASSTYPE(self) -> str:
+        return 'blktap2'
 
     def __init__(self, minor):
         self.minor = minor
@@ -602,7 +611,10 @@ def sysfs_devname(self):
         return "blktap!blktap%d" % self.minor
 
     class Pool(Attribute):
-        SYSFS_NODENAME = "pool"
+        @override
+        @property
+        def SYSFS_NODENAME(self) -> str:
+            return 'pool'
 
     def get_pool_attr(self):
         if not self._pool:
@@ -625,7 +637,10 @@ def set_pool(self, pool):
         self.set_pool_name(pool.name)
 
     class Task(Attribute):
-        SYSFS_NODENAME = "task"
+        @override
+        @property
+        def SYSFS_NODENAME(self) -> str:
+            return 'task'
 
     def get_task_attr(self):
         if not self._task:
@@ -1167,14 +1182,17 @@ def get_vdi_type(self):
         def get_vdi_path(self):
             return self.vdi.path
 
-    class Link(object):
+    class Link(ABC):
         """Relink a node under a common name"""
         # NB. We have to provide the device node path during
         # VDI.attach, but currently do not allocate the tapdisk minor
         # before VDI.activate. Therefore those link steps where we
         # relink existing devices under deterministic path names.
 
-        BASEDIR: ClassVar[str]
+        @abstractmethod
+        @property
+        def BASEDIR(self) -> str:
+            pass
 
         def _mklink(self, target):
             raise NotImplementedError("_mklink is not defined")
@@ -1309,15 +1327,24 @@ def _equals(self, target):
             return self._obj._equals(target)
 
     class PhyLink(SymLink):
-        BASEDIR = "/dev/sm/phy"
+        @override
+        @property
+        def BASEDIR(self) -> str:
+            return '/dev/sm/phy'
     # NB. Cannot use DeviceNodes, e.g. FileVDIs aren't bdevs.
 
     class NBDLink(SymLink):
-
-        BASEDIR = "/run/blktap-control/nbd"
+        @override
+        @property
+        def BASEDIR(self) -> str:
+            return '/run/blktap-control/nbd'
 
     class BackendLink(Hybrid):
-        BASEDIR = "/dev/sm/backend"
+        @override
+        @property
+        def BASEDIR(self) -> str:
+            return '/dev/sm/backend'
+
     # NB. Could be SymLinks as well, but saving major,minor pairs in
     # Links enables neat state capturing when managing Tapdisks.  Note
     # that we essentially have a tap-ctl list replacement here. For
@@ -2154,7 +2181,10 @@ def __str__(self):
 
 
 class __BlktapControl(ClassDevice):
-    SYSFS_CLASSTYPE = "misc"
+    @override
+    @property
+    def SYSFS_CLASSTYPE(self) -> str:
+        return 'misc'
 
     def __init__(self):
         ClassDevice.__init__(self)
@@ -2164,7 +2194,10 @@ def sysfs_devname(self):
         return "blktap!control"
 
     class DefaultPool(Attribute):
-        SYSFS_NODENAME = "default_pool"
+        @override
+        @property
+        def SYSFS_NODENAME(self) -> str:
+            return 'default_pool'
 
     def get_default_pool_attr(self):
         if not self._default_pool:
@@ -2211,7 +2244,10 @@ def sysfs_path(self):
         return self.path
 
     class Size(Attribute):
-        SYSFS_NODENAME = "size"
+        @override
+        @property
+        def SYSFS_NODENAME(self) -> str:
+            return 'size'
 
     def get_size_attr(self):
         if not self._size:
@@ -2228,8 +2264,10 @@ def get_size(self):
 
 
 class BusDevice(KObject):
-
-    SYSFS_BUSTYPE: ClassVar[str]
+    @abstractmethod
+    @property
+    def SYSFS_BUSTYPE(self) -> str:
+        pass
 
     @classmethod
     def sysfs_bus_path(cls):
@@ -2247,7 +2285,10 @@ class XenbusDevice(BusDevice):
 
     XBT_NIL = ""
 
-    XENBUS_DEVTYPE: ClassVar[str]
+    @abstractmethod
+    @property
+    def XENBUS_DEVTYPE(self) -> str:
+        pass
 
     def __init__(self, domid, devid):
         self.domid = int(domid)
@@ -2368,7 +2409,11 @@ def find(cls):
 
 class XenBackendDevice(XenbusDevice):
     """Xenbus backend device"""
-    SYSFS_BUSTYPE = "xen-backend"
+
+    @override
+    @property
+    def SYSFS_BUSTYPE(self) -> str:
+        return 'xen-backend'
 
     @classmethod
     def from_xs_path(cls, _path):
@@ -2386,7 +2431,10 @@ def from_xs_path(cls, _path):
 class Blkback(XenBackendDevice):
     """A blkback VBD"""
 
-    XENBUS_DEVTYPE = "vbd"
+    @override
+    @property
+    def XENBUS_DEVTYPE(self) -> str:
+        return 'vbd'
 
     def __init__(self, domid, devid):
         XenBackendDevice.__init__(self, domid, devid)
@@ -2395,8 +2443,11 @@ def __init__(self, domid, devid):
         self._q_state = None
         self._q_events = None
 
-    class XenstoreValueError(Exception):
-        KEY: ClassVar[str]
+    class XenstoreValueError(Exception, metaclass=ABCMeta):
+        @abstractmethod
+        @property
+        def KEY(self) -> str:
+            pass
 
         def __init__(self, vbd, _str):
             self.vbd = vbd
@@ -2407,7 +2458,10 @@ def __str__(self):
                 "has %s = %s" % (self.KEY, self.str)
 
     class PhysicalDeviceError(XenstoreValueError):
-        KEY = "physical-device"
+        @override
+        @property
+        def KEY(self) -> str:
+            return 'physical-device'
 
     class PhysicalDevice(object):
 
@@ -2452,7 +2506,10 @@ class QueueEvents(Attribute):
         """Blkback sysfs node to select queue-state event
         notifications emitted."""
 
-        SYSFS_NODENAME = "queue_events"
+        @override
+        @property
+        def SYSFS_NODENAME(self) -> str:
+            return 'queue_events'
 
         QUEUE_RUNNING = (1 << 0)
         QUEUE_PAUSE_DONE = (1 << 1)

drivers/flock.py

@@ -23,7 +23,9 @@
 got to grow our own.
 """
 
-from typing import ClassVar
+from typing import override
+
+from abc import ABC, abstractmethod
 
 import os
 import fcntl
@@ -70,12 +72,15 @@ def __setattr__(self, name, value):
             self.fields[idx] = value
 
 
-class FcntlLockBase:
+class FcntlLockBase(ABC):
     """Abstract base class for either reader or writer locks. A respective
     definition of LOCK_TYPE (fcntl.{F_RDLCK|F_WRLCK}) determines the
     type."""
 
-    LOCK_TYPE: ClassVar[int]
+    @abstractmethod
+    @property
+    def LOCK_TYPE(self) -> int:
+        pass
 
     if __debug__:
         ERROR_ISLOCKED = "Attempt to acquire lock held."
@@ -134,9 +139,16 @@ def test(self):
 
 class WriteLock(FcntlLockBase):
     """A simple global writer (i.e. exclusive) lock."""
-    LOCK_TYPE = fcntl.F_WRLCK
 
+    @override
+    @property
+    def LOCK_TYPE(self) -> int:
+        return fcntl.F_WRLCK
 
 class ReadLock(FcntlLockBase):
     """A simple global reader (i.e. shared) lock."""
-    LOCK_TYPE = fcntl.F_RDLCK
+
+    @override
+    @property
+    def LOCK_TYPE(self) -> int:
+        return fcntl.F_RDLCK