在脚本执行前,我们需要脚本调试这个过程,该过程用来验证脚本是否能被正确执行,若脚本本来就存在问题等到执行时再去发现问题就可能浪费大量执行时间,因此在这个阶段,我们需要执行一次脚本,并验证脚本是否正确。
首先我们需要将所有的脚本上传到节点上,并保证该节点机安装有一些压测工具,这里以grinder为例,首先需要配置grinder.properties文件,以我的例子来说明:
script1 = createUser script2 = updateUinfo script3 = updateToken script4 = getUserInfo script5 = setSpecialRelation script6 = updateUserID script7 = getToken script8 = addFriend script9 = getFriendRelation script10 = updateRelationship script11 = addGroup script12 = queryTeam script13 = queryTeamNoUser script14 = joinTeams script15 = sendTeamMsg script16 = SendCustomMessage script17 = sendGroupMessage script18 = sendBatchAttachMsg script19 = sendBatchMsg script20 = kick grinder.script = Serial.py grinder.processes = 1 grinder.threads = 1 grinder.runs = 1script.*代表是待调试脚本的名称,Serial.py是主脚本名,grinder.processes ,grinder.threads,grinder.runs 分别是grinder的进程,线程,以及运行次数,因为这部分主要是调试脚本,这里的参数全部设置为1。Serial.py实际是一个串行脚本,它负责顺序执行各脚本,代码如下所示:
from net.grinder.script.Grinder import grinder from java.util import TreeMap # TreeMap is the simplest way to sort a Java map. scripts = TreeMap(grinder.properties.getPropertySubset("script")) # Ensure modules are initialised in the process thread. for module in scripts.values(): exec("import %s" % module) def create_test_runner(module): x=\'\' exec("x = %s.TestRunner()" % module) return x class TestRunner: def __init__(self): self.testRunners = [create_test_runner(m) for m in scripts.values()] # This method is called for every run. def __call__(self): #create_test_runner() for testRunner in self.testRunners: testRunner()执行完该脚本后需要验证该脚本的正确性,我的做法是验证classb-im14-0-data.log下的日志信息,读取error列的值,具体代码如下:
info = [] f = open(\'result.txt\', \'w\') path = os.getcwd() #print path path+=\'/logs\' os.chdir(path) path = os.getcwd() #print path file=open(\'classb-im14-0-data.log\',\'r\') count=len(file.readlines()) while(count!=interfaceNum): count=len(file.readlines()) file=open(\'classb-im14-0-data.log\',\'r\') for line in file: info.append(line.strip()) if line.find("Thread")>=0: continue else: vec=line.split(\',\') if vec[5].strip()!=\'0\': #print vec[5] str=testIdToScene(vec[2].strip()) if str==None: f.write(\'testId does not exit\') excuteflag=False break else: str+=(\' Error\n\') f.write(str) flag=False if flag==True and excuteflag==True: f.write(\'All interfaces have been successfully executed\') f.close() file.close()