cinder_tempest_plugin/scenario/test_snapshots.py

# Copyright 2020 Red Hat, Inc.
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

from cinder_tempest_plugin.scenario import manager
from tempest.common import utils
from tempest.lib import decorators


class SnapshotDataIntegrityTests(manager.ScenarioTest):

    def setUp(self):
        super(SnapshotDataIntegrityTests, self).setUp()
        self.keypair = self.create_keypair()
        self.security_group = self._create_security_group()

    def _get_file_md5(self, ip_address, filename, mount_path='/mnt',
                      private_key=None, server=None):
        ssh_client = self.get_remote_client(ip_address,
                                            private_key=private_key,
                                            server=server)

        md5_sum = ssh_client.exec_command(
            'sudo md5sum %s/%s|cut -c 1-32' % (mount_path, filename))
        return md5_sum

    def _count_files(self, ip_address, mount_path='/mnt', private_key=None,
                     server=None):
        ssh_client = self.get_remote_client(ip_address,
                                            private_key=private_key,
                                            server=server)
        count = ssh_client.exec_command('sudo ls -l %s | wc -l' % mount_path)
        return int(count) - 1

    def _launch_instance_from_snapshot(self, snap):
        volume_snap = self.create_volume(snapshot_id=snap['id'],
                                         size=snap['size'])

        server_snap = self.boot_instance_from_resource(
            source_id=volume_snap['id'],
            source_type='volume',
            keypair=self.keypair,
            security_group=self.security_group)

        return server_snap

    def create_md5_new_file(self, ip_address, filename, mount_path='/mnt',
                            private_key=None, server=None):
        ssh_client = self.get_remote_client(ip_address,
                                            private_key=private_key,
                                            server=server)

        ssh_client.exec_command(
            'sudo dd bs=1024 count=100 if=/dev/urandom of=/%s/%s' %
            (mount_path, filename))
        md5 = ssh_client.exec_command(
            'sudo md5sum -b %s/%s|cut -c 1-32' % (mount_path, filename))
        ssh_client.exec_command('sudo sync')
        return md5

    def get_md5_from_file(self, instance, filename):

        instance_ip = self.get_server_ip(instance)

        md5_sum = self._get_file_md5(instance_ip, filename=filename,
                                     private_key=self.keypair['private_key'],
                                     server=instance)
        count = self._count_files(instance_ip,
                                  private_key=self.keypair['private_key'],
                                  server=instance)
        return count, md5_sum

    @decorators.idempotent_id('ff10644e-5a70-4a9f-9801-8204bb81fb61')
    @utils.services('compute', 'volume', 'image', 'network')
    def test_snapshot_data_integrity(self):
        """This test checks the data integrity after creating and restoring

        snapshots. The procedure is as follows:

        1) create a volume from image
        2) Boot an instance from the volume
        3) create file on vm and write data into it
        4) create snapshot
        5) repeat 3 and 4 two more times (simply creating 3 snapshots)

        Now restore the snapshots one by one into volume, create instances
        from it and check the number of files and file content at each
        point when snapshot was created.
        """

        # Create a volume from image
        volume = self.create_volume_from_image()

        # create an instance from bootable volume
        server = self.boot_instance_from_resource(
            source_id=volume['id'],
            source_type='volume',
            keypair=self.keypair,
            security_group=self.security_group)

        instance_ip = self.get_server_ip(server)

        # Write data to volume
        file1_md5 = self.create_md5_new_file(
            instance_ip, filename="file1",
            private_key=self.keypair['private_key'],
            server=instance_ip)

        # Create first snapshot
        snapshot1 = self.create_volume_snapshot(volume['id'], force=True)

        # Write data to volume
        file2_md5 = self.create_md5_new_file(
            instance_ip, filename="file2",
            private_key=self.keypair['private_key'],
            server=instance_ip)

        # Create second snapshot
        snapshot2 = self.create_volume_snapshot(volume['id'], force=True)

        # Write data to volume
        file3_md5 = self.create_md5_new_file(
            instance_ip, filename="file3",
            private_key=self.keypair['private_key'],
            server=instance_ip)

        # Create third snapshot
        snapshot3 = self.create_volume_snapshot(volume['id'], force=True)

        # Create volume, instance and check file and contents for snap1
        instance_1 = self._launch_instance_from_snapshot(snapshot1)
        count_snap_1, md5_file_1 = self.get_md5_from_file(instance_1,
                                                          'file1')

        self.assertEqual(count_snap_1, 1)
        self.assertEqual(file1_md5, md5_file_1)

        # Create volume, instance and check file and contents for snap2
        instance_2 = self._launch_instance_from_snapshot(snapshot2)
        count_snap_2, md5_file_2 = self.get_md5_from_file(instance_2,
                                                          'file2')

        self.assertEqual(count_snap_2, 2)
        self.assertEqual(file2_md5, md5_file_2)

        # Create volume, instance and check file and contents for snap3
        instance_3 = self._launch_instance_from_snapshot(snapshot3)
        count_snap_3, md5_file_3 = self.get_md5_from_file(instance_3,
                                                          'file3')

        self.assertEqual(count_snap_3, 3)
        self.assertEqual(file3_md5, md5_file_3)