Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use _count_volumes method in tempest

Best Python code snippet using tempest_python

test_attach_volume.py

Source:test_attach_volume.py

...132    a shelved or shelved offload instance.133    """134    min_microversion = '2.20'135    max_microversion = 'latest'136    def _count_volumes(self, server):137        # Count number of volumes on an instance138        volumes = 0139        if CONF.validation.run_validation:140            linux_client = remote_client.RemoteClient(141                self.get_server_ip(server),142                self.image_ssh_user,143                self.image_ssh_password,144                self.validation_resources['keypair']['private_key'],145                server=server,146                servers_client=self.servers_client)147            command = 'grep -c -E [vs]d.$ /proc/partitions'148            volumes = int(linux_client.exec_command(command).strip())149        return volumes150    def _shelve_server(self, server):151        # NOTE(andreaf) If we are going to shelve a server, we should152        # check first whether the server is ssh-able. Otherwise we153        # won't be able to distinguish failures introduced by shelve154        # from pre-existing ones. Also it's good to wait for cloud-init155        # to be done and sshd server to be running before shelving to156        # avoid breaking the VM157        if CONF.validation.run_validation:158            linux_client = remote_client.RemoteClient(159                self.get_server_ip(server),160                self.image_ssh_user,161                self.image_ssh_password,162                self.validation_resources['keypair']['private_key'],163                server=server,164                servers_client=self.servers_client)165            linux_client.validate_authentication()166        # If validation went ok, or it was skipped, shelve the server167        compute.shelve_server(self.servers_client, server['id'])168    def _unshelve_server_and_check_volumes(self, server, number_of_volumes):169        # Unshelve the instance and check that there are expected volumes170        self.servers_client.unshelve_server(server['id'])171        waiters.wait_for_server_status(self.servers_client,172                                       server['id'],173                                       'ACTIVE')174        if CONF.validation.run_validation:175            counted_volumes = self._count_volumes(server)176            self.assertEqual(number_of_volumes, counted_volumes)177    @test.idempotent_id('13a940b6-3474-4c3c-b03f-29b89112bfee')178    @testtools.skipUnless(CONF.compute_feature_enabled.shelve,179                          'Shelve is not available.')180    def test_attach_volume_shelved_or_offload_server(self):181        # Create server, count number of volumes on it, shelve182        # server and attach pre-created volume to shelved server183        server = self._create_server()184        num_vol = self._count_volumes(server)185        self._shelve_server(server)186        self._create_and_attach_volume(server)187        # Unshelve the instance and check that attached volume exists188        self._unshelve_server_and_check_volumes(server, num_vol + 1)189        # Get Volume attachment of the server190        volume_attachment = self.servers_client.show_volume_attachment(191            server['id'],192            self.attachment['id'])['volumeAttachment']193        self.assertEqual(server['id'], volume_attachment['serverId'])194        self.assertEqual(self.attachment['id'], volume_attachment['id'])195        # Check the mountpoint is not None after unshelve server even in196        # case of shelved_offloaded.197        self.assertIsNotNone(volume_attachment['device'])198    @test.idempotent_id('b54e86dd-a070-49c4-9c07-59ae6dae15aa')199    @testtools.skipUnless(CONF.compute_feature_enabled.shelve,200                          'Shelve is not available.')201    def test_detach_volume_shelved_or_offload_server(self):202        # Create server, count number of volumes on it, shelve203        # server and attach pre-created volume to shelved server204        server = self._create_server()205        num_vol = self._count_volumes(server)206        self._shelve_server(server)207        volume = self._create_and_attach_volume(server)208        # Detach the volume209        self._detach(server['id'], volume['id'])210        self.attachment = None211        # Unshelve the instance and check that we have the expected number of212        # volume(s)...

test_06_cleanup.py

Source:test_06_cleanup.py

...6@pytest.mark.usefixtures('client_class', 'docker_client', 'shepherd')7class TestCleanup(object):8    def _count_containers(self, docker_client, shepherd):9        return len(docker_client.containers.list(filters={'label': shepherd.reqid_label}, ignore_removed=True))10    def _count_volumes(self, docker_client, shepherd):11        return len(docker_client.volumes.list(filters={'label': shepherd.reqid_label}))12    def _count_networks(self, docker_client, shepherd):13        return len(docker_client.networks.list(filters={'label': shepherd.network_pool.network_label}))14    def test_start_loop(self, shepherd):15        assert shepherd.untracked_check_time == 016        shepherd.start_cleanup_loop(2.0)17        assert shepherd.untracked_check_time == 2.018    def test_ensure_flock_stop(self, docker_client):19        res = self.client.post('/api/flock/request/test_b')20        reqid = res.json['reqid']21        res = self.client.post('/api/flock/start/{0}'.format(reqid))22        assert res.json['containers']23        box = docker_client.containers.get(res.json['containers']['box']['id'])24        box_2 = docker_client.containers.get(res.json['containers']['box-2']['id'])25        box.remove(force=True)26        def assert_removed():27            with pytest.raises(docker.errors.NotFound):28                box = docker_client.containers.get(res.json['containers']['box-2']['id'])29        sleep_try(0.3, 10.0, assert_removed)30    def test_check_untracked_cleanup(self, docker_client, redis, shepherd):31        num_containers = self._count_containers(docker_client, shepherd)32        num_volumes = self._count_volumes(docker_client, shepherd)33        num_networks = self._count_volumes(docker_client, shepherd)34        for x in range(0, 3):35            res = self.client.post('/api/flock/request/test_vol')36            reqid = res.json['reqid']37            res = self.client.post('/api/flock/start/{0}'.format(reqid))38            assert res.json['containers']39        assert num_containers < self._count_containers(docker_client, shepherd)40        assert num_volumes < self._count_volumes(docker_client, shepherd)41        assert num_networks < self._count_networks(docker_client, shepherd)42        # wipe all redis data43        redis.flushdb()44        def assert_removed():45            assert num_containers == self._count_containers(docker_client, shepherd)46            assert num_volumes == self._count_volumes(docker_client, shepherd)47            #assert num_networks == self._count_networks(docker_client, shepherd)48        sleep_try(2.0, 20.0, assert_removed)49    def test_redis_pool_and_reqid_cleanup(self, docker_client, redis):50        reqids = []51        # start and kill containers52        res = self.client.post('/api/flock/request/test_vol', json={'user_params': {'foo': 'bar'}})53        reqid = res.json['reqid']54        res = self.client.post('/api/flock/start/{0}'.format(reqid))55        reqids.append(reqid)56        try:57            docker_client.containers.get(res.json['containers']['box-1']['id']).remove(force=True)58        except:59            pass60        try:...

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.