KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _attached_volume_name method in tempest

Best Python code snippet using tempest_python

manager.py

Source:manager.py

...26    @classmethod27    def setup_clients(cls):28        super(ScenarioTest, cls).setup_clients()29        cls.admin_volume_types_client = cls.os_admin.volume_types_client_latest30    def _attached_volume_name(31            self, disks_list_before_attach, ip_address, private_key):32        ssh = self.get_remote_client(ip_address, private_key=private_key)33        def _wait_for_volume_available_on_system():34            disks_list_after_attach = ssh.list_disks()35            return len(disks_list_after_attach) > len(disks_list_before_attach)36        if not test_utils.call_until_true(_wait_for_volume_available_on_system,37                                          CONF.compute.build_timeout,38                                          CONF.compute.build_interval):39            raise lib_exc.TimeoutException40        disks_list_after_attach = ssh.list_disks()41        volume_name = [item for item in disks_list_after_attach42                       if item not in disks_list_before_attach][0]43        return volume_name44    def _get_file_md5(self, ip_address, filename, dev_name=None,45                      mount_path='/mnt', private_key=None, server=None):46        ssh_client = self.get_remote_client(ip_address,47                                            private_key=private_key,48                                            server=server)49        if dev_name is not None:50            ssh_client.exec_command('sudo mount /dev/%s %s' % (dev_name,51                                                               mount_path))52        md5_sum = ssh_client.exec_command(53            'sudo md5sum %s/%s|cut -c 1-32' % (mount_path, filename))54        if dev_name is not None:55            ssh_client.exec_command('sudo umount %s' % mount_path)56        return md5_sum57    def _count_files(self, ip_address, dev_name=None, mount_path='/mnt',58                     private_key=None, server=None):59        ssh_client = self.get_remote_client(ip_address,60                                            private_key=private_key,61                                            server=server)62        if dev_name is not None:63            ssh_client.exec_command('sudo mount /dev/%s %s' % (dev_name,64                                                               mount_path))65        count = ssh_client.exec_command('sudo ls -l %s | wc -l' % mount_path)66        if dev_name is not None:67            ssh_client.exec_command('sudo umount %s' % mount_path)68        # We subtract 2 from the count since `wc -l` also includes the count69        # of new line character and while creating the filesystem, a70        # lost+found folder is also created71        return int(count) - 272    def _make_fs(self, ip_address, private_key, server, dev_name, fs='ext4'):73        ssh_client = self.get_remote_client(ip_address,74                                            private_key=private_key,75                                            server=server)76        ssh_client.make_fs(dev_name, fs=fs)77    def create_md5_new_file(self, ip_address, filename, dev_name=None,78                            mount_path='/mnt', private_key=None, server=None):79        ssh_client = self.get_remote_client(ip_address,80                                            private_key=private_key,81                                            server=server)82        if dev_name is not None:83            ssh_client.exec_command('sudo mount /dev/%s %s' % (dev_name,84                                                               mount_path))85        ssh_client.exec_command(86            'sudo dd bs=1024 count=100 if=/dev/urandom of=/%s/%s' %87            (mount_path, filename))88        md5 = ssh_client.exec_command(89            'sudo md5sum -b %s/%s|cut -c 1-32' % (mount_path, filename))90        ssh_client.exec_command('sudo sync')91        if dev_name is not None:92            ssh_client.exec_command('sudo umount %s' % mount_path)93        return md594    def get_md5_from_file(self, instance, instance_ip, filename,95                          dev_name=None):96        md5_sum = self._get_file_md5(instance_ip, filename=filename,97                                     dev_name=dev_name,98                                     private_key=self.keypair['private_key'],99                                     server=instance)100        count = self._count_files(instance_ip, dev_name=dev_name,101                                  private_key=self.keypair['private_key'],102                                  server=instance)103        return count, md5_sum104    def _attach_and_get_volume_device_name(self, server, volume, instance_ip,105                                           private_key):106        ssh_client = self.get_remote_client(107            instance_ip, private_key=private_key,108            server=server)109        # List disks before volume attachment110        disks_list_before_attach = ssh_client.list_disks()111        # Attach volume112        attachment = self.attach_volume(server, volume)113        # Find the difference between disks before and after attachment that114        # gives us the volume device name115        volume_device_name = self._attached_volume_name(116            disks_list_before_attach, instance_ip, private_key)117        return volume_device_name, attachment118    def create_volume_type(self, client=None, name=None, extra_specs=None):119        if not client:120            client = self.os_admin.volume_types_client_latest121        if not name:122            class_name = self.__class__.__name__123            name = data_utils.rand_name(class_name + '-volume-type')124        randomized_name = data_utils.rand_name('scenario-type-' + name)125        LOG.debug("Creating a volume type: %s with extra_specs %s",126                  randomized_name, extra_specs)127        if extra_specs is None:128            extra_specs = {}129        volume_type = self.admin_volume_types_client.create_volume_type(...

