Merge lp:~smoser/curtin/trunk.subp-more-tests into lp:~curtin-dev/curtin/trunk

[Scott Moser]

tests: more subp tests

Better test util.subp, exercising its target=/ and target=/some/root, and also verifying that exception is raised if subprocess exits non-zero.

[Server Team CI bot (server-team-bot)]

FAILED: Continuous integration, rev:421
https://jenkins.ubuntu.com/server/job/curtin-ci/256/
Executed test runs:
    FAILURE: https://jenkins.ubuntu.com/server/job/curtin-ci/nodes=amd64/256/console
    FAILURE: https://jenkins.ubuntu.com/server/job/curtin-ci/nodes=ppc64el/256/console

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/curtin-ci/256/rebuild

review: Needs Fixing (continuous-integration)

[Ryan Harper (raharper)]

Nice additional tests.

review: Approve

=== modified file 'tests/unittests/test_util.py'
--- tests/unittests/test_util.py	2016-08-08 18:40:17 +0000
+++ tests/unittests/test_util.py	2016-08-29 20:00:46 +0000
@@ -145,6 +145,8 @@
 
     stdin2err = ['bash', '-c', 'cat >&2']
     stdin2out = ['cat']
+    bin_true = ['bash', '-c', ':']
+    exit_with_value = ['bash', '-c', 'exit ${1:-0}', 'test_subp_exit_val']
     utf8_invalid = b'ab\xaadef'
     utf8_valid = b'start \xc3\xa9 end'
     utf8_valid_2 = b'd\xc3\xa9j\xc8\xa7'
@@ -161,6 +163,27 @@
         (out, _err) = util.subp(cmd, capture=True)
         self.assertEqual(out, self.utf8_valid_2.decode('utf-8'))
 
+
+    def test_subp_target_as_different_fors_of_slash_works(self):
+        # passing target=/ in any form should work.
+
+        # it is assumed that if chroot was used, then test case would
+        # fail unless user was root ('chroot /' is still priviledged)
+        util.subp(self.bin_true, target="/")
+        util.subp(self.bin_true, target="//")
+        util.subp(self.bin_true, target="///")
+        util.subp(self.bin_true, target="//etc/..//")
+
+    def test_subp_exit_nonzero_raises(self):
+        exc = None
+        try:
+            util.subp(self.exit_with_value + ["9"])
+        except util.ProcessExecutionError as e:
+            self.assertEqual(9, e.exit_code)
+            exc = e
+
+        self.assertNotEqual(exc, None)
+
     def test_subp_respects_decode_false(self):
         (out, err) = util.subp(self.stdin2out, capture=True, decode=False,
                                data=self.utf8_valid)
@@ -203,24 +226,68 @@
         self.assertEqual(err, None)
         self.assertEqual(out, None)
 
-    @mock.patch("curtin.util.subprocess.Popen")
-    def test_with_target_gets_chroot(self, m_popen):
-        noexist_pname = "mocked-away-does-not-matter"
-        sp = mock.Mock()
-        m_popen.return_value = sp
-        sp.communicate.return_value = (b'out-foo', b'err-foo')
-        sp.returncode = 0
-
-        ret = util.subp([noexist_pname], target="/mytarget",
-                        capture=True, decode=None)
+    def _subp_wrap_popen(self, cmd, kwargs,
+                         returncode=0, stdout=b'', stderr=b''):
+        # mocks the subprocess.Popen as expected from subp
+        # checks that subp returned the output of 'communicate' and
+        # returns the (args, kwargs) that Popen() was called with.
+
+        capture = kwargs.get('capture')
+
+        with mock.patch("curtin.util.subprocess.Popen") as m_popen:
+            sp = mock.Mock()
+            m_popen.return_value = sp
+            if capture:
+                sp.communicate.return_value = (stdout, stderr)
+            else:
+                sp.communicate.return_value = (None, None)
+            sp.returncode = returncode
+            ret = util.subp(cmd, **kwargs)
+
+        # popen should only ever be called once
         self.assertTrue(m_popen.called)
         self.assertEqual(1, m_popen.call_count)
+        # communicate() needs to have been called.
         self.assertTrue(sp.communicate.called)
 
-        # Popen should have been called with chroot
-        args, kwargs = m_popen.call_args
-        self.assertEqual(['chroot', '/mytarget', noexist_pname], args[0])
-        self.assertEqual((b'out-foo', b'err-foo'), ret)
+        if capture:
+            # capture response is decoded if decode is not False
+            decode = kwargs.get('decode', "replace")
+            if decode is False:
+                self.assertEqual(stdout.decode(stdout, stderr), ret)
+            else:
+                self.assertEqual((stdout.decode(errors=decode),
+                                  stderr.decode(errors=decode)), ret)
+        else:
+            # if capture is false, then return is None, None
+            self.assertEqual((None, None), ret)
+
+        popen_args, popen_kwargs = m_popen.call_args
+
+        # if target is not provided or is /, chroot should not be used
+        target = util.target_path(kwargs.get('target', None))
+        if target == "/":
+            self.assertEqual(cmd, popen_args[0])
+        else:
+            self.assertEqual(['chroot', target] + list(cmd), popen_args[0])
+        return m_popen.call_args
+        
+    def test_with_target_gets_chroot(self):
+        args, kwargs = self._subp_wrap_popen(["my-command"],
+                                             {'target': "/mytarget"})
+        self.assertIn('chroot', args[0])
+
+    def test_with_target_as_slash_does_not_chroot(self):
+        args, kwargs = self._subp_wrap_popen(
+            ['whatever'], {'capture': True, 'target': "/"})
+        self.assertNotIn('chroot', args[0])
+
+    def test_with_no_target_does_not_chroot(self):
+        args, kwargs = self._subp_wrap_popen(['whatever'], {'capture': True})
+        # note this path is reasonably tested with all of the above
+        # tests that do not mock Popen as if we did try to chroot the
+        # unit tests would fail unless they were run as root.
+        self.assertNotIn('chroot', args[0])
 
 
 class TestHuman2Bytes(TestCase):