Ensure proper root permissions in integration tests

integration-requirements.txt

@@ -1,5 +1,5 @@
 # PyPI requirements for cloud-init integration testing
 # https://cloudinit.readthedocs.io/en/latest/topics/integration_tests.html
 #
-pycloudlib @ git+https://github.com/canonical/pycloudlib.git@9211c0e5b34794595565d4626bc41ddbe14994f2
+pycloudlib @ git+https://github.com/canonical/pycloudlib.git@4b8d2cd5ac6316810ce16d081842da575625ca4f
 pytest

tests/integration_tests/instances.py

@@ -1,9 +1,11 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 import logging
 import os
+import uuid
 from tempfile import NamedTemporaryFile
 
 from pycloudlib.instance import BaseInstance
+from pycloudlib.result import Result
 
 from tests.integration_tests import integration_settings
 
@@ -18,6 +20,11 @@
 log = logging.getLogger('integration_testing')
 
 
+def _get_tmp_path():
+    tmp_filename = str(uuid.uuid4())
+    return '/var/tmp/{}.tmp'.format(tmp_filename)
+
+
 class IntegrationInstance:
     use_sudo = True
 
@@ -30,21 +37,39 @@ def __init__(self, cloud: 'IntegrationCloud', instance: BaseInstance,
     def destroy(self):
         self.instance.delete()
 
-    def execute(self, command):
-        return self.instance.execute(command)
+    def execute(self, command, *, use_sudo=None) -> Result:
+        if self.instance.username == 'root' and use_sudo is False:
+            raise Exception('Root user cannot run unprivileged')
+        if use_sudo is None:
+            use_sudo = self.use_sudo
+        return self.instance.execute(command, use_sudo=use_sudo)
 
-    def pull_file(self, remote_file, local_file):
-        self.instance.pull_file(remote_file, local_file)
+    def pull_file(self, remote_path, local_path):
+        # First copy to a temporary directory because of permissions issues
+        tmp_path = _get_tmp_path()
+        self.instance.execute('cp {} {}'.format(remote_path, tmp_path))
+        self.instance.pull_file(tmp_path, local_path)
 
     def push_file(self, local_path, remote_path):
-        self.instance.push_file(local_path, remote_path)
+        # First push to a temporary directory because of permissions issues
+        tmp_path = _get_tmp_path()
+        self.instance.push_file(local_path, tmp_path)
+        self.execute('mv {} {}'.format(tmp_path, remote_path))
 
     def read_from_file(self, remote_path) -> str:
-        tmp_file = NamedTemporaryFile('r')
-        self.pull_file(remote_path, tmp_file.name)
-        with tmp_file as f:
-            contents = f.read()
-        return contents
+        result = self.execute('cat {}'.format(remote_path))
+        if result.failed:
+            # TODO: Raise here whatever pycloudlib raises when it has
+            # a consistent error response
+            raise IOError(
+                'Failed reading remote file via cat: {}\n'
+                'Return code: {}\n'
+                'Stderr: {}\n'
+                'Stdout: {}'.format(
+                    remote_path, result.return_code,
+                    result.stderr, result.stdout)
+            )
+        return result.stdout
 
     def write_to_file(self, remote_path, contents: str):
         # Writes file locally and then pushes it rather

tests/integration_tests/modules/test_users_groups.py

@@ -70,7 +70,10 @@ class TestUsersGroups:
     def test_users_groups(self, regex, getent_args, class_client):
         """Use getent to interrogate the various expected outcomes"""
         result = class_client.execute(["getent"] + getent_args)
-        assert re.search(regex, result.stdout) is not None
+        assert re.search(regex, result.stdout) is not None, (
+            "'getent {}' resulted in '{}', "
+            "but expected to match regex {}".format(
+                ' '.join(getent_args), result.stdout, regex))
 
     def test_user_root_in_secret(self, class_client):
         """Test root user is in 'secret' group."""