TDD Py3 merge: Commits up to test_accessor.py: Test reading binary data from http and file

requirements-dev.txt

@@ -5,6 +5,10 @@ diff_cover
 mock
 pytest
 pytest-cov
+parameterized
 # dependencies also in setup.py until they can be used
 six
 future
+
+# python-2.7 only
+configparser ; python_version < "3.0"

tests/test_accessor.py

@@ -1,21 +1,46 @@
 import unittest
+import hashlib
+from tempfile import NamedTemporaryFile
+
+from parameterized import parameterized_class
 
 import xcp.accessor
 
+@parameterized_class([{"url": "file://tests/data/repo/"},
+                      {"url": "https://updates.xcp-ng.org/netinstall/8.2.1"}])
 class TestAccessor(unittest.TestCase):
-    def test_http(self):
-        raise unittest.SkipTest("comment out if you really mean it")
-        a = xcp.accessor.createAccessor("https://updates.xcp-ng.org/netinstall/8.2.1", True)
-        a.start()
-        self.assertTrue(a.access('.treeinfo'))
+    url = ""
+
+    def setup_accessor(self, arg0):
+        result = xcp.accessor.createAccessor(self.url, True)
+        result.start()
+        self.assertTrue(result.access(arg0))
+        return result
+
+    def test_access(self):
+        a = self.setup_accessor('.treeinfo')
         self.assertFalse(a.access('no_such_file'))
         self.assertEqual(a.lastError, 404)
         a.finish()
 
-    def test_file(self):
-        a = xcp.accessor.createAccessor("file://tests/data/repo/", True)
-        a.start()
-        self.assertTrue(a.access('.treeinfo'))
-        self.assertFalse(a.access('no_such_file'))
-        self.assertEqual(a.lastError, 404)
+    def test_file_binfile(self):
+        binary_file_in_repo = "boot/isolinux/mboot.c32"
+        a = self.setup_accessor(binary_file_in_repo)
+        inf = a.openAddress(binary_file_in_repo)
+        with NamedTemporaryFile("wb") as outf:
+            while True:
+                data = inf.read()
+                if not data: # EOF
+                    break
+                outf.write(data)
+            outf.flush()
+            hasher = hashlib.md5()
+            with open(outf.name, "rb") as bincontents:
+                while True:
+                    data = bincontents.read()
+                    if not data: # EOF
+                        break
+                    hasher.update(data)
+                csum = hasher.hexdigest()
+                self.assertEqual(csum, "eab52cebc3723863432dc672360f6dac")
         a.finish()

tests/test_cpio.py

@@ -14,11 +14,11 @@ def writeRandomFile(fn, size, start=b'', add=b'a'):
     with open(fn, 'wb') as f:
         m = md5()
         m.update(start)
-        assert(len(add) != 0)
+        assert add
         while size > 0:
             d = m.digest()
             if size < len(d):
-                d=d[:size]
+                d = d[:size]
             f.write(d)
             size -= len(d)
             m.update(add)

tests/test_dom0.py

@@ -30,7 +30,7 @@ def mock_version(open_mock, version):
             (2*1024, 4*1024, 8*1024), # Above max
             ]
 
-        with patch("__builtin__.open") as open_mock:
+        with patch("xcp.dom0.open") as open_mock:
             for host_gib, dom0_mib, _ in test_values:
                 mock_version(open_mock, '2.8.0')
                 expected = dom0_mib * 1024;
@@ -39,7 +39,7 @@ def mock_version(open_mock, version):
 
             open_mock.assert_called_with("/etc/xensource-inventory")
 
-        with patch("__builtin__.open") as open_mock:
+        with patch("xcp.dom0.open") as open_mock:
             for host_gib, _, dom0_mib in test_values:
                 mock_version(open_mock, '2.9.0')
                 expected = dom0_mib * 1024;

tests/test_ifrename_dynamic.py

@@ -125,10 +125,10 @@ def test_pci_matching_invert(self):
                      MACPCI("c8:cb:b8:d3:0c:cf", "0000:04:00.0", kname="eth1",
                             ppn="", label="")])
 
