Merge pull request #30 from xenserver-next/remove-always-broken-_XZProxy

Fix pytype and pylint errors from xcp.cpiofile.CpioFile._XZProxy()

Author: bernhardkaindl

tests/test_cpio.py

@@ -1,12 +1,14 @@
 from __future__ import print_function
 from hashlib import md5
+import lzma
 import os
 import sys
 import shutil
 import subprocess
 import unittest
 import warnings
 
+import xcp.cpiofile
 from xcp.cpiofile import CpioFile, CpioFileCompat, CPIO_PLAIN, CPIO_GZIPPED
 
 def writeRandomFile(fn, size, start=b'', add=b'a'):
@@ -113,9 +115,9 @@ def doArchive(self, fn, fmt=None):
         self.archiveExtract(fn)
         fn2 = "archive2" + fn[len("archive"):]
         print("creating %s" % fn2)
-        self.archiveCreate(fn2, fmt is None and 'w' or 'w|%s' % fmt)
+        self.archiveCreate(fn2, fmt is None and "w" or "w:%s" % fmt)
         if fmt is not None:
-            self.archiveExtract(fn2, 'r|%s' % fmt)
+            self.archiveExtract(fn2, "r:%s" % fmt)
 
     def test_plain(self):
         self.doArchive('archive.cpio')
@@ -132,6 +134,13 @@ def test_xz(self):
         print('Running test for XZ')
         self.doArchive('archive.cpio.xz', 'xz')
 
+    def test_cover_xzopen_pass_fileobj(self):
+        """Cover CpioFile.xzopen() not supporting receiving a fileobj argument"""
+        class MockCpioFile(CpioFile):
+            def __init__(self):  # pylint: disable=super-init-not-called
+                pass
+        with self.assertRaises(xcp.cpiofile.CompressionError):
+            MockCpioFile.xzopen(name="", mode="r", fileobj=lzma.LZMAFile("/dev/null"))
     # CpioFileCompat testing
 
     def archiveExtractCompat(self, fn, comp):

tox.ini

@@ -78,7 +78,8 @@ setenv =
     {[cov]setenv}
 commands =
     {cov,covcp,covcombine,check,fox,lint,test,pytype,mdreport}: {[test]commands}
-    {cov,covcp,covcombine}: {[cov]commands}
+    # covcombine shall not call [cov]commands: diff-cover shall check the combined cov:
+    {cov,covcp}: {[cov]commands}
     {py27-test}: pylint --py3k --disable=no-absolute-import xcp/
     covcp: cp -av {envlogdir}/coverage.xml {env:UPLOAD_DIR:.}
     covcombine: {[covcombine]commands}

xcp/cpiofile.py

@@ -572,27 +572,6 @@ def init(self):
 
 # class _BZ2Proxy
 
-class _XZProxy(_CMPProxy):
-    """Small proxy class that enables external file object
-       support for "r:xz" and "w:xz" modes.
-    """
-
-    def __init__(self, fileobj, mode):
-        _CMPProxy.__init__(self, fileobj, mode)
-        self.init()
-
-    def init(self):
-        import lzma
-        self.pos = 0
-        if self.mode == "r":
-            self.cmpobj = lzma.BZ2Decompressor()
-            self.fileobj.seek(0)
-            self.buf = b""
-        else:
-            self.cmpobj = lzma.BZ2Compressor()
-
-# class _XZProxy
-
 
 #------------------------
 # Extraction file object
@@ -1162,11 +1141,13 @@ def xzopen(cls, name, mode="r", fileobj=None, compresslevel=6):
             raise CompressionError("lzma module is not available")
 
         if fileobj is not None:
-            fileobj = _XZProxy(fileobj, mode)
-        else:
-            # FIXME: not compatible with python3 API
-            fileobj = lzma.LZMAFile(name, mode, options={'level': compresslevel, 'dict_size': 20 })
-
+            raise CompressionError("passing fileobj not implemented for LZMA")
+        kwargs = {}
+        if sys.version_info < (3, 0):
+            kwargs["options"] = {"level": compresslevel}
+        elif "w" in mode:
+            kwargs["preset"] = compresslevel
+        fileobj = lzma.LZMAFile(name, mode, **kwargs)
         try:
             t = cls.cpioopen(name, mode, fileobj)
         except IOError:
@@ -1179,7 +1160,7 @@ def xzopen(cls, name, mode="r", fileobj=None, compresslevel=6):
         "cpio": "cpioopen",   # uncompressed cpio
         "gz":  "gzopen",    # gzip compressed cpio
         "bz2": "bz2open",   # bzip2 compressed cpio
-        "xz":  "xzopen "    # xz compressed cpio
+        "xz":  "xzopen",  # xz compressed cpio
     }
 
     #--------------------------------------------------------------------------