xcp.accessor, xcp.repository: Use binary mode for file I/O

.github/workflows/main.yml

@@ -3,48 +3,54 @@ name: Unit tests
 on: [push, pull_request]
 
 jobs:
-  test_py2:
-    runs-on: ubuntu-20.04
+  test:
+    strategy:
+      matrix:
+        include:
+        - pyversion: '2.7'
+          os: ubuntu-20.04
+        - pyversion: '3'
+          os: ubuntu-latest
+    runs-on: ${{ matrix.os }}
 
     steps:
       - uses: actions/checkout@v2
         with:
           fetch-depth: 0
-      - name: Set up Python 2.7
+      - name: Set up Python ${{ matrix.pyversion }}
         uses: actions/setup-python@v2
         with:
-          python-version: '2.7'
+          python-version: ${{ matrix.pyversion }}
 
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
           pip install -r requirements-dev.txt
           # FIXME: branding.py still has no permanent home
           curl https://gist.github.com/ydirson/3c36a7e19d762cc529a6c82340894ccc/raw/5ca39f621b1feab813e171f535c1aad1bd483f1d/branding.py -O -L
-          pip install pyliblzma
           pip install -e .
           command -v xz
 
       - name: Test
         run: |
           pytest --cov -rP
           coverage xml
-          coverage html
-          coverage html -d htmlcov-tests --include="tests/*"
-          diff-cover --html-report coverage-diff.html coverage.xml
+          coverage html -d htmlcov-${{ matrix.pyversion }}
+          coverage html -d htmlcov-tests-${{ matrix.pyversion }} --include="tests/*"
+          diff-cover --compare-branch=origin/master --html-report coverage-diff-${{ matrix.pyversion }}.html coverage.xml
 
       - name: Pylint
         run: |
           pylint --version
           pylint --exit-zero xcp/ tests/ setup.py
           pylint --exit-zero --msg-template="{path}:{line}: [{msg_id}({symbol}), {obj}] {msg}" xcp/ tests/ setup.py > pylint.txt
-          diff-quality --violations=pylint --html-report pylint-diff.html pylint.txt
+          diff-quality --compare-branch=origin/master --violations=pylint --html-report pylint-diff-${{ matrix.pyversion }}.html pylint.txt
 
       - uses: actions/upload-artifact@v3
         with:
           name: Coverage and pylint reports
           path: |
-            coverage-diff.html
-            pylint-diff.html
-            htmlcov
-            htmlcov-tests
+            coverage-diff-${{ matrix.pyversion }}.html
+            pylint-diff-${{ matrix.pyversion }}.html
+            htmlcov-${{ matrix.pyversion }}
+            htmlcov-tests-${{ matrix.pyversion }}

requirements-dev.txt

@@ -5,6 +5,11 @@ 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"
+pyliblzma ; python_version < "3.0"

tests/branding.py

@@ -0,0 +1,36 @@
+BRAND_CONSOLE_URL = 'https://xcp-ng.org'
+BRAND_CONSOLE = 'XCP-ng Center'
+BRAND_GUEST_SHORT = 'VM'
+BRAND_GUESTS_SHORT = 'VMs'
+BRAND_GUESTS = 'Virtual Machines'
+BRAND_GUEST = 'Virtual Machine'
+BRAND_SERVERS = 'XCP-ng Hosts'
+BRAND_SERVER = 'XCP-ng Host'
+BRAND_VDI = ''
+COMPANY_DOMAIN = 'xcp-ng.org'
+COMPANY_NAME_LEGAL = 'Open Source'
+COMPANY_NAME = 'Open Source'
+COMPANY_NAME_SHORT = 'Open Source'
+COMPANY = 'Open Source'
+COMPANY_PRODUCT_BRAND = 'XCP-ng'
+COMPANY_WEBSITE = 'https://xcp-ng.org'
+COPYRIGHT_YEARS = '2018-2022'
+ISO_PV_TOOLS_COPYRIGHT = 'XCP-ng'
+ISO_PV_TOOLS_LABEL = 'XCP-ng VM Tools'
+ISO_PV_TOOLS_PUBLISHER = 'XCP-ng'
+PLATFORM_MAJOR_VERSION = '3'
+PLATFORM_MICRO_VERSION = '1'
+PLATFORM_MINOR_VERSION = '2'
+PLATFORM_NAME = 'XCP'
+PLATFORM_ORGANISATION = 'xen.org'
+PLATFORM_VERSION = '3.2.1'
+PLATFORM_WEBSITE = 'www.xen.org'
+PRODUCT_BRAND = 'XCP-ng'
+PRODUCT_BRAND_DASHED = 'XCP-ng'
+PRODUCT_MAJOR_VERSION = '8'
+PRODUCT_MICRO_VERSION = '1'
+PRODUCT_MINOR_VERSION = '2'
+PRODUCT_NAME = 'xenenterprise'
+PRODUCT_VERSION = '8.2.1'
+PRODUCT_VERSION_TEXT = '8.2'
+PRODUCT_VERSION_TEXT_SHORT = '8.2'

tests/test_accessor.py

@@ -1,21 +1,42 @@
 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)
