override: Add manual PEP 484 type annotations

Signed-off-by: Philipp Hahn <[email protected]>

Author: pmhahn

libvirt-override-virConnect.py

@@ -1,4 +1,4 @@
-    def listAllSnapshots(self, flags=0):
+    def listAllSnapshots(self, flags: int = 0) -> List['virDomainSnapshot']:
         """List all snapshots and returns a list of snapshot objects"""
         ret = libvirtmod.virDomainListAllSnapshots(self._o, flags)
         if ret is None:
@@ -10,7 +10,7 @@ def listAllSnapshots(self, flags=0):
 
         return retlist
 
-    def listAllCheckpoints(self, flags=0):
+    def listAllCheckpoints(self, flags: int = 0) -> List['virDomainCheckpoint']:
         """List all checkpoints and returns a list of checkpoint objects"""
         ret = libvirtmod.virDomainListAllCheckpoints(self._o, flags)
         if ret is None:
@@ -22,7 +22,7 @@ def listAllCheckpoints(self, flags=0):
 
         return retlist
 
-    def createWithFiles(self, files, flags=0):
+    def createWithFiles(self, files: List[int], flags: int = 0) -> 'virDomain':
         """Launch a defined domain. If the call succeeds the domain moves from the
         defined to the running domains pools.
 
@@ -60,28 +60,28 @@ def createWithFiles(self, files, flags=0):
             raise libvirtError('virDomainCreateWithFiles() failed')
         return ret
 
-    def fsFreeze(self, mountpoints=None, flags=0):
+    def fsFreeze(self, mountpoints: List[str] = None, flags: int = 0) -> int:
         """Freeze specified filesystems within the guest """
         ret = libvirtmod.virDomainFSFreeze(self._o, mountpoints, flags)
         if ret == -1:
             raise libvirtError('virDomainFSFreeze() failed')
         return ret
 
-    def fsThaw(self, mountpoints=None, flags=0):
+    def fsThaw(self, mountpoints: List[str] = None, flags: int = 0) -> int:
         """Thaw specified filesystems within the guest """
         ret = libvirtmod.virDomainFSThaw(self._o, mountpoints, flags)
         if ret == -1:
             raise libvirtError('virDomainFSThaw() failed')
         return ret
 
-    def getTime(self, flags=0):
+    def getTime(self, flags: int = 0) -> int:
         """Extract information about guest time """
         ret = libvirtmod.virDomainGetTime(self._o, flags)
         if ret == None:
             raise libvirtError('virDomainGetTime() failed')
         return ret
 
-    def setTime(self, time=None, flags=0):
+    def setTime(self, time: int = None, flags: int = 0) -> int:
         """Set guest time to the given value. @time is a dict containing
         'seconds' field for seconds and 'nseconds' field for nanoseconds """
         ret = libvirtmod.virDomainSetTime(self._o, time, flags)

libvirt-override-virDomain.py

@@ -1,12 +1,12 @@
-    def getConnect(self):
+    def getConnect(self) -> 'virConnect':
         """Get the connection that owns the domain that a checkpoint was created for"""
         return self.connect()
 
-    def getDomain(self):
+    def getDomain(self) -> 'virDomain':
         """Get the domain that a checkpoint was created for"""
         return self.domain()
 
-    def listAllChildren(self, flags=0):
+    def listAllChildren(self, flags: int = 0) -> List['virDomainCheckpoint']:
         """List all child checkpoints and returns a list of checkpoint objects"""
         ret = libvirtmod.virDomainCheckpointListAllChildren(self._o, flags)
         if ret is None:

libvirt-override-virDomainCheckpoint.py

@@ -1,12 +1,12 @@
-    def getConnect(self):
+    def getConnect(self) -> 'virConnect':
         """Get the connection that owns the domain that a snapshot was created for"""
         return self.connect()
 
-    def getDomain(self):
+    def getDomain(self) -> 'virDomain':
         """Get the domain that a snapshot was created for"""
         return self.domain()
 
-    def listAllChildren(self, flags=0):
+    def listAllChildren(self, flags: int = 0) -> List['virDomainSnapshot']:
         """List all child snapshots and returns a list of snapshot objects"""
         ret = libvirtmod.virDomainSnapshotListAllChildren(self._o, flags)
         if ret is None:

libvirt-override-virDomainSnapshot.py

@@ -1,4 +1,4 @@
-    def listAllPorts(self, flags=0):
+    def listAllPorts(self, flags: int = 0) -> List['virNetworkPort']:
         """List all ports on the network and returns a list of network port objects"""
         ret = libvirtmod.virNetworkListAllPorts(self._o, flags)
         if ret is None:

libvirt-override-virNetwork.py

@@ -1,4 +1,4 @@
-    def listAllVolumes(self, flags=0):
+    def listAllVolumes(self, flags: int = 0) -> List['virStorageVol']:
         """List all storage volumes and returns a list of storage volume objects"""
         ret = libvirtmod.virStoragePoolListAllVolumes(self._o, flags)
         if ret is None:

libvirt-override-virStoragePool.py

@@ -1,4 +1,4 @@
-    def __del__(self):
+    def __del__(self) -> None:
         try:
             if self.cb:
                 libvirtmod.virStreamEventRemoveCallback(self._o)
@@ -9,7 +9,7 @@ def __del__(self):
             libvirtmod.virStreamFree(self._o)
         self._o = None
 
-    def _dispatchStreamEventCallback(self, events, cbData):
+    def _dispatchStreamEventCallback(self, events: int, cbData: Dict[str, Any]) -> int:
         """
         Dispatches events to python user's stream event callbacks
         """
@@ -19,14 +19,14 @@ def _dispatchStreamEventCallback(self, events, cbData):
         cb(self, events, opaque)
         return 0
 
-    def eventAddCallback(self, events, cb, opaque):
+    def eventAddCallback(self, events: int, cb: Callable[['virStream', int, _T], None], opaque: _T) -> None:
         self.cb = cb
         cbData = {"stream": self, "cb": cb, "opaque": opaque}
         ret = libvirtmod.virStreamEventAddCallback(self._o, events, cbData)
         if ret == -1:
             raise libvirtError('virStreamEventAddCallback() failed')
 
-    def recvAll(self, handler, opaque):
+    def recvAll(self, handler: Callable[['virStream', bytes, _T], int], opaque: _T) -> None:
         """Receive the entire data stream, sending the data to the
         requested data sink. This is simply a convenient alternative
         to virStreamRecv, for apps that do blocking-I/O.
