Testcpio improvements, cpiofile improvements and port to py3

requirements-dev.txt

@@ -5,3 +5,6 @@ diff_cover
 mock
 pytest
 pytest-cov
+# dependencies also in setup.py until they can be used
+six
+future

setup.py

@@ -35,5 +35,7 @@
                 ],
       requires=[
           'branding',
+          'six',
+          'future',
       ],
      )

tests/test_cpio.py

@@ -1,29 +1,27 @@
+from __future__ import print_function
+from hashlib import md5
 import os
+import sys
 import shutil
 import subprocess
 import unittest
 import warnings
 
-from xcp.cpiofile import CpioFile, CpioInfo, CpioFileCompat, CPIO_PLAIN, CPIO_GZIPPED
+from xcp.cpiofile import CpioFile, CpioFileCompat, CPIO_PLAIN, CPIO_GZIPPED
 
-try:
-    from hashlib import md5
-except:
-    from md5 import md5
-
-def writeRandomFile(fn, size, start='', add='a'):
-    f = open(fn, 'wb')
-    m = md5()
-    m.update(start)
-    assert(len(add) != 0)
-    while size > 0:
-        d = m.digest()
-        if size < len(d):
-            d=d[:size]
-        f.write(d)
-        size -= len(d)
-        m.update(add)
-    f.close()
+def writeRandomFile(fn, size, start=b'', add=b'a'):
+    "Create a pseudo-random reproducible file from seeds `start` amd `add`"
+    with open(fn, 'wb') as f:
+        m = md5()
+        m.update(start)
+        assert(len(add) != 0)
+        while size > 0:
+            d = m.digest()
+            if size < len(d):
+                d=d[:size]
+            f.write(d)
+            size -= len(d)
+            m.update(add)
 
 
 def check_call(cmd):
@@ -40,20 +38,27 @@ def setUp(self):
         shutil.rmtree('archive', True)
         os.mkdir('archive')
         writeRandomFile('archive/data', 10491)
-        self.md5data = md5(open('archive/data').read()).hexdigest()
-        check_call("find archive | cpio -o -H newc > archive.cpio")
+        with open('archive/data', 'rb') as fd:
+            self.md5data = md5(fd.read()).hexdigest()
+        # fixed timestamps for cpio reproducibility
+        os.utime('archive/data', (0, 0))
+        os.utime('archive', (0, 0))
+
+        check_call(
+            "find archive | cpio --reproducible -o -H newc > archive.cpio")
         check_call("gzip -c < archive.cpio > archive.cpio.gz")
         check_call("bzip2 -c < archive.cpio > archive.cpio.bz2")
         try:
-            import lzma
-            self.doXZ = subprocess.call("xz --check=crc32 --lzma2=dict=1MiB < archive.cpio > archive.cpio.xz", shell=True) == 0
+            import lzma         # pylint: disable=unused-variable
+            self.doXZ = subprocess.call("xz --check=crc32 --lzma2=dict=1MiB"
+                                        " < archive.cpio > archive.cpio.xz", shell=True) == 0
         except Exception as ex:
             # FIXME will issue warning even if test_xz is not requested
             warnings.warn("will not test cpio.xz: %s" % ex)
             self.doXZ = False
 
     def tearDown(self):
-        check_call("rm -rf archive archive.cpio* archive2")
+        check_call("rm -rf archive archive.cpio* archive2 archive2.cpio*")
 
     # TODO check with file (like 'r:*')
     # TODO use cat to check properly for pipes
@@ -77,10 +82,12 @@ def archiveExtract(self, fn, fmt='r|*'):
         arc.close()
 
     def archiveCreate(self, fn, fmt='w'):
-        os.unlink(fn)
+        if os.path.exists(fn):
+            os.unlink(fn)
         arc = CpioFile.open(fn, fmt)
         f = arc.getcpioinfo('archive/data')
-        arc.addfile(f, open('archive/data'))
+        with open('archive/data', 'rb') as fd:
+            arc.addfile(f, fd)
         # test recursively add "."
         os.chdir('archive')
         arc.add(".")
@@ -89,17 +96,24 @@ def archiveCreate(self, fn, fmt='w'):
         arc.close()
         # special case for XZ, test check type (crc32)
         if fmt.endswith('xz'):