-        self.assertEqual(dr.rules,[
+        self.assertEqual(set(dr.rules), set([
                 MACPCI("c8:cb:b8:d3:0c:ce", "0000:04:00.0", tname="eth1"),
                 MACPCI("c8:cb:b8:d3:0c:cf", "0000:04:00.0", tname="eth0")
-                ])
+        ]))
 
     def test_pci_missing(self):
 

tests/test_ifrename_logic.py

@@ -518,6 +518,7 @@ def test_ibft_nic_to_ibft(self):
 
 
 class TestInputSanitisation(unittest.TestCase):
+    # pylint: disable=no-member
 
     def setUp(self):
         """

tests/test_ifrename_static.py

@@ -375,10 +375,10 @@ def test_pci_matching(self):
 
         sr.generate(self.state)
 
-        self.assertEqual(sr.rules,[
-                MACPCI("c8:cb:b8:d3:0c:cf", "0000:04:00.0", tname="eth1"),
-                MACPCI("c8:cb:b8:d3:0c:ce", "0000:04:00.0", tname="eth0")
-                ])
+        self.assertEqual(set(sr.rules), set([
+            MACPCI("c8:cb:b8:d3:0c:cf", "0000:04:00.0", tname="eth1"),
+            MACPCI("c8:cb:b8:d3:0c:ce", "0000:04:00.0", tname="eth0")
+        ]))
 
     def test_pci_matching_invert(self):
 
@@ -389,10 +389,10 @@ def test_pci_matching_invert(self):
 
         sr.generate(self.state)
 
-        self.assertEqual(sr.rules,[
-                MACPCI("c8:cb:b8:d3:0c:ce", "0000:04:00.0", tname="eth1"),
-                MACPCI("c8:cb:b8:d3:0c:cf", "0000:04:00.0", tname="eth0")
-                ])
+        self.assertEqual(set(sr.rules), set([
+            MACPCI("c8:cb:b8:d3:0c:ce", "0000:04:00.0", tname="eth1"),
+            MACPCI("c8:cb:b8:d3:0c:cf", "0000:04:00.0", tname="eth0")
+        ]))
 
     def test_pci_matching_mixed(self):
 
@@ -403,10 +403,10 @@ def test_pci_matching_mixed(self):
 
         sr.generate(self.state)
 
-        self.assertEqual(sr.rules,[
-                MACPCI("c8:cb:b8:d3:0c:cf", "0000:04:00.0", tname="eth0"),
-                MACPCI("c8:cb:b8:d3:0c:ce", "0000:04:00.0", tname="eth1")
-                ])
+        self.assertEqual(set(sr.rules), set([
+            MACPCI("c8:cb:b8:d3:0c:cf", "0000:04:00.0", tname="eth0"),
+            MACPCI("c8:cb:b8:d3:0c:ce", "0000:04:00.0", tname="eth1")
+        ]))
 
     def test_pci_missing(self):
 

tests/test_repository.py

@@ -6,7 +6,7 @@
 
 class TestRepository(unittest.TestCase):
     def test_http(self):
-        raise unittest.SkipTest("comment out if you really mean it")
+        #raise unittest.SkipTest("comment out if you really mean it")
         a = xcp.accessor.createAccessor("https://updates.xcp-ng.org/netinstall/8.2.1", True)
         repo_ver = repository.BaseRepository.getRepoVer(a)
         self.assertEqual(repo_ver, Version([3, 2, 1]))

xcp/accessor.py

@@ -71,7 +71,7 @@ def access(self, name):
 
         return True
 
-    def openAddress(self, address):
+    def openAddress(self, address, mode="", **kwargs):
         """should be overloaded"""
         pass
 
@@ -99,9 +99,9 @@ def __init__(self, location, ro):
         super(FilesystemAccessor, self).__init__(ro)
         self.location = location
 
-    def openAddress(self, address):
+    def openAddress(self, address, mode="rb", **kwargs):
         try:
-            filehandle = open(os.path.join(self.location, address), 'r')
+            filehandle = open(os.path.join(self.location, address), mode, **kwargs)
         except OSError as e:
             if e.errno == errno.EIO:
                 self.lastError = 5
@@ -165,9 +165,9 @@ def finish(self):
             os.rmdir(self.location)
             self.location = None
 
-    def writeFile(self, in_fh, out_name):
+    def writeFile(self, in_fh, out_name, mode="wb", **kwargs):
         logger.info("Copying to %s" % os.path.join(self.location, out_name))
-        out_fh = open(os.path.join(self.location, out_name), 'w')
+        out_fh = open(os.path.join(self.location, out_name), mode, **kwargs)
         return self._writeFile(in_fh, out_fh)
 
     def __del__(self):
@@ -220,9 +220,9 @@ def __init__(self, baseAddress, ro):
         super(FileAccessor, self).__init__(ro)
         self.baseAddress = baseAddress
 
-    def openAddress(self, address):
+    def openAddress(self, address, mode="rb", **kwargs):
         try:
-            file = open(os.path.join(self.baseAddress, address))
+            file = open(os.path.join(self.baseAddress, address), mode, **kwargs)
         except IOError as e:
             if e.errno == errno.EIO:
                 self.lastError = 5
@@ -240,9 +240,9 @@ def openAddress(self, address):
             return False
         return file
 
-    def writeFile(self, in_fh, out_name):
+    def writeFile(self, in_fh, out_name, mode="wb", **kwargs):
         logger.info("Copying to %s" % os.path.join(self.baseAddress, out_name))
-        out_fh = open(os.path.join(self.baseAddress, out_name), 'w')
+        out_fh = open(os.path.join(self.baseAddress, out_name), mode, **kwargs)
         return self._writeFile(in_fh, out_fh)
 
     def __repr__(self):
@@ -331,13 +331,13 @@ def access(self, path):
             self.lastError = 500
             return False
 
-    def openAddress(self, address):
+    def openAddress(self, address, mode="rb", **kwargs):
         logger.debug("Opening "+address)
         self._cleanup()
         url = urllib.parse.unquote(address)
 
         self.ftp.voidcmd('TYPE I')
-        s = self.ftp.transfercmd('RETR ' + url).makefile('rb')
+        s = self.ftp.transfercmd('RETR ' + url).makefile(mode, **kwargs)
         self.cleanup = True
         return s
 

xcp/bootloader.py

@@ -336,19 +336,19 @@ def create_label(title):
         try:
             for line in fh:
                 l = line.strip()
-                menu_match = re.match("menuentry ['\"]([^']*)['\"](.*){", l)
+                menu_match = re.match(r"menuentry ['\"]([^']*)['\"](.*){", l)
 
                 # Only parse unindented default and timeout lines to prevent
                 # changing these lines in if statements.
                 if l.startswith('set default=') and l == line.rstrip():
                     default = l.split('=')[1]
-                    match = re.match("['\"](.*)['\"]$", default)
+                    match = re.match(r"['\"](.*)['\"]$", default)
                     if match:
                         default = match.group(1)
                 elif l.startswith('set timeout=') and l == line.rstrip():
                     timeout = int(l.split('=')[1]) * 10
                 elif l.startswith('serial'):
-                    match = re.match("serial --unit=(\d+) --speed=(\d+)", l)
+                    match = re.match(r"serial --unit=(\d+) --speed=(\d+)", l)
                     if match:
                         serial = {
                             'port': int(match.group(1)),

xcp/cmd.py

@@ -24,15 +24,16 @@
 """Command processing"""
 
 import subprocess
+import six
 
 import xcp.logger as logger
 
 def runCmd(command, with_stdout = False, with_stderr = False, inputtext = None):
-    cmd = subprocess.Popen(command, bufsize = 1,
-                           stdin = (inputtext and subprocess.PIPE or None),
-                           stdout = subprocess.PIPE,
-                           stderr = subprocess.PIPE,
-                           shell = isinstance(command, str))
+    cmd = subprocess.Popen(command, bufsize=1,
+                           stdin=(inputtext and subprocess.PIPE or None),
+                           stdout=subprocess.PIPE,
+                           stderr=subprocess.PIPE,
+                           shell=isinstance(command, six.string_types))
 
     (out, err) = cmd.communicate(inputtext)
     rv = cmd.returncode

xcp/cpiofile.py

@@ -310,7 +310,7 @@ def _init_write_gz(self):
         self.__write(b"\037\213\010\010%s\002\377" % timestamp)
         if self.name.endswith(".gz"):
             self.name = self.name[:-3]
-        self.__write(self.name + NUL)
+        self.__write(six.ensure_binary(self.name) + NUL)
 
     def write(self, s):
         """Write string s to the stream.
@@ -951,7 +951,7 @@ def __init__(self, name=None, mode="r", fileobj=None):
         self.mode = {"r": "rb", "a": "r+b", "w": "wb"}[mode]
 
         if not fileobj:
-            fileobj = file(name, self.mode)
+            fileobj = bltn_open(name, self.mode)
             self._extfileobj = False
         else:
             if name is None and hasattr(fileobj, "name"):
@@ -1109,7 +1109,7 @@ def gzopen(cls, name, mode="r", fileobj=None, compresslevel=9):
             raise CompressionError("gzip module is not available")
 
         if fileobj is None:
-            fileobj = file(name, mode + "b")
+            fileobj = bltn_open(name, mode + "b")
 
         try:
             t = cls.cpioopen(name, mode, gzip.GzipFile(name, mode, compresslevel, fileobj))
@@ -1354,7 +1354,7 @@ def add(self, name, arcname=None, recursive=True):
 
         # Append the cpio header and data to the archive.
         if cpioinfo.isreg():
-            f = file(name, "rb")
+            f = bltn_open(name, "rb")
             self.addfile(cpioinfo, f)
             f.close()
 
@@ -1420,20 +1420,20 @@ def extractall(self, path=".", members=None):
                 # Extract directory with a safe mode, so that
                 # all files below can be extracted as well.
                 try:
