KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use __get_api_client method in yandex-tank

Best Python code snippet using yandex-tank

plugin.py

Source:plugin.py

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

gs_adaptor.py

Source:gs_adaptor.py

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

dumper.py

Source:dumper.py

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

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.