Merge branch 'master' into accessor-pylint-remains

Author: bernhardkaindl

tests/test_accessor.py

@@ -41,3 +41,11 @@ def test_filesystem_accessor_access(self):
 
         a = xcp.accessor.FilesystemAccessor("tests/data/repo/", True)
         self.check_repo_access(a)
+
+
+def test_access_handles_exception():
+    class AccessorHandlesException(xcp.accessor.Accessor):
+        def openAddress(self, address):
+            raise IOError("Test Accessor.access returning False on Exception to cove code")
+
+    assert AccessorHandlesException(True).access("filename") is False

tests/test_ftpaccessor.py

@@ -71,4 +71,5 @@ def test_download_binary_file(ftp_accessor):
     """Download a binary file and compare the returned file content"""
     remote_ftp_filehandle = ftp_accessor.openAddress("filename")
     assert remote_ftp_filehandle.read() == binary_data
+    assert ftp_accessor.access("filename") is True  # covers FTPAccessor._cleanup()
     ftp_accessor.finish()

xcp/accessor.py

@@ -29,11 +29,15 @@
 import os
 import tempfile
 import errno
+from typing import cast, TYPE_CHECKING
 
 from six.moves import urllib  # pyright: ignore
 
-import xcp.mount as mount
-import xcp.logger as logger
+from xcp import logger, mount
+
+if TYPE_CHECKING:
+    from typing import IO
+    from typing_extensions import Literal
 
 # maps errno codes to HTTP error codes
 # needed for error code consistency
@@ -60,15 +64,16 @@ def access(self, name):
             f = self.openAddress(name)
             if not f:
                 return False
-            f.close()
-        except Exception as e:
+            f.close()  # pylint: disable=no-member
+        except Exception:
             return False
 
         return True
 
     def openAddress(self, address):
-        """should be overloaded"""
-        pass
+        # type:(str) -> IO[bytes] | Literal[False]
+        """must be overloaded"""
+        return False  # pragma: no cover
 
     def canEject(self):
         return False
@@ -97,13 +102,7 @@ def __init__(self, location, ro):
     def openAddress(self, address):
         try:
             filehandle = open(os.path.join(self.location, address), "rb")
-        except OSError as e:
-            if e.errno == errno.EIO:
-                self.lastError = 5
-            else:
-                self.lastError = mapError(e.errno)
-            return False
-        except IOError as e:
+        except (IOError, OSError) as e:
             if e.errno == errno.EIO:
                 self.lastError = 5
             else:
@@ -218,13 +217,7 @@ def __init__(self, baseAddress, ro):
     def openAddress(self, address):
         try:
             reader = open(os.path.join(self.baseAddress, address), "rb")
-        except IOError as e:
-            if e.errno == errno.EIO:
-                self.lastError = 5
-            else:
-                self.lastError = mapError(e.errno)
-            return False
-        except OSError as e:
+        except (IOError, OSError) as e:
             if e.errno == errno.EIO:
                 self.lastError = 5
             else:
@@ -265,7 +258,7 @@ def __init__(self, baseAddress, ro):
     def _cleanup(self):
         if self.cleanup:
             # clean up after RETR
-            self.ftp.voidresp()
+            cast(ftplib.FTP, self.ftp).voidresp()
             self.cleanup = False
 
     def start(self):
@@ -275,9 +268,9 @@ def start(self):
             port = ftplib.FTP_PORT
             if self.url_parts.port:
                 port = self.url_parts.port
-            self.ftp.connect(self.url_parts.hostname, port)
-            username = self.url_parts.username
-            password = self.url_parts.password
+            self.ftp.connect(cast(str, self.url_parts.hostname), port)
+            username = cast(str, self.url_parts.username)
+            password = cast(str, self.url_parts.password)
             if username:
                 username = urllib.parse.unquote(username)
             if password:
@@ -296,7 +289,7 @@ def finish(self):
             return
         self.start_count -= 1
         if self.start_count == 0:
-            self.ftp.quit()
+            cast(ftplib.FTP, self.ftp).quit()
             self.cleanup = False
             self.ftp = None
 
@@ -306,17 +299,12 @@ def access(self, path):  # pylint: disable=arguments-differ,arguments-renamed
             self._cleanup()
             url = urllib.parse.unquote(path)
 
+            assert self.ftp
             if self.ftp.size(url) is not None:
                 return True
             lst = self.ftp.nlst(os.path.dirname(url))
             return os.path.basename(url) in list(map(os.path.basename, lst))
-        except IOError as e:
-            if e.errno == errno.EIO:
-                self.lastError = 5
-            else:
-                self.lastError = mapError(e.errno)
-            return False
-        except OSError as e:
+        except (IOError, OSError) as e:
             if e.errno == errno.EIO:
                 self.lastError = 5
             else:
@@ -331,6 +319,7 @@ def openAddress(self, address):
         self._cleanup()
         url = urllib.parse.unquote(address)
 
+        assert self.ftp
         self.ftp.voidcmd('TYPE I')
         socket = self.ftp.transfercmd('RETR ' + url)
         buffered_reader = socket.makefile('rb')
@@ -344,7 +333,7 @@ def writeFile(self, in_fh, out_name):
         fname = urllib.parse.unquote(out_name)
 
         logger.debug("Storing as " + fname)
-        self.ftp.storbinary('STOR ' + fname, in_fh)
+        cast(ftplib.FTP, self.ftp).storbinary('STOR ' + fname, in_fh)
 
     def __repr__(self):
         return "<FTPAccessor: %s>" % self.baseAddress