Use tests from z-o-t master

Author: lmlg

src/tests/target.py

@@ -27,30 +27,92 @@
 import zaza.openstack.utilities.openstack as openstack_utils
 
 
-class CephFSTests(test_utils.OpenStackBaseTest):
+class CephFSTests(unittest.TestCase):
     """Encapsulate CephFS tests."""
 
-    RESOURCE_PREFIX = 'zaza-cephfstests'
-    INSTANCE_USERDATA = """#cloud-config
-packages:
-- ceph-fuse
-- python
-mounts:
-  - [ 'none', '/mnt/cephfs', 'fuse.ceph', 'ceph.id=admin,ceph.conf=/etc/ceph/ceph.conf,_netdev,defaults', '0', '0' ]
-write_files:
--   content: |
-{}
-    path: /etc/ceph/ceph.conf
--   content: |
-{}
-    path: /etc/ceph/ceph.client.admin.keyring
-""" # noqa
+    mounts_share = False
+    mount_dir = '/mnt/cephfs'
+    CEPH_MON = 'ceph-mon'
+
+    def tearDown(self):
+        """Cleanup after running tests."""
+        if self.mounts_share:
+            for unit in ['ubuntu/0', 'ubuntu/1']:
+                try:
+                    zaza.utilities.generic.run_via_ssh(
+                        unit_name=unit,
+                        cmd='sudo fusermount -u {0} && sudo rmdir {0}'.format(
+                            self.mount_dir))
+                except subprocess.CalledProcessError:
+                    logging.warning(
+                        "Failed to cleanup mounts on {}".format(unit))
+
+    def _mount_share(self, unit_name: str,
+                     retry: bool = True):
+        self._install_dependencies(unit_name)
+        self._install_keyring(unit_name)
+        ssh_cmd = (
+            'sudo mkdir -p {0} && '
+            'sudo ceph-fuse {0}'.format(self.mount_dir)
+        )
+        if retry:
+            for attempt in Retrying(
+                    stop=stop_after_attempt(5),
+                    wait=wait_exponential(multiplier=3,
+                                          min=2, max=10)):
+                with attempt:
+                    zaza.utilities.generic.run_via_ssh(
+                        unit_name=unit_name,
+                        cmd=ssh_cmd)
+        else:
+            zaza.utilities.generic.run_via_ssh(
+                unit_name=unit_name,
+                cmd=ssh_cmd)
+        self.mounts_share = True
+
+    def _install_keyring(self, unit_name: str):
+
+        keyring = model.run_on_leader(
+            self.CEPH_MON, 'cat /etc/ceph/ceph.client.admin.keyring')['Stdout']
+        config = model.run_on_leader(
+            self.CEPH_MON, 'cat /etc/ceph/ceph.conf')['Stdout']
+        commands = [
+            'sudo mkdir -p /etc/ceph',
+            "echo '{}' | sudo tee /etc/ceph/ceph.conf".format(config),
+            "echo '{}' | "
+            'sudo tee /etc/ceph/ceph.client.admin.keyring'.format(keyring)
+        ]
+        for cmd in commands:
+            zaza.utilities.generic.run_via_ssh(
+                unit_name=unit_name,
+                cmd=cmd)
+
+    def _install_dependencies(self, unit: str):
+        zaza.utilities.generic.run_via_ssh(
+            unit_name=unit,
+            cmd='sudo apt-get install -yq ceph-fuse')
 
     @classmethod
     def setUpClass(cls):
         """Run class setup for running tests."""
         super(CephFSTests, cls).setUpClass()
 
+    @retry(
+        stop=stop_after_attempt(5),
+        wait=wait_exponential(multiplier=3, min=2, max=10))
+    def _write_testing_file_on_instance(self, instance_name: str):
+        zaza.utilities.generic.run_via_ssh(
+            unit_name=instance_name,
+            cmd='echo -n "test" | sudo tee {}/test'.format(self.mount_dir))
+
+    @retry(
+        stop=stop_after_attempt(5),
+        wait=wait_exponential(multiplier=3, min=2, max=10))
+    def _verify_testing_file_on_instance(self, instance_name: str):
+        output = zaza.model.run_on_unit(
+            instance_name, 'sudo cat {}/test'.format(self.mount_dir))['Stdout']
+        self.assertEqual('test', output.strip())
+
     def test_cephfs_share(self):
         """Test that CephFS shares can be accessed on two instances.
 
@@ -60,55 +122,11 @@ def test_cephfs_share(self):
         4. read it on the other
         5. profit
         """
-        keyring = model.run_on_leader(
-            'ceph-mon', 'cat /etc/ceph/ceph.client.admin.keyring')['Stdout']
-        conf = model.run_on_leader(
-            'ceph-mon', 'cat /etc/ceph/ceph.conf')['Stdout']
-        # Spawn Servers
-        instance_1, instance_2 = self.launch_guests(
-            userdata=self.INSTANCE_USERDATA.format(
-                _indent(conf, 8),
-                _indent(keyring, 8)))
-
-        # Write a file on instance_1
-        def verify_setup(stdin, stdout, stderr):
-            status = stdout.channel.recv_exit_status()
-            self.assertEqual(status, 0)
-
-        fip_1 = neutron_tests.floating_ips_from_instance(instance_1)[0]
-        fip_2 = neutron_tests.floating_ips_from_instance(instance_2)[0]
-        username = guest.boot_tests['bionic']['username']
-        password = guest.boot_tests['bionic'].get('password')
-        privkey = openstack_utils.get_private_key(nova_utils.KEYPAIR_NAME)
-
-        for attempt in Retrying(
-                stop=stop_after_attempt(3),
-                wait=wait_exponential(multiplier=1, min=2, max=10)):
-            with attempt:
-                openstack_utils.ssh_command(
-                    username, fip_1, 'instance-1',
-                    'sudo mount -a && '
-                    'echo "test" | sudo tee /mnt/cephfs/test',
-                    password=password, privkey=privkey, verify=verify_setup)
-
-        def verify(stdin, stdout, stderr):
-            status = stdout.channel.recv_exit_status()
-            self.assertEqual(status, 0)
-            out = ""
-            for line in iter(stdout.readline, ""):
-                out += line
-            self.assertEqual(out, "test\n")
-
-        openstack_utils.ssh_command(
-            username, fip_2, 'instance-2',
-            'sudo mount -a && '
-            'sudo cat /mnt/cephfs/test',
-            password=password, privkey=privkey, verify=verify)
-
-
-def _indent(text, amount, ch=' '):
-    padding = amount * ch
-    return ''.join(padding + line for line in text.splitlines(True))
+        self._mount_share('ubuntu/0')
+        self._mount_share('ubuntu/1')
+
+        self._write_testing_file_on_instance('ubuntu/0')
+        self._verify_testing_file_on_instance('ubuntu/1')
 
 
 class CharmOperationTest(test_utils.BaseCharmTest):