KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use wait_for_finish method in yandex-tank
Best Python code snippet using yandex-tank
client.py
Source:client.py 
1import os2import re3import json4import glob5import time6import hashlib7import logging8import urllib.parse9from . import errors10from . import requester11logger = logging.Logger('yadisk-api')12class YandexDisk(object):13 _SLEEP = 314 _requester_cls = requester.Requester15 def __init__(self, token):16 self._requester = self._requester_cls(token=token)17 def get_disk_info(self):18 """19 Get info about your disk20 Docs: https://tech.yandex.ru/disk/api/reference/capacity-docpage/21 """22 logger.info('Get disk info')23 return self._requester.get(url='disk/').json()24 def get_meta_info(25 self,26 path,27 sort=None,28 limit=None,29 offset=None,30 fields=None,31 preview_size=None,32 preview_crop=None,33 trash=False,34 ):35 """36 Get meta info about file/directory from disk or trash37 Docs: https://tech.yandex.ru/disk/api/reference/meta-docpage/38 """39 params = {40 'path': path,41 'sort': sort,42 'limit': limit,43 'offset': offset,44 'fields': fields,45 'preview_size': preview_size,46 'preview_crop': preview_crop,47 }48 logger.info('Get meta info (from trash={}) with params {}'.format(trash, params))49 return self._requester.get(50 url='disk/{}resources'.format('trash/' if trash else ''),51 params=params,52 ).json()53 def get_files_list(54 self,55 limit=None,56 offset=None,57 media_type=None,58 fields=None,59 preview_size=None,60 preview_crop=None,61 ):62 """63 Get files list from disk64 Docs: https://tech.yandex.ru/disk/api/reference/all-files-docpage/65 """66 params = {67 'limit': limit,68 'offset': offset,69 'fields': fields,70 'preview_size': preview_size,71 'preview_crop': preview_crop,72 'media_type': media_type,73 }74 logger.info('Get files list with params {}'.format(params))75 return self._requester.get(url='disk/resources/files', params=params).json()76 def get_last_uploaded(77 self,78 limit=None,79 media_type=None,80 fields=None,81 preview_size=None,82 preview_crop=None,83 ):84 """85 Get last uploaded files list86 Docs: https://tech.yandex.ru/disk/api/reference/recent-upload-docpage/87 """88 params = {89 'limit': limit,90 'fields': fields,91 'preview_size': preview_size,92 'preview_crop': preview_crop,93 'media_type': media_type,94 }95 logger.info('Get last uploaded with params {}'.format(params))96 return self._requester.get(url='disk/resources/files', params=params).json()97 def set_meta_to_resource(self, path, data, fields=None):98 """99 Set meta-data to file/directory100 Docs: https://tech.yandex.ru/disk/api/reference/meta-add-docpage/101 """102 url_params = {'path': path}103 if fields:104 url_params['fields'] = fields105 params_string = urllib.parse.urlencode(url_params, doseq=False)106 logger.info('Set meta {!r} to resource {!r}'.format(data, path))107 return self._requester.patch(108 url='disk/resources/?{}'.format(params_string),109 data=json.dumps({'custom_properties': data}),110 ).json()111 def upload_file(self, file_object, path='/', overwrite=False, skip_exists=True):112 """113 Upload file to yandex disk114 Docs: https://tech.yandex.ru/disk/api/reference/upload-docpage/115 Args:116 file_object (file): file to upload117 path (str): path to file place118 overwrite (bool): overwrite file if it exist119 skip_exists (bool): not upload file if it exists and has same hash/dates120 """121 if (122 overwrite123 and skip_exists124 and self._is_same_file(file_object, path)125 ):126 logger.debug('File {!r} already uploaded'.format(path))127 return True128 logger.info('Upload file to {!r}'.format(path))129 upload_path_url = self._requester.get(130 url='disk/resources/upload',131 params={132 'path': path,133 'overwrite': overwrite,134 },135 )136 self._requester.put(137 url=upload_path_url.json()['href'],138 files={'file': file_object},139 absolute_url=True,140 )141 return True142 def upload_directory(self, local_path, path='/', overwrite=False, skip_exists=True):143 """144 Upload all files in directory to disk145 Args:146 local_path (str): directory path on your storage147 path (str): path at yadisk148 overwrite (bool): overwrite files if it exists149 skip_exists (bool): not upload file if it exists and has same hash/dates150 """151 for is_directory, item_path, related_path in self._iter_directory_content(local_path):152 if not related_path:153 continue154 disk_path = os.path.join(path, related_path)155 if is_directory:156 try:157 self.create_folder(path=disk_path)158 except errors.DiskPathError as exc:159 if 'Ñже ÑÑÑеÑÑвÑÐµÑ Ð¿Ð°Ð¿ÐºÐ° Ñ Ñаким именем' not in str(exc):160 raise161 continue162 with open(item_path, 'rb') as f:163 self.upload_file(164 file_object=f,165 path=disk_path,166 overwrite=overwrite,167 skip_exists=skip_exists,168 )169 def upload_file_from_url(170 self,171 url,172 path,173 fields=None,174 disable_redirects=False,175 wait_for_finish=True,176 sleep=None,177 ):178 """179 Upload file from url to yandex disk180 Docs: https://tech.yandex.ru/disk/api/reference/upload-ext-docpage/181 Args:182 url (str): url to download file from it183 path (str): path to yandex disk184 fields (list[str]|None): fields in result185 disable_redirects (bool): disable redirects186 wait_for_finish (bool): wait for operation finish187 sleep (int): sleep time in seconds if need wait to finish188 """189 params_string = urllib.parse.urlencode(190 {191 'url': url,192 'path': path,193 'fields': fields,194 'disable_redirects': disable_redirects,195 },196 doseq=False,197 )198 logger.info('Upload file from url {!r} to {!r}'.format(url, path))199 return self._waiting_for_finish(200 self._requester.post(201 url='disk/resources/upload?{}'.format(params_string),202 ),203 wait_for_finish=wait_for_finish,204 sleep=sleep,205 ).json()206 def download_file(self, path, stream=False):207 """208 Download file from your disk209 Docs: https://tech.yandex.ru/disk/api/reference/content-docpage/210 Args:211 path (str): path to file212 stream (bool): stream response213 Returns:214 bytes: file content215 """216 logger.info('Download file from {!r}'.format(path))217 url_response = self._requester.get(218 url='disk/resources/download',219 params={'path': path}220 )221 return self._requester.get(222 url=url_response.json()['href'],223 absolute_url=True,224 stream=stream,225 ).content226 def copy_resource(227 self,228 from_path,229 to_path,230 overwrite=False,231 fields=None,232 wait_for_finish=True,233 sleep=None,234 ):235 """236 Copy file/directory in disk237 Docs: https://tech.yandex.ru/disk/api/reference/copy-docpage/238 Args:239 from_path (str): source path240 to_path (str): destination path241 overwrite (bool): overwrite file/directory if exists242 fields (list[str]|None): response fields list243 wait_for_finish (bool): wait for operation finish244 sleep (int): sleep time in seconds if need wait to finish245 """246 logger.info('Copy resource from {!r} to {!r}'.format(from_path, to_path))247 params_string = urllib.parse.urlencode(248 {249 'from': from_path,250 'path': to_path,251 'overwrite': overwrite,252 'fields': fields,253 },254 doseq=False,255 )256 return self._waiting_for_finish(257 self._requester.post(url='disk/resources/copy?{}'.format(params_string)),258 wait_for_finish=wait_for_finish,259 sleep=sleep,260 ).json()261 def move_resource(262 self,263 from_path,264 to_path,265 overwrite=False,266 wait_for_finish=True,267 sleep=None,268 ):269 """270 Move file/directory in disk271 Docs: https://tech.yandex.ru/disk/api/reference/move-docpage/272 Args:273 from_path (str): source path274 to_path (str): destination path275 overwrite (bool): overwrite file/directory if exists276 wait_for_finish (bool): wait for operation finish277 sleep (int): sleep time in seconds if need wait to finish278 """279 logger.info('Move resource from {!r} to {!r}'.format(from_path, to_path))280 params_string = urllib.parse.urlencode(281 {282 'from': from_path,283 'path': to_path,284 'overwrite': overwrite,285 },286 doseq=False,287 )288 return self._waiting_for_finish(289 self._requester.post(url='disk/resources/move?{}'.format(params_string)),290 wait_for_finish=wait_for_finish,291 sleep=sleep,292 ).json()293 def delete_resource(self, path, permanently=False, wait_for_finish=True, sleep=None):294 """295 Remove file/directory to trash or at all296 Docs: https://tech.yandex.ru/disk/api/reference/delete-docpage/297 Args:298 path (str): path to file/directory299 permanently (bool): true if your want remove resource without trash300 wait_for_finish (bool): wait for operation finish301 sleep (int): sleep time in seconds if need wait to finish302 """303 logger.info('Delete resource from {!r}'.format(path))304 params_string = urllib.parse.urlencode(305 {306 'path': path,307 'permanently': permanently,308 },309 doseq=False,310 )311 self._waiting_for_finish(312 self._requester.delete(313 url='disk/resources?{}'.format(params_string)314 ),315 wait_for_finish=wait_for_finish,316 sleep=sleep,317 )318 return True319 def create_folder(self, path, fields=None):320 """321 Create folder in your disk322 Docs: https://tech.yandex.ru/disk/api/reference/create-folder-docpage/323 """324 logger.info('Create folder {!r}'.format(path))325 params_string = urllib.parse.urlencode(326 {327 'path': path,328 'fields': fields,329 },330 doseq=False,331 )332 return self._requester.put(url='disk/resources/?{}'.format(params_string)).json()333 def publish_resource(self, path):334 """335 Publish_resource336 Docs: https://tech.yandex.ru/disk/api/reference/publish-docpage/337 Args:338 path (str): path to resource339 """340 logger.info('Publish resource {!r}'.format(path))341 params_string = urllib.parse.urlencode({'path': path})342 return self._requester.put(343 url='disk/resources/publish?{}'.format(params_string)344 ).json()345 def unpublish_resource(self, path):346 """347 Unpublish resource348 Docs: https://tech.yandex.ru/disk/api/reference/publish-docpage/#unpublish-q349 Args:350 path (str): path to resource351 """352 logger.info('Unpublish resource {!r}'.format(path))353 params_string = urllib.parse.urlencode({'path': path})354 return self._requester.put(355 url='disk/resources/unpublish?{}'.format(params_string)356 ).json()357 def empty_trash(self, path=None, wait_for_finish=True, sleep=None):358 """359 Empty trash or delete resource from trash360 Docs: https://tech.yandex.ru/disk/api/reference/trash-delete-docpage/361 """362 logger.info('Empty trash')363 path_param = '?{}'.format(364 urllib.parse.urlencode({'path': path})365 ) if path else ''366 self._waiting_for_finish(367 self._requester.delete(url='disk/trash/resources/{}'.format(path_param)),368 wait_for_finish=wait_for_finish,369 sleep=sleep,370 )371 return True372 def restore_from_trash(self, path, name=None, overwrite=False):373 """374 Restore resource from trash375 Docs: https://tech.yandex.ru/disk/api/reference/trash-restore-docpage/376 """377 logger.info('Restore {!r} from trash'.format(path))378 params_string = urllib.parse.urlencode(379 {380 'path': path,381 'name': name,382 'overwrite': overwrite,383 },384 doseq=False,385 )386 return self._waiting_for_finish(387 self._requester.put(388 url='disk/trash/resources/restore?{}'.format(params_string),389 )390 ).json()391 def _waiting_for_finish(self, response, wait_for_finish=True, sleep=None):392 """393 Waiting for finish operation, if you want394 Args:395 response (requests.Response): response object396 wait_for_finish (bool): wait for operation finish397 sleep (int): sleep time in seconds if need wait to finish398 Returns:399 requests.Response400 """401 if (402 wait_for_finish403 and response.status_code == requester.STATUS_ACCEPTED404 ):405 check_status_url = response.json()['href']406 sleep = sleep or self._SLEEP407 while True:408 time.sleep(sleep)409 response = self._requester.get(check_status_url, absolute_url=True)410 if (411 response.status_code == requester.STATUS_OK412 and response.json()['status'] == 'success'413 ):414 return response415 return response416 def _iter_directory_content(self, path, start_path=None):417 """418 All files generator419 Args:420 path (str): path to find all files421 Yields:422 bool, str, str:423 1. is directory flag424 2. local path to file/dir425 3. related path to file/dir426 """427 if start_path is None:428 start_path = path429 path_list = glob.glob(path)430 for path_item in path_list:431 related_path = re.sub(r'{}\/?\*?'.format(start_path), '', path_item)432 is_directory = os.path.isdir(path_item)433 yield is_directory, path_item, related_path434 yield from self._iter_directory_content(os.path.join(path_item, '*'), start_path=start_path)435 def _is_same_file(self, fileobject, path):436 """437 Exists same file438 """439 # check file changed440 try:441 file_meta_info = self.get_meta_info(path=path, fields=['md5'])442 except errors.NotFoundError:443 return False444 current_file_md5 = hashlib.md5(fileobject.read()).hexdigest()...
socrata.py
Source:socrata.py
...103 output_schema = add_geometry_to_output_schema(output_schema, geometry)104 # Handle pre-1.x versions of Socrata-py105 if isinstance(output_schema, tuple):106 _, output_schema = output_schema107 output_schema.wait_for_finish()108 job: Job = revision.apply(output_schema=output_schema)109 if wait_for_finish is True:110 job.wait_for_finish()111 return revision112def update_existing_dataset(113 client: Socrata, dataframe: DataFrame, dataset_id: str, *, wait_for_finish: bool = False114):115 """Use a dataframe to update an existing Socrata dataset."""116 view = client.views.lookup(dataset_id)117 # Handle pre-1.x versions of Socrata-py118 if isinstance(view, tuple):119 ok, view = view120 ok, revision = view.revisions.create_replace_revision()121 ok, upload = revision.create_upload('autocensus-update')122 ok, source = upload.df(dataframe)123 source.wait_for_finish()124 output = source.get_latest_input_schema().get_latest_output_schema()125 output.wait_for_finish()126 job = revision.apply(output_schema=output)127 else:128 revision = view.revisions.create_replace_revision()129 upload = revision.create_upload('autocensus-update')130 source = upload.df(dataframe)131 source.wait_for_finish()132 output = source.get_latest_input_schema().get_latest_output_schema()133 output.wait_for_finish()134 job = revision.apply(output_schema=output)135 if wait_for_finish is True:136 job.wait_for_finish()137 return revision138def build_dataset_name(estimate: int, years: Iterable) -> str:139 """Produce a nicely formatted dataset name."""140 unique_years = sorted(set(years))141 if len(unique_years) > 1:142 years_range = f'{min(unique_years)}â{max(unique_years)}'143 else:144 years_range = unique_years[0]145 query_name = f'American Community Survey {estimate}-Year Estimates, {years_range}'146 return query_name147def to_socrata(148 domain: Union[URL, str],149 dataframe: DataFrame,150 dataset_id: Optional[str] = None,...
pipette_foreach.py
Source:pipette_foreach.py
...23 msg = self.serial_connection.serial_port.readline()24 if msg:25 print("Got response.", msg)26 return msg27 def wait_for_finish(self):28 query_cmd = "/1Q\r"29 i = 130 while True:31 self.serial_connection.send_commond_string_foreach(query_cmd)32 time.sleep(WAIT_TIME)33 res = self.serial_connection.serial_port.readline()34 if not (b'/0@\x03' in res):35 print(i, res)36 i = i+137 if i > 300:38 print("Pipette no response for 30 seconds, system exited")39 # sys.exit()40 if b'/0`' in res:41 print("Finish request.")42 return "finish"43 def send_commond_string_foreach(self, cmd):44 self.wait_for_finish()45 self.serial_connection.send_commond_string_foreach(cmd)46 def initialization(self):47 init_cmd = '/1ZR\r'48 self.send_commond_string_foreach(init_cmd)49 time.sleep(1)50 self.wait_for_response()51 self.wait_for_finish()52 print("initialization finished.")53 normal_mode = "/1N0R\r"54 # N0 mode and N1 mode55 self.send_commond_string_foreach(normal_mode)56 self.wait_for_response()57 self.wait_for_finish()58 def increase_range(self):59 cmd = '/1u1,3500R\r'60 self.send_commond_string_foreach(cmd)61 self.wait_for_response()62 self.wait_for_finish()63 def aspirate(self, volume):64 '''aspirate volume = ? uL'''65 mL_per_step = 0.31966 steps = int(volume/mL_per_step)67 aspirate_cmd = "/1P" + str(steps) + "R\r"68 self.send_commond_string_foreach(aspirate_cmd)69 self.wait_for_response()70 self.wait_for_finish()71 def send_drop_tip_cmd(self):72 eject_tip = "/1ER\r"73 self.send_commond_string_foreach(eject_tip)74 time.sleep(0.5)75 self.wait_for_response()76 self.wait_for_finish()77 def dispense(self, volume=0): # volume in uL78 dispense_all = "/1A0R\r"79 self.send_commond_string_foreach(dispense_all)80 self.wait_for_response()81 self.wait_for_finish()82 def set_speed(self, speed):83 '''speed = 0-40, default = 11, bigger nmuber is slower '''84 cmd = "/1S" + str(speed) + "R\r"85 self.send_commond_string_foreach(cmd)86 self.wait_for_response()87 self.wait_for_finish()88 def is_tip_attached(self):89 check_tip_cmd = "/1?31R\r"90 self.send_commond_string_foreach(check_tip_cmd)91 res = self.wait_for_response()92 self.wait_for_finish()93 if b"`1\x03" in res:94 print("tip attached! code:", res)95 return True96 else:97 print("tip NOT attached! code:", res)98 return False99 def send_pickup_tip_cmd(self): # do nothing, not needed for foreach model100 pass101 def set_transport_air_volume(self, volume=20): # volume in uL102 if volume > 0:...
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 yandex-tank 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?