@@ -59,7 +59,7 @@ def handler(stream, # virStream instance
                     pass
                 raise e
 
-    def sendAll(self, handler, opaque):
+    def sendAll(self, handler: Callable[['virStream', int, _T], bytes], opaque: _T) -> None:
         """
         Send the entire data stream, reading the data from the
         requested data source. This is simply a convenient alternative
@@ -92,7 +92,7 @@ def handler(stream, # virStream instance
                 raise libvirtError("cannot use sendAll with "
                                    "nonblocking stream")
 
-    def recv(self, nbytes):
+    def recv(self, nbytes: int) -> bytes:
         """Reads a series of bytes from the stream. This method may
         block the calling application for an arbitrary amount
         of time.
@@ -110,7 +110,7 @@ def recv(self, nbytes):
             raise libvirtError('virStreamRecv() failed')
         return ret
 
-    def send(self, data):
+    def send(self, data: bytes) -> int:
         """Write a series of bytes to the stream. This method may
         block the calling application for an arbitrary amount
         of time. Once an application has finished sending data
@@ -129,7 +129,7 @@ def send(self, data):
             raise libvirtError('virStreamSend() failed')
         return ret
 
-    def recvHole(self, flags=0):
+    def recvHole(self, flags: int = 0) -> int:
         """This method is used to determine the length in bytes
         of the empty space to be created in a stream's target
         file when uploading or downloading sparsely populated
@@ -140,7 +140,7 @@ def recvHole(self, flags=0):
             raise libvirtError('virStreamRecvHole() failed')
         return ret
 
-    def sendHole(self, length, flags=0):
+    def sendHole(self, length: int, flags: int = 0) -> int:
         """Rather than transmitting empty file space, this method
         directs the stream target to create length bytes of empty
         space.  This method would be used when uploading or
@@ -152,7 +152,7 @@ def sendHole(self, length, flags=0):
             raise libvirtError('virStreamSendHole() failed')
         return ret
 
-    def recvFlags(self, nbytes, flags=0):
+    def recvFlags(self, nbytes: int, flags: int = 0) -> Union[bytes, int]:
         """Reads a series of bytes from the stream. This method may
         block the calling application for an arbitrary amount
         of time. This is just like recv except it has flags
@@ -171,7 +171,7 @@ def recvFlags(self, nbytes, flags=0):
             raise libvirtError('virStreamRecvFlags() failed')
         return ret
 
-    def sparseRecvAll(self, handler, holeHandler, opaque):
+    def sparseRecvAll(self, handler: Callable[['virStream', bytes, _T], Union[bytes, int]], holeHandler: Callable[['virStream', int, _T], Optional[int]], opaque: _T) -> None:
         """Receive the entire data stream, sending the data to
         the requested data sink handler and calling the skip
         holeHandler to generate holes for sparse stream targets.
@@ -219,7 +219,7 @@ def holeHandler(stream, # virStream instance
                 self.abort()
                 raise RuntimeError("sparseRecvAll handler returned %d" % ret)
 
-    def sparseSendAll(self, handler, holeHandler, skipHandler, opaque):
+    def sparseSendAll(self, handler: Callable[['virStream', int, _T], Union[bytes, int]], holeHandler: Callable[['virStream', _T], Tuple[bool, int]], skipHandler: Callable[['virStream', int, _T], int], opaque: _T) -> None:
         """Send the entire data stream, reading the data from the
         requested data source. This is simply a convenient
         alternative to virStreamSend, for apps that do
@@ -269,6 +269,7 @@ def skipHandler(stream, # virStream instance
             if not got:
                 break
 
+            assert isinstance(got, bytes)
             ret = self.send(got)
             if ret == -2:
                 raise libvirtError("cannot use sparseSendAll with "