test_stamp_pattern.py

Source:test_stamp_pattern.py

...44    def skip_checks(cls):45        super(TestStampPattern, cls).skip_checks()46        if not CONF.volume_feature_enabled.snapshot:47            raise cls.skipException("Cinder volume snapshots are disabled")48    def _attached_volume_name(49            self, disks_list_before_attach, ip_address, private_key):50        ssh = self.get_remote_client(ip_address, private_key=private_key)51        def _wait_for_volume_available_on_system():52            disks_list_after_attach = ssh.list_disks()53            return len(disks_list_after_attach) > len(disks_list_before_attach)54        if not test_utils.call_until_true(_wait_for_volume_available_on_system,55                                          CONF.compute.build_timeout,56                                          CONF.compute.build_interval):57            raise lib_exc.TimeoutException58        disks_list_after_attach = ssh.list_disks()59        volume_name = [item for item in disks_list_after_attach60                       if item not in disks_list_before_attach][0]61        return volume_name62    @decorators.attr(type='slow')63    @decorators.idempotent_id('10fd234a-515c-41e5-b092-8323060598c5')64    @testtools.skipUnless(CONF.compute_feature_enabled.snapshot,65                          'Snapshotting is not available.')66    @testtools.skipUnless(CONF.network.public_network_id,67                          'The public_network_id option must be specified.')68    @utils.services('compute', 'network', 'volume', 'image')69    def test_stamp_pattern(self):70        # prepare for booting an instance71        keypair = self.create_keypair()72        security_group = self.create_security_group()73        # boot an instance and create a timestamp file in it74        volume = self.create_volume()75        server = self.create_server(76            key_name=keypair['name'],77            security_groups=[{'name': security_group['name']}])78        # create and add floating IP to server179        ip_for_server = self.get_server_ip(server)80        # Make sure the machine ssh-able before attaching the volume81        linux_client = self.get_remote_client(82            ip_for_server, private_key=keypair['private_key'],83            server=server)84        disks_list_before_attach = linux_client.list_disks()85        self.nova_volume_attach(server, volume)86        volume_device_name = self._attached_volume_name(87            disks_list_before_attach, ip_for_server, keypair['private_key'])88        timestamp = self.create_timestamp(ip_for_server,89                                          volume_device_name,90                                          private_key=keypair['private_key'],91                                          server=server)92        self.nova_volume_detach(server, volume)93        # snapshot the volume94        volume_snapshot = self.create_volume_snapshot(volume['id'])95        # snapshot the instance96        snapshot_image = self.create_server_snapshot(server=server)97        # create second volume from the snapshot(volume2)98        volume_from_snapshot = self.create_volume(99            snapshot_id=volume_snapshot['id'])100        # boot second instance from the snapshot(instance2)101        server_from_snapshot = self.create_server(102            image_id=snapshot_image['id'],103            key_name=keypair['name'],104            security_groups=[{'name': security_group['name']}])105        # create and add floating IP to server_from_snapshot106        ip_for_snapshot = self.get_server_ip(server_from_snapshot)107        # Make sure the machine ssh-able before attaching the volume108        # Just a live machine is responding109        # for device attache/detach as expected110        linux_client = self.get_remote_client(111            ip_for_snapshot, private_key=keypair['private_key'],112            server=server_from_snapshot)113        disks_list_before_attach = linux_client.list_disks()114        # attach volume2 to instance2115        self.nova_volume_attach(server_from_snapshot, volume_from_snapshot)116        volume_device_name = self._attached_volume_name(117            disks_list_before_attach, ip_for_snapshot, keypair['private_key'])118        # check the existence of the timestamp file in the volume2119        timestamp2 = self.get_timestamp(ip_for_snapshot,120                                        volume_device_name,121                                        private_key=keypair['private_key'],122                                        server=server_from_snapshot)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.