Merge branch 'master' into ubuntu/xenial

1 files changed, 93 insertions, 2 deletions

diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py
index 37a984ac..d2031f59 100644
--- a/tests/unittests/test_util.py
+++ b/tests/unittests/test_util.py
@@ -371,8 +371,30 @@ class TestReadDMIData(helpers.FilesystemMockingTestCase):
         self._create_sysfs_parent_directory()
         expected_dmi_value = 'dmidecode-used'
         self._configure_dmidecode_return('use-dmidecode', expected_dmi_value)
-        self.assertEqual(expected_dmi_value,
-                         util.read_dmi_data('use-dmidecode'))
+        with mock.patch("cloudinit.util.os.uname") as m_uname:
+            m_uname.return_value = ('x-sysname', 'x-nodename',
+                                    'x-release', 'x-version', 'x86_64')
+            self.assertEqual(expected_dmi_value,
+                             util.read_dmi_data('use-dmidecode'))
+
+    def test_dmidecode_not_used_on_arm(self):
+        self.patch_mapping({})
+        self._create_sysfs_parent_directory()
+        dmi_val = 'from-dmidecode'
+        dmi_name = 'use-dmidecode'
+        self._configure_dmidecode_return(dmi_name, dmi_val)
+
+        expected = {'armel': None, 'aarch64': None, 'x86_64': dmi_val}
+        found = {}
+        # we do not run the 'dmi-decode' binary on some arches
+        # verify that anything requested that is not in the sysfs dir
+        # will return None on those arches.
+        with mock.patch("cloudinit.util.os.uname") as m_uname:
+            for arch in expected:
+                m_uname.return_value = ('x-sysname', 'x-nodename',
+                                        'x-release', 'x-version', arch)
+                found[arch] = util.read_dmi_data(dmi_name)
+        self.assertEqual(expected, found)
 
     def test_none_returned_if_neither_source_has_data(self):
         self.patch_mapping({})
@@ -486,4 +508,73 @@ class TestReadSeeded(helpers.TestCase):
         self.assertEqual(found_md, {'key1': 'val1'})
         self.assertEqual(found_ud, ud)
 
+
+class TestSubp(helpers.TestCase):
+
+    stdin2err = ['bash', '-c', 'cat >&2']
+    stdin2out = ['cat']
+    utf8_invalid = b'ab\xaadef'
+    utf8_valid = b'start \xc3\xa9 end'
+    utf8_valid_2 = b'd\xc3\xa9j\xc8\xa7'
+
+    def printf_cmd(self, *args):
+        # bash's printf supports \xaa.  So does /usr/bin/printf
+        # but by using bash, we remove dependency on another program.
+        return(['bash', '-c', 'printf "$@"', 'printf'] + list(args))
+
+    def test_subp_handles_utf8(self):
+        # The given bytes contain utf-8 accented characters as seen in e.g.
+        # the "deja dup" package in Ubuntu.
+        cmd = self.printf_cmd(self.utf8_valid_2)
+        (out, _err) = util.subp(cmd, capture=True)
+        self.assertEqual(out, self.utf8_valid_2.decode('utf-8'))
+
+    def test_subp_respects_decode_false(self):
+        (out, err) = util.subp(self.stdin2out, capture=True, decode=False,
+                               data=self.utf8_valid)
+        self.assertTrue(isinstance(out, bytes))
+        self.assertTrue(isinstance(err, bytes))
+        self.assertEqual(out, self.utf8_valid)
+
+    def test_subp_decode_ignore(self):
+        # this executes a string that writes invalid utf-8 to stdout
+        (out, _err) = util.subp(self.printf_cmd('abc\\xaadef'),
+                                capture=True, decode='ignore')
+        self.assertEqual(out, 'abcdef')
+
+    def test_subp_decode_strict_valid_utf8(self):
+        (out, _err) = util.subp(self.stdin2out, capture=True,
+                                decode='strict', data=self.utf8_valid)
+        self.assertEqual(out, self.utf8_valid.decode('utf-8'))
+
+    def test_subp_decode_invalid_utf8_replaces(self):
+        (out, _err) = util.subp(self.stdin2out, capture=True,
+                                data=self.utf8_invalid)
+        expected = self.utf8_invalid.decode('utf-8', errors='replace')
+        self.assertEqual(out, expected)
+
+    def test_subp_decode_strict_raises(self):
+        args = []
+        kwargs = {'args': self.stdin2out, 'capture': True,
+                  'decode': 'strict', 'data': self.utf8_invalid}
+        self.assertRaises(UnicodeDecodeError, util.subp, *args, **kwargs)
+
+    def test_subp_capture_stderr(self):
+        data = b'hello world'
+        (out, err) = util.subp(self.stdin2err, capture=True,
+                               decode=False, data=data)
+        self.assertEqual(err, data)
+        self.assertEqual(out, b'')
+
+    def test_returns_none_if_no_capture(self):
+        (out, err) = util.subp(self.stdin2out, data=b'', capture=False)
+        self.assertEqual(err, None)
+        self.assertEqual(out, None)
+
+    def test_bunch_of_slashes_in_path(self):
+        self.assertEqual("/target/my/path/",
+                         util.target_path("/target/", "//my/path/"))
+        self.assertEqual("/target/my/path/",
+                         util.target_path("/target/", "///my/path/"))
+
 # vi: ts=4 expandtab