UTF-8 encoding fixes + tests for xcp.pci, .cmd and .net.biosdevname

.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/data/pci.ids

@@ -11,6 +11,8 @@
 	1314  Wrestler HDMI Audio
 		174b 1001  PURE Fusion Mini
 	1636  Renoir
+	67b0  Hawaii XT / Grenada XT [Radeon R9 290X/390X]
+		1787 2020  R9 290X IceQ X² Turbo
 	7340  Navi 14 [Radeon RX 5500/5500M / Pro 5500M]
 
 # List of known device classes, subclasses and programming interfaces

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_biosdevname.py

@@ -1,6 +1,8 @@
+import subprocess
 import unittest
 from mock import patch, Mock
 
+from xcp import xcp_popen_text_kwargs
 from xcp.net.biosdevname import has_ppn_quirks, all_devices_all_names
 
 class TestQuirks(unittest.TestCase):
@@ -40,6 +42,11 @@ def test(self):
 
         # check after the fact that we mocked the proper calls
         self.assertEqual(popen_mock.call_count, 2)
+        popen_mock.assert_called_with(['/sbin/biosdevname', '--policy', 'all_ethN', '-d'],
+                                      stdout=subprocess.PIPE,
+                                      stderr=subprocess.PIPE,
+                                      **xcp_popen_text_kwargs)
+
         calls = popen_mock.call_args_list
         self.assertEqual(calls[0].args[0], ['/sbin/biosdevname', '--policy', 'physical', '-d'])
         self.assertEqual(calls[1].args[0], ['/sbin/biosdevname', '--policy', 'all_ethN', '-d'])

tests/test_cmd.py

@@ -1,8 +1,14 @@
+import subprocess
 import unittest
 from mock import patch, Mock, DEFAULT
 
+from xcp import xcp_popen_text_kwargs
 from xcp.cmd import OutputCache
 
+from .xcptestlib import set_c_locale
+
+set_c_locale()
+
 class TestCache(unittest.TestCase):
     def setUp(self):
         self.c = OutputCache()
@@ -30,12 +36,14 @@ def test_fileContents(self):
             self.assertEqual(data, "line1\nline2\n")
 
     def test_runCmd(self):
-        output_data = "line1\nline2\n"
+        output_data = "output with\nUTF-8:\u25b6\U0001f601\n"
+        stderr_data = "error with\nUTF-8:\u2614\U0001f629\n"
+
         with patch("xcp.cmd.subprocess.Popen") as popen_mock:
             # mock Popen .communicate and .returncode for
-            # `output_data`on stdout, nothing on stderr, and exit
+            # `output_data` on stdout, `stderr_data` on stderr, and an exit
             # value of 42
-            communicate_mock = Mock(return_value=(output_data, ""))
+            communicate_mock = Mock(return_value=(output_data, stderr_data))
             popen_mock.return_value.communicate = communicate_mock
             def communicate_side_effect(_input_text):
                 popen_mock.return_value.returncode = 42
@@ -44,11 +52,23 @@ def communicate_side_effect(_input_text):
 
             # uncached runCmd
             data = self.c.runCmd(['ls', '/tmp'], True)
-            popen_mock.assert_called_once()
+            popen_mock.assert_called_once_with(["ls", "/tmp"],
+                                               bufsize=1,
+                                               stdin=None,
+                                               stdout=subprocess.PIPE,
+                                               stderr=subprocess.PIPE,
+                                               shell=False,
+                                               **xcp_popen_text_kwargs)
             self.assertEqual(data, (42, output_data))
 
             # rerun as cached
             popen_mock.reset_mock()
-            data = self.c.runCmd(['ls', '/tmp'], True)
+            data = self.c.runCmd(['ls', '/tmp'], with_stdout=True, with_stderr=True)
             popen_mock.assert_not_called()
-            self.assertEqual(data, (42, output_data))
+            self.assertEqual(data, (42, output_data, stderr_data))
+
+    def test_runCmdCatStdin(self):
+        """Call cat with a given UTF-8 input text and expect it to be returned"""
+        stdin = "output with\nUTF-8:\u25b6\U0001f601"
+        ret = self.c.runCmd("cat", inputtext=stdin, with_stdout=True, with_stderr=True)
+        self.assertEqual(ret, (0, stdin, ""))

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_ip.py

@@ -0,0 +1,45 @@
+# Unit test for xcp/net/ip.py
+import unittest
+from subprocess import PIPE
+
+from mock import Mock, patch
+
+import xcp.net.ip
+from xcp import xcp_popen_text_kwargs
+
+
+class TestIp(unittest.TestCase):
+    def test_ip_link_set_name(self):
+        with patch("xcp.net.ip.Popen") as popen_mock:
+
+            # Setup the return values and return codes returned by popen_mock:
+            ip_link_show_lo_stdout = (
+                "1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state"
+                "UNKNOWN mode DEFAULT group default qlen 1000"
+            )
+            communicate_mock = Mock(
+                side_effect=iter([(ip_link_show_lo_stdout, ""), ("", "")])
+            )
+            popen_mock.return_value.communicate = communicate_mock
+            popen_mock.return_value.returncode = 0
+
+            # Call the testee function with xcp.net.ip.Popen() mocked using popen_mock:
+            xcp.net.ip.ip_link_set_name("lo", "lo0")
+
+        # check the number or calls to Popen() by done by ip_link_set_name()
+        self.assertEqual(popen_mock.call_count, 4)
+
+        # check the captured regular arguments passed to Popen() by ip_link_set_name()
+        calls = popen_mock.call_args_list
+        self.assertEqual(calls[0].args, (["ip", "link", "show", "lo"],))
+        self.assertEqual(calls[1].args, (["ip", "link", "set", "lo", "down"],))
+        self.assertEqual(calls[2].args, (["ip", "link", "set", "lo", "name", "lo0"],))
+        self.assertEqual(calls[3].args, (["ip", "link", "set", "lo0", "up"],))
+
+        # check the captured keyword arguments passed to Popen() by ip_link_set_name()
+        xcp_popen_text_kwargs_stdout = xcp_popen_text_kwargs.copy()
+        xcp_popen_text_kwargs_stdout["stdout"] = PIPE
+        self.assertEqual(calls[0].kwargs, xcp_popen_text_kwargs_stdout)
+        self.assertEqual(calls[1].kwargs, xcp_popen_text_kwargs)
+        self.assertEqual(calls[2].kwargs, xcp_popen_text_kwargs)
+        self.assertEqual(calls[3].kwargs, xcp_popen_text_kwargs)