How to use _count_volumes method in tempest

Best Python code snippet using tempest_python

test_attach_volume.py

Source:test_attach_volume.py Github

copy

Full Screen

...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)...

Full Screen

Full Screen

test_06_cleanup.py

Source:test_06_cleanup.py Github

copy

Full Screen

...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:...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run tempest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful