Best Python code snippet using tempest_python
test_object_services.py
Source:test_object_services.py
...88 metadata,89 metadata_prefix='')90 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)91 self.assertHeaders(resp, 'Object', 'POST')92 resp, _ = self.object_client.list_object_metadata(93 self.container_name,94 object_name)95 self.assertIn('x-object-meta-test-meta', resp)96 self.assertEqual(resp['x-object-meta-test-meta'], 'Meta')97 def test_update_object_metadata_with_remove_metadata(self):98 # update object metadata with remove metadata99 object_name = data_utils.rand_name(name='TestObject')100 data = data_utils.arbitrary_string()101 create_metadata = {'X-Object-Meta-test-meta1': 'Meta1'}102 self.object_client.create_object(self.container_name,103 object_name,104 data,105 metadata=create_metadata)106 update_metadata = {'X-Remove-Object-Meta-test-meta1': 'Meta1'}107 resp, _ = self.object_client.update_object_metadata(108 self.container_name,109 object_name,110 update_metadata,111 metadata_prefix='')112 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)113 self.assertHeaders(resp, 'Object', 'POST')114 resp, _ = self.object_client.list_object_metadata(115 self.container_name,116 object_name)117 self.assertNotIn('x-object-meta-test-meta1', resp)118 @test.attr(type='smoke')119 def test_update_object_metadata_with_create_and_remove_metadata(self):120 # creation and deletion of metadata with one request121 object_name = data_utils.rand_name(name='TestObject')122 data = data_utils.arbitrary_string()123 create_metadata = {'X-Object-Meta-test-meta1': 'Meta1'}124 self.object_client.create_object(self.container_name,125 object_name,126 data,127 metadata=create_metadata)128 update_metadata = {'X-Object-Meta-test-meta2': 'Meta2',129 'X-Remove-Object-Meta-test-meta1': 'Meta1'}130 resp, _ = self.object_client.update_object_metadata(131 self.container_name,132 object_name,133 update_metadata,134 metadata_prefix='')135 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)136 self.assertHeaders(resp, 'Object', 'POST')137 resp, _ = self.object_client.list_object_metadata(138 self.container_name,139 object_name)140 self.assertNotIn('x-object-meta-test-meta1', resp)141 self.assertIn('x-object-meta-test-meta2', resp)142 self.assertEqual(resp['x-object-meta-test-meta2'], 'Meta2')143 @test.attr(type='smoke')144 def test_update_object_metadata_with_x_object_manifest(self):145 # update object metadata with x_object_manifest146 # uploading segments147 object_name, data_segments = self._upload_segments()148 # creating a manifest file149 data_empty = ''150 self.object_client.create_object(self.container_name,151 object_name,152 data_empty,153 metadata=None)154 object_prefix = '%s/%s' % (self.container_name, object_name)155 update_metadata = {'X-Object-Manifest': object_prefix}156 resp, _ = self.object_client.update_object_metadata(157 self.container_name,158 object_name,159 update_metadata,160 metadata_prefix='')161 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)162 self.assertHeaders(resp, 'Object', 'POST')163 resp, _ = self.object_client.list_object_metadata(164 self.container_name,165 object_name)166 self.assertIn('x-object-manifest', resp)167 self.assertNotEqual(len(resp['x-object-manifest']), 0)168 def test_update_object_metadata_with_x_object_metakey(self):169 # update object metadata with a blenk value of metadata170 object_name, data = self._create_object()171 update_metadata = {'X-Object-Meta-test-meta': ''}172 resp, _ = self.object_client.update_object_metadata(173 self.container_name,174 object_name,175 update_metadata,176 metadata_prefix='')177 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)178 self.assertHeaders(resp, 'Object', 'POST')179 resp, _ = self.object_client.list_object_metadata(180 self.container_name,181 object_name)182 self.assertIn('x-object-meta-test-meta', resp)183 self.assertEqual(resp['x-object-meta-test-meta'], '')184 @test.attr(type='smoke')185 def test_update_object_metadata_with_x_remove_object_metakey(self):186 # update object metadata with a blank value of remove metadata187 object_name = data_utils.rand_name(name='TestObject')188 data = data_utils.arbitrary_string()189 create_metadata = {'X-Object-Meta-test-meta': 'Meta'}190 self.object_client.create_object(self.container_name,191 object_name,192 data,193 metadata=create_metadata)194 update_metadata = {'X-Remove-Object-Meta-test-meta': ''}195 resp, _ = self.object_client.update_object_metadata(196 self.container_name,197 object_name,198 update_metadata,199 metadata_prefix='')200 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)201 self.assertHeaders(resp, 'Object', 'POST')202 resp, _ = self.object_client.list_object_metadata(203 self.container_name,204 object_name)205 self.assertNotIn('x-object-meta-test-meta', resp)206 @test.attr(type='smoke')207 def test_list_object_metadata(self):208 # get object metadata209 object_name = data_utils.rand_name(name='TestObject')210 data = data_utils.arbitrary_string()211 metadata = {'X-Object-Meta-test-meta': 'Meta'}212 self.object_client.create_object(self.container_name,213 object_name,214 data,215 metadata=metadata)216 resp, _ = self.object_client.list_object_metadata(217 self.container_name,218 object_name)219 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)220 self.assertHeaders(resp, 'Object', 'HEAD')221 self.assertIn('x-object-meta-test-meta', resp)222 self.assertEqual(resp['x-object-meta-test-meta'], 'Meta')223 @test.attr(type='smoke')224 def test_list_no_object_metadata(self):225 # get empty list of object metadata226 object_name, data = self._create_object()227 resp, _ = self.object_client.list_object_metadata(228 self.container_name,229 object_name)230 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)231 self.assertHeaders(resp, 'Object', 'HEAD')232 self.assertNotIn('x-object-meta-', str(resp))233 @test.attr(type='smoke')234 def test_list_object_metadata_with_x_object_manifest(self):235 # get object metadata with x_object_manifest236 # uploading segments237 object_name, data_segments = self._upload_segments()238 # creating a manifest file239 object_prefix = '%s/%s' % (self.container_name, object_name)240 metadata = {'X-Object-Manifest': object_prefix}241 data_empty = ''242 resp, _ = self.object_client.create_object(243 self.container_name,244 object_name,245 data_empty,246 metadata=metadata)247 resp, _ = self.object_client.list_object_metadata(248 self.container_name,249 object_name)250 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)251 # Check only the existence of common headers with custom matcher252 self.assertThat(resp, custom_matchers.ExistsAllResponseHeaders(253 'Object', 'HEAD'))254 self.assertIn('x-object-manifest', resp)255 # Etag value of a large object is enclosed in double-quotations.256 # This is a special case, therefore the formats of response headers257 # are checked without a custom matcher.258 self.assertTrue(resp['etag'].startswith('\"'))259 self.assertTrue(resp['etag'].endswith('\"'))260 self.assertTrue(resp['etag'].strip('\"').isalnum())261 self.assertTrue(re.match("^\d+\.?\d*\Z", resp['x-timestamp']))262 self.assertNotEqual(len(resp['content-type']), 0)263 self.assertTrue(re.match("^tx[0-9a-f]*-[0-9a-f]*$",264 resp['x-trans-id']))265 self.assertNotEqual(len(resp['date']), 0)266 self.assertEqual(resp['accept-ranges'], 'bytes')267 self.assertEqual(resp['x-object-manifest'],268 '%s/%s' % (self.container_name, object_name))269 @test.attr(type='smoke')270 def test_get_object(self):271 # retrieve object's data (in response body)272 # create object273 object_name, data = self._create_object()274 # get object275 resp, body = self.object_client.get_object(self.container_name,276 object_name)277 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)278 self.assertHeaders(resp, 'Object', 'GET')279 self.assertEqual(body, data)280 @test.attr(type='smoke')281 def test_get_object_with_metadata(self):282 # get object with metadata283 object_name = data_utils.rand_name(name='TestObject')284 data = data_utils.arbitrary_string()285 metadata = {'X-Object-Meta-test-meta': 'Meta'}286 self.object_client.create_object(self.container_name,287 object_name,288 data,289 metadata=metadata)290 resp, body = self.object_client.get_object(291 self.container_name,292 object_name,293 metadata=None)294 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)295 self.assertHeaders(resp, 'Object', 'GET')296 self.assertIn('x-object-meta-test-meta', resp)297 self.assertEqual(resp['x-object-meta-test-meta'], 'Meta')298 self.assertEqual(body, data)299 @test.attr(type='smoke')300 def test_get_object_with_range(self):301 # get object with range302 object_name = data_utils.rand_name(name='TestObject')303 data = data_utils.arbitrary_string(100)304 self.object_client.create_object(self.container_name,305 object_name,306 data,307 metadata=None)308 rand_num = random.randint(3, len(data) - 1)309 metadata = {'Range': 'bytes=%s-%s' % (rand_num - 3, rand_num - 1)}310 resp, body = self.object_client.get_object(311 self.container_name,312 object_name,313 metadata=metadata)314 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)315 self.assertHeaders(resp, 'Object', 'GET')316 self.assertEqual(body, data[rand_num - 3: rand_num])317 @test.attr(type='smoke')318 def test_get_object_with_x_object_manifest(self):319 # get object with x_object_manifest320 # uploading segments321 object_name, data_segments = self._upload_segments()322 # creating a manifest file323 object_prefix = '%s/%s' % (self.container_name, object_name)324 metadata = {'X-Object-Manifest': object_prefix}325 data_empty = ''326 resp, body = self.object_client.create_object(327 self.container_name,328 object_name,329 data_empty,330 metadata=metadata)331 resp, body = self.object_client.get_object(332 self.container_name,333 object_name,334 metadata=None)335 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)336 # Check only the existence of common headers with custom matcher337 self.assertThat(resp, custom_matchers.ExistsAllResponseHeaders(338 'Object', 'GET'))339 self.assertIn('x-object-manifest', resp)340 # Etag value of a large object is enclosed in double-quotations.341 # This is a special case, therefore the formats of response headers342 # are checked without a custom matcher.343 self.assertTrue(resp['etag'].startswith('\"'))344 self.assertTrue(resp['etag'].endswith('\"'))345 self.assertTrue(resp['etag'].strip('\"').isalnum())346 self.assertTrue(re.match("^\d+\.?\d*\Z", resp['x-timestamp']))347 self.assertNotEqual(len(resp['content-type']), 0)348 self.assertTrue(re.match("^tx[0-9a-f]*-[0-9a-f]*$",349 resp['x-trans-id']))350 self.assertNotEqual(len(resp['date']), 0)351 self.assertEqual(resp['accept-ranges'], 'bytes')352 self.assertEqual(resp['x-object-manifest'],353 '%s/%s' % (self.container_name, object_name))354 self.assertEqual(''.join(data_segments), body)355 @test.attr(type='smoke')356 def test_get_object_with_if_match(self):357 # get object with if_match358 object_name = data_utils.rand_name(name='TestObject')359 data = data_utils.arbitrary_string(10)360 create_md5 = hashlib.md5(data).hexdigest()361 create_metadata = {'Etag': create_md5}362 self.object_client.create_object(self.container_name,363 object_name,364 data,365 metadata=create_metadata)366 list_metadata = {'If-Match': create_md5}367 resp, body = self.object_client.get_object(368 self.container_name,369 object_name,370 metadata=list_metadata)371 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)372 self.assertHeaders(resp, 'Object', 'GET')373 self.assertEqual(body, data)374 @test.attr(type='smoke')375 def test_get_object_with_if_modified_since(self):376 # get object with if_modified_since377 object_name = data_utils.rand_name(name='TestObject')378 data = data_utils.arbitrary_string()379 time_now = time.time()380 self.object_client.create_object(self.container_name,381 object_name,382 data,383 metadata=None)384 http_date = time.ctime(time_now - 86400)385 list_metadata = {'If-Modified-Since': http_date}386 resp, body = self.object_client.get_object(387 self.container_name,388 object_name,389 metadata=list_metadata)390 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)391 self.assertHeaders(resp, 'Object', 'GET')392 self.assertEqual(body, data)393 def test_get_object_with_if_none_match(self):394 # get object with if_none_match395 object_name = data_utils.rand_name(name='TestObject')396 data = data_utils.arbitrary_string(10)397 create_md5 = hashlib.md5(data).hexdigest()398 create_metadata = {'Etag': create_md5}399 self.object_client.create_object(self.container_name,400 object_name,401 data,402 metadata=create_metadata)403 list_data = data_utils.arbitrary_string(15)404 list_md5 = hashlib.md5(list_data).hexdigest()405 list_metadata = {'If-None-Match': list_md5}406 resp, body = self.object_client.get_object(407 self.container_name,408 object_name,409 metadata=list_metadata)410 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)411 self.assertHeaders(resp, 'Object', 'GET')412 self.assertEqual(body, data)413 @test.attr(type='smoke')414 def test_get_object_with_if_unmodified_since(self):415 # get object with if_unmodified_since416 object_name, data = self._create_object()417 time_now = time.time()418 http_date = time.ctime(time_now + 86400)419 list_metadata = {'If-Unmodified-Since': http_date}420 resp, body = self.object_client.get_object(421 self.container_name,422 object_name,423 metadata=list_metadata)424 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)425 self.assertHeaders(resp, 'Object', 'GET')426 self.assertEqual(body, data)427 @test.attr(type='smoke')428 def test_get_object_with_x_newest(self):429 # get object with x_newest430 object_name, data = self._create_object()431 list_metadata = {'X-Newest': 'true'}432 resp, body = self.object_client.get_object(433 self.container_name,434 object_name,435 metadata=list_metadata)436 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)437 self.assertHeaders(resp, 'Object', 'GET')438 self.assertEqual(body, data)439 @test.attr(type='smoke')440 def test_copy_object_in_same_container(self):441 # create source object442 src_object_name = data_utils.rand_name(name='SrcObject')443 src_data = data_utils.arbitrary_string(size=len(src_object_name) * 2,444 base_text=src_object_name)445 resp, _ = self.object_client.create_object(self.container_name,446 src_object_name,447 src_data)448 # create destination object449 dst_object_name = data_utils.rand_name(name='DstObject')450 dst_data = data_utils.arbitrary_string(size=len(dst_object_name) * 3,451 base_text=dst_object_name)452 resp, _ = self.object_client.create_object(self.container_name,453 dst_object_name,454 dst_data)455 # copy source object to destination456 resp, _ = self.object_client.copy_object_in_same_container(457 self.container_name, src_object_name, dst_object_name)458 self.assertEqual(resp['status'], '201')459 self.assertHeaders(resp, 'Object', 'PUT')460 # check data461 resp, body = self.object_client.get_object(self.container_name,462 dst_object_name)463 self.assertEqual(body, src_data)464 @test.attr(type='smoke')465 def test_copy_object_to_itself(self):466 # change the content type of an existing object467 # create object468 object_name = data_utils.rand_name(name='TestObject')469 data = data_utils.arbitrary_string()470 self.object_client.create_object(self.container_name,471 object_name, data)472 # get the old content type473 resp_tmp, _ = self.object_client.list_object_metadata(474 self.container_name, object_name)475 # change the content type of the object476 metadata = {'content-type': 'text/plain; charset=UTF-8'}477 self.assertNotEqual(resp_tmp['content-type'], metadata['content-type'])478 resp, _ = self.object_client.copy_object_in_same_container(479 self.container_name, object_name, object_name, metadata)480 self.assertEqual(resp['status'], '201')481 self.assertHeaders(resp, 'Object', 'PUT')482 # check the content type483 resp, _ = self.object_client.list_object_metadata(self.container_name,484 object_name)485 self.assertEqual(resp['content-type'], metadata['content-type'])486 @test.attr(type='smoke')487 def test_copy_object_2d_way(self):488 # create source object489 src_object_name = data_utils.rand_name(name='SrcObject')490 src_data = data_utils.arbitrary_string(size=len(src_object_name) * 2,491 base_text=src_object_name)492 resp, _ = self.object_client.create_object(self.container_name,493 src_object_name, src_data)494 # create destination object495 dst_object_name = data_utils.rand_name(name='DstObject')496 dst_data = data_utils.arbitrary_string(size=len(dst_object_name) * 3,497 base_text=dst_object_name)498 resp, _ = self.object_client.create_object(self.container_name,499 dst_object_name, dst_data)500 # copy source object to destination501 resp, _ = self.object_client.copy_object_2d_way(self.container_name,502 src_object_name,503 dst_object_name)504 self.assertEqual(resp['status'], '201')505 self.assertHeaders(resp, 'Object', 'COPY')506 self.assertIn('last-modified', resp)507 self.assertIn('x-copied-from', resp)508 self.assertIn('x-copied-from-last-modified', resp)509 self.assertNotEqual(len(resp['last-modified']), 0)510 self.assertEqual(511 resp['x-copied-from'],512 self.container_name + "/" + src_object_name)513 self.assertNotEqual(len(resp['x-copied-from-last-modified']), 0)514 # check data515 resp, body = self.object_client.get_object(self.container_name,516 dst_object_name)517 self.assertEqual(body, src_data)518 @test.attr(type='smoke')519 def test_copy_object_across_containers(self):520 # create a container to use as asource container521 src_container_name = data_utils.rand_name(name='TestSourceContainer')522 self.container_client.create_container(src_container_name)523 self.containers.append(src_container_name)524 # create a container to use as a destination container525 dst_container_name = data_utils.rand_name(526 name='TestDestinationContainer')527 self.container_client.create_container(dst_container_name)528 self.containers.append(dst_container_name)529 # create object in source container530 object_name = data_utils.rand_name(name='Object')531 data = data_utils.arbitrary_string(size=len(object_name) * 2,532 base_text=object_name)533 resp, _ = self.object_client.create_object(src_container_name,534 object_name, data)535 # set object metadata536 meta_key = data_utils.rand_name(name='test-')537 meta_value = data_utils.rand_name(name='MetaValue-')538 orig_metadata = {meta_key: meta_value}539 resp, _ = self.object_client.update_object_metadata(src_container_name,540 object_name,541 orig_metadata)542 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)543 self.assertHeaders(resp, 'Object', 'POST')544 # copy object from source container to destination container545 resp, _ = self.object_client.copy_object_across_containers(546 src_container_name, object_name, dst_container_name,547 object_name)548 self.assertEqual(resp['status'], '201')549 self.assertHeaders(resp, 'Object', 'PUT')550 # check if object is present in destination container551 resp, body = self.object_client.get_object(dst_container_name,552 object_name)553 self.assertEqual(body, data)554 actual_meta_key = 'x-object-meta-' + meta_key555 self.assertIn(actual_meta_key, resp)556 self.assertEqual(resp[actual_meta_key], meta_value)557 @test.attr(type='gate')558 def test_object_upload_in_segments(self):559 # create object560 object_name = data_utils.rand_name(name='LObject')561 data = data_utils.arbitrary_string()562 segments = 10563 data_segments = [data + str(i) for i in moves.xrange(segments)]564 # uploading segments565 for i in moves.xrange(segments):566 resp, _ = self.object_client.create_object_segments(567 self.container_name, object_name, i, data_segments[i])568 self.assertEqual(resp['status'], '201')569 # creating a manifest file570 metadata = {'X-Object-Manifest': '%s/%s/'571 % (self.container_name, object_name)}572 resp, _ = self.object_client.create_object(self.container_name,573 object_name, data='')574 self.assertHeaders(resp, 'Object', 'PUT')575 resp, _ = self.object_client.update_object_metadata(576 self.container_name, object_name, metadata, metadata_prefix='')577 self.assertHeaders(resp, 'Object', 'POST')578 resp, _ = self.object_client.list_object_metadata(579 self.container_name, object_name)580 # Etag value of a large object is enclosed in double-quotations.581 # After etag quotes are checked they are removed and the response is582 # checked if all common headers are present and well formatted583 self.assertTrue(resp['etag'].startswith('\"'))584 self.assertTrue(resp['etag'].endswith('\"'))585 resp['etag'] = resp['etag'].strip('"')586 self.assertHeaders(resp, 'Object', 'HEAD')587 self.assertIn('x-object-manifest', resp)588 self.assertEqual(resp['x-object-manifest'],589 '%s/%s/' % (self.container_name, object_name))590 # downloading the object591 resp, body = self.object_client.get_object(592 self.container_name, object_name)...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!