-            f = open(fn, 'rb')
-            f.seek(6)
-            self.assertEqual(f.read(2), '\x00\x01')
-            f.close()
+            with open(fn, 'rb') as f:
+                # check xz magic
+                self.assertEqual(f.read(6), b"\xfd7zXZ\0")
+                # check stream flags
+                if sys.version_info < (3, 0):
+                    expected_flags = b'\x00\x01' # pylzma defaults to CRC32
+                else:
+                    expected_flags = b'\x00\x04' # python3 defaults to CRC64
+                self.assertEqual(f.read(2), expected_flags)
         self.archiveExtract(fn)
 
     def doArchive(self, fn, fmt=None):
         self.archiveExtract(fn)
-        self.archiveCreate(fn, fmt is None and 'w' or 'w|%s' % fmt )
-        if not fmt is None:
-            self.archiveExtract(fn, 'r|%s' % fmt)
+        fn2 = "archive2" + fn[len("archive"):]
+        print("creating %s" % fn2)
+        self.archiveCreate(fn2, fmt is None and 'w' or 'w|%s' % fmt)
+        if fmt is not None:
+            self.archiveExtract(fn2, 'r|%s' % fmt)
 
     def test_plain(self):
         self.doArchive('archive.cpio')
@@ -113,7 +127,7 @@ def test_bz2(self):
     def test_xz(self):
         if not self.doXZ:
             raise unittest.SkipTest("lzma package or xz tool not available")
-        print 'Running test for XZ'
+        print('Running test for XZ')
         self.doArchive('archive.cpio.xz', 'xz')
 
     # CpioFileCompat testing

tests/test_ifrename_dynamic.py

@@ -1,12 +1,10 @@
+from __future__ import unicode_literals
 import json
 import logging
 import unittest
 from copy import deepcopy
 
-try:
-    import cStringIO as StringIO
-except ImportError:
-    import StringIO
+from io import StringIO
 
 from xcp.net.ifrename.dynamic import DynamicRules
 from xcp.net.ifrename.macpci import MACPCI
@@ -16,7 +14,7 @@
 class TestLoadAndParse(unittest.TestCase):
 
     def setUp(self):
-        self.logbuf = StringIO.StringIO()
+        self.logbuf = StringIO()
         openLog(self.logbuf, logging.NOTSET)
 
     def tearDown(self):
@@ -26,7 +24,7 @@ def tearDown(self):
 
     def test_null(self):
 
-        fd = StringIO.StringIO("")
+        fd = StringIO("")
         dr = DynamicRules(fd=fd)
 
         self.assertTrue(dr.load_and_parse())
@@ -36,7 +34,7 @@ def test_null(self):
 
     def test_empty(self):
 
-        fd = StringIO.StringIO(
+        fd = StringIO(
             '{"lastboot":[],"old":[]}'
             )
         dr = DynamicRules(fd=fd)
@@ -48,7 +46,7 @@ def test_empty(self):
 
     def test_one_invalid(self):
 
-        fd = StringIO.StringIO(
+        fd = StringIO(
             '{"lastboot":[["","",""]],"old":[]}'
             )
         dr = DynamicRules(fd=fd)
@@ -60,7 +58,7 @@ def test_one_invalid(self):
 
     def test_one_valid_lastboot(self):
 
-        fd = StringIO.StringIO(
+        fd = StringIO(
             '{"lastboot":[["01:23:45:67:89:0a","00:10.2","eth2"]],"old":[]}'
             )
         dr = DynamicRules(fd=fd)
@@ -74,7 +72,7 @@ def test_one_valid_lastboot(self):
 
     def test_one_valid_lastboot2(self):
 
-        fd = StringIO.StringIO(
+        fd = StringIO(
             '{"lastboot":[],"old":[["01:23:45:67:89:0a","00:10.2","eth2"]]}'
             )
         dr = DynamicRules(fd=fd)
@@ -88,7 +86,7 @@ def test_one_valid_lastboot2(self):
 class TestGenerate(unittest.TestCase):
 
     def setUp(self):
-        self.logbuf = StringIO.StringIO()
+        self.logbuf = StringIO()
         openLog(self.logbuf, logging.NOTSET)
 
     def tearDown(self):
@@ -149,7 +147,7 @@ def test_pci_missing(self):
 class TestSave(unittest.TestCase):
 
     def setUp(self):
-        self.logbuf = StringIO.StringIO()
+        self.logbuf = StringIO()
         openLog(self.logbuf, logging.NOTSET)
 
     def tearDown(self):
@@ -162,10 +160,7 @@ def test_one_invalid_lastboot(self):
         dr = DynamicRules()
         dr.lastboot = [["foo", "bar", "baz"]]
 
-        try:
-            json.loads(dr.write(False))
-        except Exception:
-            self.fail()
+        json.loads(dr.write(False))
 
     def test_one_ibft_lastboot(self):