KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _upload_segments method in tempest
Best Python code snippet using tempest_python
test_object_services.py
Source:test_object_services.py 
...38 data = data_utils.arbitrary_string()39 self.object_client.create_object(self.container_name,40 object_name, data, metadata=metadata)41 return object_name, data42 def _upload_segments(self):43 # create object44 object_name = data_utils.rand_name(name='LObject')45 data = data_utils.arbitrary_string()46 segments = 1047 data_segments = [data + str(i) for i in moves.xrange(segments)]48 # uploading segments49 for i in moves.xrange(segments):50 resp, _ = self.object_client.create_object_segments(51 self.container_name, object_name, i, data_segments[i])52 self.assertEqual(resp['status'], '201')53 return object_name, data_segments54 @test.attr(type='smoke')55 def test_create_object(self):56 # create object57 object_name = data_utils.rand_name(name='TestObject')58 data = data_utils.arbitrary_string()59 resp, _ = self.object_client.create_object(self.container_name,60 object_name, data)61 # create another object62 object_name = data_utils.rand_name(name='TestObject')63 data = data_utils.arbitrary_string()64 resp, _ = self.object_client.create_object(self.container_name,65 object_name, data)66 self.assertEqual(resp['status'], '201')67 self.assertHeaders(resp, 'Object', 'PUT')68 @test.attr(type='smoke')69 def test_delete_object(self):70 # create object71 object_name = data_utils.rand_name(name='TestObject')72 data = data_utils.arbitrary_string()73 resp, _ = self.object_client.create_object(self.container_name,74 object_name, data)75 # delete object76 resp, _ = self.object_client.delete_object(self.container_name,77 object_name)78 self.assertIn(int(resp['status']), test.HTTP_SUCCESS)79 self.assertHeaders(resp, 'Object', 'DELETE')80 @test.attr(type='smoke')81 def test_update_object_metadata(self):82 # update object metadata83 object_name, data = self._create_object()84 metadata = {'X-Object-Meta-test-meta': 'Meta'}85 resp, _ = self.object_client.update_object_metadata(86 self.container_name,87 object_name,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)...
load_ds.py
Source:load_ds.py
...52 for sp_idx, start, end in zip(imzml_reader.pixel_indexes, sp_lens[:-1], sp_lens[1:]):53 sp_idxs[start:end] = sp_idx54 sp_idxs = sp_idxs[by_mz]55 return mzs, ints, sp_idxs56def _upload_segments(storage, ds_segm_size_mb, imzml_reader, mzs, ints, sp_idxs):57 # Split into segments no larger than ds_segm_size_mb58 total_n_mz = len(sp_idxs)59 row_size = (4 if imzml_reader.mz_precision == 'f' else 8) + 4 + 460 segm_n = int(np.ceil(total_n_mz * row_size / (ds_segm_size_mb * 2 ** 20)))61 segm_bounds = np.linspace(0, total_n_mz, segm_n + 1, dtype=np.int64)62 segm_ranges = list(zip(segm_bounds[:-1], segm_bounds[1:]))63 ds_segm_lens = np.diff(segm_bounds)64 ds_segments_bounds = np.column_stack([mzs[segm_bounds[:-1]], mzs[segm_bounds[1:] - 1]])65 def upload_segm(start_end):66 start, end = start_end67 df = pd.DataFrame(68 {'mz': mzs[start:end], 'int': ints[start:end], 'sp_i': sp_idxs[start:end]},69 index=pd.RangeIndex(start, end),70 )71 return save_cobj(storage, df)72 with ThreadPoolExecutor(4) as executor:73 ds_segms_cobjs = list(executor.map(upload_segm, segm_ranges))74 return ds_segms_cobjs, ds_segments_bounds, ds_segm_lens75def _upload_imzml_browser_files_to_cos(storage, imzml_cobject, imzml_reader, mzs, ints, sp_idxs):76 """Save imzml browser files on COS"""77 def upload_file(data, key):78 return storage.put_cloudobject(data.astype('f').tobytes(), key=key)79 prefix = f'imzml_browser/{imzml_cobject.key.split("/")[1]}'80 keys = [f'{prefix}/{var}.npy' for var in ['mzs', 'ints', 'sp_idxs']]81 with ThreadPoolExecutor(3) as executor:82 cobjs = list(executor.map(upload_file, [mzs, ints, sp_idxs], keys))83 chunk_records_number = 102484 data = mzs[::chunk_records_number].astype('f')85 cobjs.append(storage.put_cloudobject(data.tobytes(), key=f'{prefix}/mz_index.npy'))86 data = imzml_reader.imzml_reader87 cobjs.append(save_cobj(storage, data, key=f'{prefix}/portable_spectrum_reader.pickle'))88 return cobjs89def _load_ds(90 imzml_cobject: CloudObject,91 ibd_cobject: CloudObject,92 ds_segm_size_mb: int,93 *,94 storage: Storage,95 perf: SubtaskProfiler,96) -> Tuple[LithopsImzMLReader, np.ndarray, List[CObj[pd.DataFrame]], np.ndarray, List[CObj]]:97 logger.info('Loading .imzML file...')98 imzml_reader = LithopsImzMLReader(storage, imzml_cobject, ibd_cobject)99 perf.record_entry(100 'loaded imzml',101 n_peaks=np.sum(imzml_reader.imzml_reader.intensityLengths),102 mz_dtype=imzml_reader.imzml_reader.mzPrecision,103 int_dtype=imzml_reader.imzml_reader.intensityPrecision,104 )105 logger.info('Reading spectra')106 mzs, ints, sp_lens = _load_spectra(storage, imzml_reader)107 perf.record_entry('read spectra', n_peaks=len(mzs))108 logger.info('Sorting spectra')109 mzs, ints, sp_idxs = _sort_spectra(imzml_reader, mzs, ints, sp_lens)110 perf.record_entry('sorted spectra')111 logger.info('Uploading segments')112 ds_segms_cobjs, ds_segments_bounds, ds_segm_lens = _upload_segments(113 storage, ds_segm_size_mb, imzml_reader, mzs, ints, sp_idxs114 )115 perf.record_entry('uploaded segments', n_segms=len(ds_segms_cobjs))116 logger.info('Uploading related to imzml browser files')117 imzml_browser_cobjs = _upload_imzml_browser_files_to_cos(118 storage, imzml_cobject, imzml_reader, mzs, ints, sp_idxs119 )120 perf.record_entry('uploaded imzml browser files')121 return imzml_reader, ds_segments_bounds, ds_segms_cobjs, ds_segm_lens, imzml_browser_cobjs122def _upload_imzml_browser_files(storage: Storage, imzml_browser_cobjs: List[CObj]):123 """Move files from COS to S3"""124 conf = SMConfig.get_conf()125 s3_client = get_s3_client(sm_config=conf)126 transfer_config = TransferConfig(...
recorder.py
Source:recorder.py
...213 except RuntimeError: # Closing time214 pass215 else:216 upload.add_done_callback(_upload_complete)217 def _upload_segments(self, segments: list[_Segment]) -> str:218 files = [self._workdir_path / segment.file for segment in segments]219 try:220 ipfs_files = self._ipfs.add(221 *files, trickle=True, wrap_with_directory=True, cid_version=1222 )223 finally:224 for file in files:225 os.remove(file)226 ipfs_dir = next(file for file in ipfs_files if not file["Name"])227 for segment in segments:228 url = self._ipfs_url(ipfs_dir["Hash"], path=segment.file)229 self._playlist.append(url, segment.duration)230 m3u8 = self._workdir_path / f"{self._streamer.name}.m3u8"231 self._playlist.write(m3u8)...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
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!!
Was this article helpful?
