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

How to use convert_to_proto_message method in yandex-tank

Best Python code snippet using yandex-tank

client.py

Source:client.py

...703        intervals = []704        for interval in intervals_data:705            intervals.append(trail_service_pb2.Trail.Intervals(to=interval['to'], count=int(interval['count'])))706        return intervals707    def convert_to_proto_message(self, items):708        trails = []709        for item in items:710            trail_data = item["trail"]711            trail = trail_service_pb2.Trail(712                overall=int(item["overall"]),713                case_id=item["case"],714                time=trail_data["time"],715                reqps=int(trail_data["reqps"]),716                resps=int(trail_data["resps"]),717                expect=trail_data["expect"],718                input=int(trail_data["input"]),719                output=int(trail_data["output"]),720                connect_time=trail_data["connect_time"],721                send_time=trail_data["send_time"],722                latency=trail_data["latency"],723                receive_time=trail_data["receive_time"],724                threads=int(trail_data["threads"]),725                q50=trail_data.get('q50'),726                q75=trail_data.get('q75'),727                q80=trail_data.get('q80'),728                q85=trail_data.get('q85'),729                q90=trail_data.get('q90'),730                q95=trail_data.get('q95'),731                q98=trail_data.get('q98'),732                q99=trail_data.get('99'),733                q100=trail_data.get('q100'),734                http_codes=self.build_codes(item['http_codes']),735                net_codes=self.build_codes(item['net_codes']),736                time_intervals=self.build_intervals(item['time_intervals']),737            )738            trails.append(trail)739        return trails740    def send_trails(self, instance_id, cloud_job_id, trails):741        try:742            request = trail_service_pb2.CreateTrailRequest(743                compute_instance_id=str(instance_id),744                job_id=str(cloud_job_id),745                data=trails746            )747            result = self.trail_stub.Create(748                request,749                timeout=self.connection_timeout,750                metadata=[('authorization', f'Bearer {self.token}')]751                # credentials=self._get_call_creds(self.token)752            )753            logger.debug(f'Send trails: {trails}')754            return result.code755        except grpc.RpcError as err:756            if err.code() in (grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED):757                raise self.NotAvailable('Connection is closed. Try to set it again.')758            raise err759        except Exception as err:760            raise err761    def push_test_data(762            self,763            cloud_job_id,764            data_item,765            stat_item,766            interrupted_event):767        items = []768        ts = data_item["ts"]769        for case_name, case_data in data_item["tagged"].items():770            if case_name == "":771                case_name = "__NOTAG__"772            push_item = self.second_data_to_push_item(case_data, stat_item, ts,773                                                      0, case_name)774            items.append(push_item)775        overall = self.second_data_to_push_item(data_item["overall"],776                                                stat_item, ts, 1, '')777        items.append(overall)778        api_timeouts = self.api_timeouts()779        while not interrupted_event.is_set():780            try:781                code = self.send_trails(self.compute_instance_id, cloud_job_id, self.convert_to_proto_message(items))782                if code == 0:783                    break784            except self.NotAvailable as err:785                if not self.core_interrupted.is_set():786                    try:787                        timeout = next(api_timeouts)788                    except StopIteration:789                        raise err790                    self._set_connection(self.token)791                    logger.warn("GRPC error, will retry in %ss...", timeout)792                    time.sleep(timeout)793                    continue794                else:795                    break...

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.