How to use __get_api_client method in yandex-tank

Best Python code snippet using yandex-tank

plugin.py

Source: plugin.py Github

copy

Full Screen

...461 except:462 logger.error(463 "Couldn't get username from the OS. Please, set the 'meta.operator' option explicitly in your config file.")464 raise465 def __get_api_client(self):466 if self.backend_type == BackendTypes.LUNAPARK:467 client = APIClient468 self._api_token = None469 elif self.backend_type == BackendTypes.OVERLOAD:470 client = OverloadClient471 self._api_token = self.read_token(self.get_option("token_file", ""))472 else:473 raise RuntimeError("Backend type doesn't match any of the expected")474 return client(base_url=self.get_option('api_address'),475 writer_url=self.get_option('writer_endpoint', ""),476 network_attempts=int(self.get_option('network_attempts', 60)),477 api_attempts=int(self.get_option('api_attempts', 60)),478 maintenance_attempts=int(self.get_option('maintenance_attempts', 10)),479 network_timeout=int(self.get_option('network_timeout', 10)),480 api_timeout=int(self.get_option('api_timeout', 10)),481 maintenance_timeout=int(self.get_option('maintenance_timeout', 60)),482 connection_timeout=int(self.get_option('connection_timeout', 30)),483 user_agent=self._get_user_agent(),484 api_token=self.api_token)485 @property486 def lp_job(self):487 if self._lp_job is None:488 self._lp_job = self.__get_lp_job()489 return self._lp_job490 def __get_lp_job(self):491 api_client = self.__get_api_client()492 info = self.generator_info493 port = info.port494 loadscheme = [] if isinstance(info.rps_schedule,495 str) else info.rps_schedule496 return LPJob(client=api_client,497 target_host=self.target,498 target_port=port,499 number=self.get_option('jobno', ''),500 token=self.get_option('upload_token', ''),501 person=self.__get_operator(),502 task=self.task,503 name=self.get_option('job_name', 'none').decode('utf8'),504 description=self.get_option('job_dsc', '').decode('utf8'),505 tank=self.core.job.tank,...

Full Screen

Full Screen

gs_adaptor.py

Source: gs_adaptor.py Github

copy

Full Screen

...70 "projectId":GS_PROJECT_ID71 } 72 logger.debug(str(request_dict)) 73 try:74 gs = self.__get_api_client()[0]75 gs.buckets().insert(body=request_dict).execute()76 except:77 logger.debug("Error creating bucket: " + self.bucket_name)78 pass # Do nothing if bucket already exists79 80 81 def get_pilotdata_size(self):82 # unlimited size83 return None84 85 86 def delete_pilotdata(self):87 self.__state=State.Done88 89 90 def get_state(self):91 return self.__state92 93 94 def create_du(self, du_id):95 gs = self.__get_api_client()[0]96 o = gs.objects().insert(bucket=self.bucket_name, name=str(du_id)+"/​du_info",97 body={'media': {98 "contentType":"text/​ascii",99 "data": du_id 100 } 101 } 102 ).execute()103 logger.debug("Created GS: " + str(o))104 105 106 def put_du(self, du):107 logger.debug("Copy DU to Google Storage")108 du_items = du.list()109 for i in du_items.keys(): 110 try:111 local_filename=du_items[i]["local"]112 remote_path = os.path.join(str(du.id), os.path.basename(local_filename))113 self._put_file(local_filename, remote_path)114 except:115 logger.debug("Could not copy file: " + (str(i)))116 117 118 def copy_du(self, du, pd_new):119 pass120 121 122 def get_du(self, du, target_url):123 du_id=du.id124 logger.debug("Get DU: " + str(du_id))125 gs = self.__get_api_client()[0]126 result = gs.objects().list(bucket=self.bucket_name, prefix=du_id).execute() 127 #delimiter="/​",128 #prefix=[du_id]).execute() 129 logger.debug("Result: " + str(result))130 for i in result["items"]:131 full_filename = i["name"]132 self._get_file(full_filename, os.path.join(target_url, os.path.basename(full_filename)))133 134 135 def remove_du(self, du):136 self.__remove_directory(os.path.join(self.bucket_name, du.id))137 138 139 ###########################################################################140 # Pure File Management APIs141 def _put_file(self, source, target):142 logger.debug("Put file: %s to %s"%(source, target))143 gs = self.__get_api_client()[0]144 #media = MediaFileUpload(source, 145 # resumable=False)146 o = gs.objects().insert(bucket=self.bucket_name, 147 name=target,148 media_body=source).execute() 149 logger.debug("Put file result: %s"%str(o))150 151 152 def _get_file(self, source, target):153 logger.debug("GET file: %s to %s"%(source, target))154 gs, http = self.__get_api_client()155 f = gs.objects().get(bucket=self.bucket_name, 156 object=source).execute() 157 logger.debug("Get file result: %s"%str(f))158 downloadUrl = f["media"]['link']159 if downloadUrl:160 response, content = http.request(downloadUrl)161 logger.debug("Download file response: %d"%(response.status))162 with open(target, 'wb') as f:163 f.write(content)164 165 166 def transfer(self, source_url, target_url):167 pass168 169 def create_remote_directory(self, target_url):170 return True171 172 173 def get_path(self, target_url):174 result = urlparse.urlparse(target_url)175 target_path = result.path176 return target_path177 178 179 ###########################################################################180 # Auxiliary functions181 def __get_api_client(self):182 http = httplib2.Http()183 http = self.credentials.authorize(http)184 gs = build("storage", "v1beta1", http=http)185 return gs, http186 187 188 def __get_bucket_name(self, service_url):189 bucket_name = service_url.replace("gs:/​/​", "")190 bucket_name = bucket_name.replace("/​", "")191 return bucket_name192 193 194 def __print_traceback(self):195 exc_type, exc_value, exc_traceback = sys.exc_info()...