-                    os.makedirs(os.path.join(path, cpioinfo.name), 0o777)
+                    os.makedirs(os.path.join(path, six.ensure_text(cpioinfo.name)), 0o777)
                 except EnvironmentError:
                     pass
                 directories.append(cpioinfo)
             else:
                 self.extract(cpioinfo, path)
 
         # Reverse sort directories.
-        directories.sort(lambda a, b: cmp(a.name, b.name))
+        directories.sort(key=lambda x: x.name)
         directories.reverse()
 
         # Set correct owner, mtime and filemode on directories.
         for cpioinfo in directories:
-            path = os.path.join(path, cpioinfo.name)
+            path = os.path.join(path, six.ensure_text(cpioinfo.name))
             try:
                 self.chown(cpioinfo, path)
                 self.utime(cpioinfo, path)
@@ -1462,7 +1462,7 @@ def extract(self, member, path=""):
             cpioinfo._link_path = path
 
         try:
-            self._extract_member(cpioinfo, os.path.join(path, cpioinfo.name))
+            self._extract_member(cpioinfo, os.path.join(path, six.ensure_text(cpioinfo.name)))
         except EnvironmentError as e:
             if self.errorlevel > 0:
                 raise
@@ -1594,7 +1594,7 @@ def makefile(self, cpioinfo, targetpath):
 
         if extractinfo:
             source = self.extractfile(extractinfo)
-            target = file(targetpath, "wb")
+            target = bltn_open(targetpath, "wb")
             copyfileobj(source, target)
             source.close()
             target.close()
@@ -1926,5 +1926,5 @@ def is_cpiofile(name):
     except CpioError:
         return False
 
-def cpioOpen(*al, **ad):
-    return CpioFile.open(*al, **ad)
+bltn_open = open
+open = CpioFile.open

xcp/dom0.py

@@ -96,7 +96,7 @@ def default_memory(host_mem_kib):
     return default_memory_for_version(host_mem_kib, platform_version)
 
 
-_size_and_unit_re = re.compile("^(-?\d+)([bkmg]?)$", re.IGNORECASE)
+_size_and_unit_re = re.compile(r"^(-?\d+)([bkmg]?)$", re.IGNORECASE)
 
 def _parse_size_and_unit(s):
     m = _size_and_unit_re.match(s)