基礎類封裝-解決系統token刷新機制帶來的token變動問題

 

背景：

當前測試系統token存在刷新機制，每隔20min刷新一個最新的token,當在接口測試過程當中，長時間運行致使登陸的token失效

刷新機制：登陸系統後獲取一個toekn_A,15min後刷新一個token_B,20min之內token_A和token_B都生效，20min後token_A失效，token_B生效(以此循環)，故爲此須要解決自動化測試過程當中的token刷新機制

解決思路：

1.將登陸後返回的token_A存放到測試環境的數據庫中

2.主進程在運行過程當中開啓一個子線程去請求心跳接口，按心跳間隔獲取最新的token_B信息，若是token_A等於token_B則不更新數據，反之則將最新的token_B存放到數據庫中

  1 import os
  2 import sys 3 sys.path.insert(0,os.getcwd()) 4 import time 5 import json 6 import threading 7 from lib.tools import * 8 from lib.mbh_init_db import * 9 from lib.logger import logger 10 from util.encry import md5_key 11 from interface.tool import Tool 12 from httprun.HttpRun import HttpRun 13 from util.RandomStr import RandomStr 14 from config.VarConfig import cert_file 15 from config.golbalSetting import hostname,cmapi 16 17 18 class CMAPI(threading.Thread): 19 20 # 定義心跳間隔,單位s 21 interval = 45 22 # 定義域名信息 23 domain = cmapi 24 # 忽略SSL驗證 25 verify = False 26 # 定義crt證書文件 27 cert = cert_file 28 # 定義操做終端 29 sobeyhive_http_system = getattr(Tool,'sobeyhive_http_system') 30 # 定義工具類型 31 sobeyhive_http_tool = getattr(Tool,'sobeyhive_http_tool') 32 # 定義site 33 sobeyhive_http_site = getattr(Tool,'sobeyhive_http_site') 34 # 定義user-agent 35 user_agent = getattr(Tool,'user_agent') 36 # 定義數據表的名稱 37 init_table = table_name 38 # 定義日誌 39 logger = logger 40 41 def __init__(self,username,password): 42 threading.Thread.__init__(self) # 定義繼承類初始化方法 43 self.username = username 44 self.password = password 45 self.httprun = HttpRun().http_run 46 self.db = initDB() 47 self.logoutflag = False # 設置登出標記 48 self.validuser = False # 設置用戶是否有效 49 # 定義用戶信息 50 self.logininfoid = None # 心跳維護使用 51 self.loginip = None # 登陸ip 52 self.loginname = None # 登陸用戶名 53 self.sitecode = None # 多站點 54 self.usercode = None 55 self.userid = None 56 self.usertoken = None 57 self.organizationcode = None 58 self.refreshtoken = None # 定義刷新的totkn，從子線程中獲取 59 # cookies 60 self.cookies = None 61 62 def login(self): 63 """ 64 login 65 :return dict type 66 """ 67 self.logger.logger.info('當前方法: %s'%p.get_current_function_name()) 68 url = self.domain + "/CMApi/api/basic/account/login" 69 self.logger.logger.info('請求地址: %s' % url) 70 headers = { 71 "Content-Type": "application/json" 72  } 73 self.logger.logger.info('請求頭: %s' % headers) 74 passwd = md5_key(self.password) if self.password else None 75 playload = { 76 "LOGINNAME": self.username, 77 "LOGINPWD": passwd, 78 "LOGINSUBSYSTEM": self.sobeyhive_http_system['WEBCM'], 79 "LOGINIP": RandomStr.generate_uuid() 80  } 81 self.logger.logger.info('請求參數: %s' % playload) 82 try: 83 response = self.httprun(method='post', url=url, data=json.dumps(playload), headers=headers,verify=self.verify) 84 if response is None: 85 sys.exit('login response is None,Please Check environment!') 86 res = response.json() 87 if res['code'] == '0': 88 setattr(self, 'logininfoid', res['ext']['logininfoid']) 89 setattr(self, 'loginip', res['ext']['loginip']) 90 setattr(self, 'loginname', res['ext']['loginname']) 91 setattr(self, 'sitecode', res['ext']['organizations'][0]['sitecode']) 92 setattr(self, 'usercode', res['ext']['usercode']) 93 setattr(self, 'userid', res['ext']['userid']) 94 setattr(self, 'usertoken', res['ext']['usertoken']) 95 setattr(self, 'organizationcode', res['ext']['organizations'][0]['organizationcode']) 96 setattr(self, 'cookies', response.cookies) 97 # print(json.dumps(re,indent=2,ensure_ascii=False)) 98 # 存儲用戶名及token,用戶後續的心跳更新,先判斷表中是否存在token,存在則更新，不存在則插入 99 query_sql = "select username from {0} where username = '{1}'".format(self.init_table,self.username) 100 result = self.db.query_db(query_sql,state="all") 101 if isinstance(result,tuple) and len(result)==0: 102 insert_sql = "insert into {0} value('{1}','{2}');".format(self.init_table, self.loginname, self.usertoken) 103  self.db.execute_db(insert_sql) 104 else: 105 if self.username in result[0].get('username'): 106 updata_sql = "update {0} set usertoken = '{1}' where username='{2}'".format(self.init_table,self.usertoken,self.username) 107  self.db.execute_db(updata_sql) 108 else: 109 self.logger.logger.info('請求耗時: %s 毫秒' % (int(response.elapsed.microseconds) // 1000)) 110 self.logger.logger.error('登陸失敗,Status Code: %s'% response.status_code) 111 return res 112 except Exception as e: 113  self.logger.logger.exception(e) 114 return None 115 else: 116 self.logger.logger.info('請求耗時: %s 毫秒'%(int(response.elapsed.microseconds)//1000)) 117 self.logger.logger.info('響應結果: %s' % res) 118 self.validuser = True 119 return res 120 121 def testfunctionpopedom(self): 122 """判斷登陸權限""" 123 self.logger.logger.info('當前方法: %s' % p.get_current_function_name()) 124 if self.validuser: 125 usertoken = self.get_token() 126 url = self.domain + "/CMApi/api/basic/account/testfunctionpopedom?usertoken={0}".format(usertoken) 127 self.logger.logger.info('請求地址: %s' % url) 128 headers = { 129 "Content-Type": "application/json", 130 "sobeyhive-http-site": self.sitecode, 131 "sobeyhive-http-system": CMAPI.sobeyhive_http_system['WEBCM'], 132 "sobeyhive-http-token": usertoken, 133 "sobeyhive-http-tool": CMAPI.sobeyhive_http_tool['WEBCM'], 134 "sobeyhive-http-usercode": self.usercode 135  } 136 self.logger.logger.info('請求頭: %s' % headers) 137 payload = [ 138  { 139 "system":"WEBCM", 140 "tool":"DEFAULT", 141 "popedomname":"CANUSEWEBCM" 142  } 143  ] 144 try: 145 response = self.httprun('post', url, headers=headers, data=json.dumps(payload),verify=self.verify) 146 res = response.json() 147 except Exception as e: 148  self.logger.logger.exception(e) 149 else: 150 self.logger.logger.info('請求耗時: %s 毫秒' % (int(response.elapsed.microseconds) // 1000)) 151 self.logger.logger.info('響應結果: %s' % res) 152 return res 153 else: 154 return 155 156 def heartbeat(self): 157 """ 158 cmapi心跳接口 159 :return: 160 """ 161 self.logger.logger.info('當前方法: %s' % p.get_current_function_name()) 162 usertoken = self.get_token() 163 if self.validuser: 164 url = self.domain + "/CMApi/api/basic/account/heatbeat?loginInfoID={0}&systemtype=WEBCM".format(self.logininfoid) 165 self.logger.logger.info('請求地址: %s' % url) 166 headers = { 167 "Content-Type": "application/json", 168 "sobeyhive-http-site": self.sitecode, 169 "sobeyhive-http-system": CMAPI.sobeyhive_http_system['WEBCM'], 170 "sobeyhive-http-token": usertoken, 171 "sobeyhive-http-tool": CMAPI.sobeyhive_http_tool['WEBCM'], 172 "sobeyhive-http-usercode": self.usercode 173  } 174 self.logger.logger.info('請求頭: %s' % headers) 175 try: 176 response = self.httprun('get',url, headers=headers,verify=self.verify) 177 res = response.json() 178 if response.status_code == 200: 179 pass 180 else: 181 self.logger.logger.error("requests failed! Status Code:", res.status_code) 182 except Exception as e: 183  self.logger.logger.exception(e) 184 else: 185 self.logger.logger.info('請求耗時: %s 毫秒' % (int(response.elapsed.microseconds) // 1000)) 186 self.logger.logger.info('響應結果: %s' % res) 187 return res 188 else: 189 return 190 191 def logout(self): 192 """ 193 cmapi登出接口 194 :return: 195 """ 196 self.logger.logger.info('當前方法: %s' % p.get_current_function_name()) 197 url = self.domain + "/CMApi/api/basic/account/logout" 198 self.logger.logger.info('請求地址: %s' % url) 199 try: 200 response = self.httprun(method="get",url=url,verify=self.verify) 201 res = response.json() 202 self.logoutflag = True # 設置登出標誌 203 if res['code'] == "1": 204 self.logger.logger.info("登出成功！") 205 return True 206 else: 207 return False 208 except Exception as e: 209 self.logger.logger.info('請求耗時: %s 毫秒' % (int(response.elapsed.microseconds) // 1000)) 210 self.logger.logger.info('響應結果: %s' % res) 211  self.logger.logger.exception(e) 212 213 def run(self): 214 """ 215 開啓子線程，按固定間隔獲取最新token 216 :return: 217 """ 218 while True: 219 if not self.logoutflag: 220 # 設置心跳間隔 221  time.sleep(self.interval) 222 res = CMAPI.heartbeat(self) 223  self.logger.logger.info(res) 224 #TODO： 更新token到mysql中,先查詢token值 225 query_sql = "select usertoken from {0} where username = '{1}'".format(self.init_table,self.username) 226 result = self.db.query_db(query_sql) 227 if res['ext'] != '': 228 if result[0].get('usertoken') != res['ext']: 229 update_sql = "update {0} set usertoken = '{1}' where username='{2}'".format(self.init_table,res['ext'],self.username) 230  self.db.execute_db(update_sql) 231 self.refreshtoken = res['ext'] 232 else: 233 self.logger.logger.error("token更新失敗：",res) 234 else: 235 break 236 237 def get_token(self): 238 """獲取用戶的token信息""" 239 sql = "select usertoken from {0} where username = '{1}';".format(self.init_table, self.username) 240  self.db.select_db(db_name) 241 query_result = self.db.query_db(sql) 242 if query_result: 243 return query_result[0].get('usertoken') 244 else: 245 return None 246 247 248 249 250 251 252 253 if __name__ == "__main__": 254 print("父進程PID:%s" % os.getpid()) 255 t = CMAPI('tyw','tyw') 256  t.login() 257 t.start() # 開啓子進程，當存在token刷新機制時開啓 258  t.testfunctionpopedom() 259 # t.get_token() 260 t.logout()