Full Screen

Full Screen

dumper.py

Source: dumper.py Github

copy

Full Screen

...4from src.base import BaseObject5class Santinel2Dumper(BaseObject):6 def __init__(self) -> None:7 super().__init__()8 self.api_client = self.__get_api_client()9 def __get_api_client(self):10 return SentinelAPI(11 AUTH["USERNAME"], AUTH["PASSWORD"], "https:/​/​scihub.copernicus.eu/​dhus"12 )13 def get_products_from_query(14 self,15 footprint,16 date_range,17 processinglevel="Level-2A",18 cloudcoverpercentage=(0, 10),19 ):20 """21 Function used for searching available data for a footprint22 :param footprint: A Polygon that contains the targeted area23 :type footprint: shapely.geometry.Polygon...

Full Screen

Full Screen

Blogs

Check out the latest blogs from LambdaTest on this topic:

Webinar: Move Forward With An Effective Test Automation Strategy [Voices of Community]

The key to successful test automation is to focus on tasks that maximize the return on investment (ROI), ensuring that you are automating the right tests and automating them in the right way. This is where test automation strategies come into play.

Dec’22 Updates: The All-New LT Browser 2.0, XCUI App Automation with HyperExecute, And More!

Greetings folks! With the new year finally upon us, we’re excited to announce a collection of brand-new product updates. At LambdaTest, we strive to provide you with a comprehensive test orchestration and execution platform to ensure the ultimate web and mobile experience.

Developers and Bugs – why are they happening again and again?

Entering the world of testers, one question started to formulate in my mind: “what is the reason that bugs happen?”.

Test Optimization for Continuous Integration

“Test frequently and early.” If you’ve been following my testing agenda, you’re probably sick of hearing me repeat that. However, it is making sense that if your tests detect an issue soon after it occurs, it will be easier to resolve. This is one of the guiding concepts that makes continuous integration such an effective method. I’ve encountered several teams who have a lot of automated tests but don’t use them as part of a continuous integration approach. There are frequently various reasons why the team believes these tests cannot be used with continuous integration. Perhaps the tests take too long to run, or they are not dependable enough to provide correct results on their own, necessitating human interpretation.

13 Best Java Testing Frameworks For 2023

The fact is not alien to us anymore that cross browser testing is imperative to enhance your application’s user experience. Enhanced knowledge of popular and highly acclaimed testing frameworks goes a long way in developing a new app. It holds more significance if you are a full-stack developer or expert programmer.

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 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful