简单的性能自动化测试架构设计和实现(pylot)-python

发表于:2013-5-17 14:59

字体: | 上一篇 | 下一篇 | 我要投稿

 作者:waterxi(cnblogs)    来源:51Testing软件测试网采编

  根据用户的策略,会对testcase.xml中的case项进行分割和组合,最终得到每个agent对应的testcase.xml,agent会按照这个testcase.xml进行测试,通常来说,策略包含以下两种:a. 每个agent执行所有case;b. 将所有case平均分给agent执行。

  重要代码:

def load_xml_cases_dom(dom):
     cases = []
     param_map = {}
     for child in dom.getiterator():
         if child.tag != dom.getroot().tag and child.tag == 'param':
             name = child.attrib.get('name')
             value = child.attrib.get('value')
             param_map[name] = value
         if child.tag != dom.getroot().tag and child.tag == 'case':
             req = Request()
             repeat = child.attrib.get('repeat')
             if repeat:
                 req.repeat = int(repeat)
             else:
                 req.repeat = 1
             for element in child:
                 if element.tag.lower() == 'url':
                     req.url = element.text
                 if element.tag.lower() == 'method':
                     req.method = element.text
                 if element.tag.lower() == 'body':
                     file_payload = element.attrib.get('file')
                     if file_payload:
                         req.body = open(file_payload, 'rb').read()
                     else:
                         req.body = element.text
                 if element.tag.lower() == 'verify':
                     req.verify = element.text
                 if element.tag.lower() == 'verify_negative':
                     req.verify_negative = element.text
                 if element.tag.lower() == 'timer_group':
                     req.timer_group = element.text
                 if element.tag.lower() == 'add_header':
                     splat = element.text.split(':')
                     x = splat[0].strip()
                     del splat[0]
                     req.add_header(x, ''.join(splat).strip())
             req = resolve_parameters(req, param_map)  # substitute vars
             cases.append(req)
     return case

  2、Request请求代理部分:

  请求代理会根据分配给自己的testcase.xml进行初始化,获取所有配置参数和case列表,当接收到agent manager发给的执行消息后,会开启一个线程对case列表中的每一个case进行处理;初始化agent时,agent manager会传进来全局的几个队列,包括result queue,msg queue,error queue;这些queue中的信息最终会由agent manager统一处理;

  重要代码:

    def run(self):
         agent_start_time = time.strftime('%H:%M:%S', time.localtime())
         total_latency = 0
         total_connect_latency = 0
         total_bytes = 0
 
         while self.running:
             self.cookie_jar = cookielib.CookieJar()
             for req in self.msg_queue:
                 for repeat in range(req.repeat):
                     if self.running:
          request=Request(req)
                         # send the request message
                         resp, content, req_start_time, req_end_time, connect_end_time = request.send(req)

                         # get times for logging and error display
                         tmp_time = time.localtime()
                         cur_date = time.strftime('%d %b %Y', tmp_time)
                         cur_time = time.strftime('%H:%M:%S', tmp_time)

                         # check verifications and status code for errors
                         is_error = False
                         if resp.code >= 400 or resp.code == 0:
                             is_error = True
                         if not req.verify == '':
                             if not re.search(req.verify, content, re.DOTALL):
                                 is_error = True
                         if not req.verify_negative == '':
                             if re.search(req.verify_negative, content, re.DOTALL):
                                 is_error = True
 
                         if is_error:                   
                             self.error_count += 1
                             error_string = 'Agent %s:  %s - %d %s,  url: %s' % (self.id + 1, cur_time, resp.code, resp.msg, req.url)
                             self.error_queue.append(error_string)
                             log_tuple = (self.id + 1, cur_date, cur_time, req_end_time, req.url.replace(',', ''), resp.code, resp.msg.replace(',', ''))
                             self.log_error('%s,%s,%s,%s,%s,%s,%s' % log_tuple)  # write as csv                           
 
                         resp_bytes = len(content)
                         latency = (req_end_time - req_start_time)
                         connect_latency = (connect_end_time - req_start_time)                       
 
                         self.count += 1
                         total_bytes += resp_bytes
                         total_latency += latency
                         total_connect_latency += connect_latency                       
 
                         # update shared stats dictionary
                         self.runtime_stats[self.id] = StatCollection(resp.code, resp.msg, latency, self.count, self.error_count, total_latency, total_connect_latency, total_bytes)
                         self.runtime_stats[self.id].agent_start_time = agent_start_time
 
                         # put response stats/info on queue for reading by the consumer (ResultWriter) thread
                         q_tuple = (self.id + 1, cur_date, cur_time, req_end_time, req.url.replace(',', ''), resp.code, resp.msg.replace(',', ''), resp_bytes, latency, connect_latency, req.timer_group)
                         self.results_queue.put(q_tuple)
 
                         expire_time = (self.interval - latency)
                         if expire_time > 0:
                             time.sleep(expire_time)  # sleep remainder of interval so we keep even pacing                   
 
                     else:  # don't go through entire range if stop has been called
                         break

52/5<12345>
《2023软件测试行业现状调查报告》独家发布~

精彩评论

  • zdlhappy
    2014-5-09 09:43:23

    请教一个问题,我装好了所有需要装的,但是result.html上还是没有图片,日志里也没报错,请大神指点!!

关注51Testing

联系我们

快捷面板 站点地图 联系我们 广告服务 关于我们 站长统计 发展历程

法律顾问:上海兰迪律师事务所 项棋律师
版权所有 上海博为峰软件技术股份有限公司 Copyright©51testing.com 2003-2024
投诉及意见反馈:webmaster@51testing.com; 业务联系:service@51testing.com 021-64471599-8017

沪ICP备05003035号

沪公网安备 31010102002173号