Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use second_data_to_push_item method in yandex-tank

Best Python code snippet using yandex-tank

client.py

Source:client.py

...335            res = self.get_job_summary(jobno)336            data['description'] = (res['dsc'] + "\n" + comment).strip()337        response = self.__post('api/job/' + str(jobno) + '/edit.json', data)338        return response339    def second_data_to_push_item(self, data, stat, timestamp, overall, case):340        """341        @data: SecondAggregateDataItem342        """343        api_data = {344            'overall': overall,345            'case': case,346            'net_codes': [],347            'http_codes': [],348            'time_intervals': [],349            'trail': {350                'time': str(timestamp),351                'reqps': stat["metrics"]["reqps"],352                'resps': data["interval_real"]["len"],353                'expect': data["interval_real"]["total"] / 1000.0 /354                data["interval_real"]["len"],355                'disper': 0,356                'self_load':357                    0,  # TODO abs(round(100 - float(data.selfload), 2)),358                'input': data["size_in"]["total"],359                'output': data["size_out"]["total"],360                'connect_time': data["connect_time"]["total"] / 1000.0 /361                data["connect_time"]["len"],362                'send_time':363                    data["send_time"]["total"] / \364                1000.0 / data["send_time"]["len"],365                'latency':366                    data["latency"]["total"] / 1000.0 / data["latency"]["len"],367                'receive_time': data["receive_time"]["total"] / 1000.0 /368                data["receive_time"]["len"],369                'threads': stat["metrics"]["instances"],  # TODO370            }371        }372        for q, value in zip(data["interval_real"]["q"]["q"],373                            data["interval_real"]["q"]["value"]):374            api_data['trail']['q' + str(q)] = value / 1000.0375        for code, cnt in data["net_code"]["count"].items():376            api_data['net_codes'].append({'code': int(code),377                                          'count': int(cnt)})378        for code, cnt in data["proto_code"]["count"].items():379            api_data['http_codes'].append({'code': int(code),380                                           'count': int(cnt)})381        api_data['time_intervals'] = self.convert_hist(data["interval_real"][382            "hist"])383        return api_data384    def convert_hist(self, hist):385        data = hist['data']386        bins = hist['bins']387        return [388            {389                "from": 0,  # deprecated390                "to": b / 1000.0,391                "count": count,392            } for b, count in zip(bins, data)393        ]394    def push_test_data(395            self,396            jobno,397            upload_token,398            data_item,399            stat_item,400            trace=False):401        items = []402        uri = 'api/job/{0}/push_data.json?upload_token={1}'.format(403            jobno, upload_token)404        ts = data_item["ts"]405        for case_name, case_data in data_item["tagged"].items():406            if case_name == "":407                case_name = "__NOTAG__"408            if (len(case_name)) > 128:409                raise RuntimeError('tag (case) name is too long: ' + case_name)410            push_item = self.second_data_to_push_item(case_data, stat_item, ts,411                                                      0, case_name)412            items.append(push_item)413        overall = self.second_data_to_push_item(data_item["overall"],414                                                stat_item, ts, 1, '')415        items.append(overall)416        api_timeouts = self.api_timeouts()417        while True:418            try:419                if self.writer_url:420                    res = self.__make_writer_request(421                        params={422                            "jobno": jobno,423                            "upload_token": upload_token,424                        },425                        json={426                            "trail": items,427                        },...

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.