+    def test_access(self):
+        a = xcp.accessor.createAccessor(self.url, True)
         a.start()
         self.assertTrue(a.access('.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)
+    def test_file_binfile(self):
+        BINFILE = "boot/isolinux/mboot.c32"
+        a = xcp.accessor.createAccessor(self.url, True)
         a.start()
-        self.assertTrue(a.access('.treeinfo'))
-        self.assertFalse(a.access('no_such_file'))
-        self.assertEqual(a.lastError, 404)
+        self.assertTrue(a.access(BINFILE))
+        inf = a.openAddress(BINFILE)
+        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

@@ -1,22 +1,15 @@
 import unittest
+from parameterized import parameterized_class
 
 import xcp.accessor
 from xcp import repository
 from xcp.version import Version
 
+@parameterized_class([{"url": "file://tests/data/repo/"},
+                      {"url": "https://updates.xcp-ng.org/netinstall/8.2.1"}])
 class TestRepository(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)
-        repo_ver = repository.BaseRepository.getRepoVer(a)
-        self.assertEqual(repo_ver, Version([3, 2, 1]))
-        product_ver = repository.BaseRepository.getProductVersion(a)
-        self.assertEqual(product_ver, Version([8, 2, 1]))
-        repos = repository.BaseRepository.findRepositories(a)
-        self.assertEqual(len(repos), 1)
-
-    def test_file(self):
-        a = xcp.accessor.createAccessor("file://tests/data/repo/", True)
+    def test_basicinfo(self):
+        a = xcp.accessor.createAccessor(self.url, True)
         repo_ver = repository.BaseRepository.getRepoVer(a)
         self.assertEqual(repo_ver, Version([3, 2, 1]))
         product_ver = repository.BaseRepository.getProductVersion(a)

tests/test_xmlunwrap.py

@@ -18,18 +18,18 @@ def test(self):
 
         self.assertEqual([getText(el)
                           for el in getElementsByTagName(self.top_el, ["fred"])],
-                         ["text1", "text2"])
+                         [b"text1", b"text2"])
 
-        x = getMapAttribute(self.top_el, ["mode"], [('test', 42), ('stuff', 77)])
+        x = getMapAttribute(self.top_el, ["mode"], [(b'test', 42), (b'stuff', 77)])
         self.assertEqual(x, 42)
-        x = getMapAttribute(self.top_el, ["made"], [('test', 42), ('stuff', 77)],
-                            default='stuff')
+        x = getMapAttribute(self.top_el, ["made"], [(b'test', 42), (b'stuff', 77)],
+                            default=b'stuff')
         self.assertEqual(x, 77)
 
         x = getStrAttribute(self.top_el, ["mode"])
-        self.assertEqual(x, "test")
+        self.assertEqual(x, b"test")
         x = getStrAttribute(self.top_el, ["made"])
-        self.assertEqual(x, "")
+        self.assertEqual(x, b"")
         x = getStrAttribute(self.top_el, ["made"], None)
         self.assertEqual(x, None)
 

xcp/accessor.py

@@ -99,9 +99,10 @@ def __init__(self, location, ro):
         super(FilesystemAccessor, self).__init__(ro)
         self.location = location
 
-    def openAddress(self, address):
+    def openAddress(self, address, **kwargs):
         try:
-            filehandle = open(os.path.join(self.location, address), 'r')
+            kwargs["mode"] = "r" if "encoding" in kwargs else "rb"
+            filehandle = open(os.path.join(self.location, address), **kwargs)
         except OSError as e:
             if e.errno == errno.EIO:
                 self.lastError = 5
@@ -165,9 +166,10 @@ def finish(self):
             os.rmdir(self.location)
             self.location = None
 
-    def writeFile(self, in_fh, out_name):
+    def writeFile(self, in_fh, out_name, **kwargs):
         logger.info("Copying to %s" % os.path.join(self.location, out_name))
-        out_fh = open(os.path.join(self.location, out_name), 'w')
+        kwargs["mode"] = "w" if "encoding" in kwargs else "wb"
+        out_fh = open(os.path.join(self.location, out_name), **kwargs)
         return self._writeFile(in_fh, out_fh)
 
     def __del__(self):
@@ -220,9 +222,10 @@ def __init__(self, baseAddress, ro):
         super(FileAccessor, self).__init__(ro)
         self.baseAddress = baseAddress
 
-    def openAddress(self, address):
+    def openAddress(self, address, **kwargs):
         try:
-            file = open(os.path.join(self.baseAddress, address))
+            kwargs["mode"] = "r" if "encoding" in kwargs else "rb"
+            file = open(os.path.join(self.baseAddress, address), **kwargs)
         except IOError as e:
             if e.errno == errno.EIO:
                 self.lastError = 5
@@ -240,9 +243,10 @@ def openAddress(self, address):
             return False
         return file
 
-    def writeFile(self, in_fh, out_name):
+    def writeFile(self, in_fh, out_name, **kwargs):
         logger.info("Copying to %s" % os.path.join(self.baseAddress, out_name))
-        out_fh = open(os.path.join(self.baseAddress, out_name), 'w')
+        kwargs["mode"] = "w" if "encoding" in kwargs else "wb"
+        out_fh = open(os.path.join(self.baseAddress, out_name), **kwargs)
         return self._writeFile(in_fh, out_fh)
 
     def __repr__(self):
@@ -331,13 +335,14 @@ def access(self, path):
             self.lastError = 500
             return False
 
-    def openAddress(self, address):
+    def openAddress(self, address, **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')
+        kwargs["mode"] = "r" if "encoding" in kwargs else "rb"
+        s = self.ftp.transfercmd('RETR ' + url).makefile(**kwargs)
         self.cleanup = True
         return s