#-*- coding:utf-8 -*- from ssat_sdk.utils.LoggingUtil import printLog from ssat_sdk.device_manage import ScbcCopyKey from ssat_sdk.sat_environment import getOtaParamUrl_dict import urllib import urllib2 import urllib3 import json import requests import os import httplib httplib.HTTPConnection._http_vsn = 10 httplib.HTTPConnection._http_vsn_str = 'HTTP/1.0' # HTTP协议传参时的自动化测试标识，string型，每次请求必带的参数，KEY为type SAT_TYPE = "1" class Ota_Client(): def __init__(self): # 各服务器getParam接口 # self.getParamDict = getOtaParamUrl_dict() self.getParamDict = { "CN": "https://cn.ota.qhmoka.com/ota-services/strategy/getParam", "NA": "https://na.ota.qhmoka.com/ota-services/strategy/getParam", "LA": "https://la.ota.qhmoka.com/ota-services/strategy/getParam", "ME": "https://me.ota.qhmoka.com/ota-services/strategy/getParam", "EU": "https://eu.ota.qhmoka.com/ota-services/strategy/getParam", "AP": "https://ap.ota.qhmoka.com/ota-services/strategy/getParam", "TEST": "https://test.uc.qhmoka.com/ota-services/strategy/getParam" } # http request中统一使用的headers self.headers = { "Content-Type": "application/json" } def postRequest(self, url, post_data, headers): response = requests.post(url, data=post_data, headers=headers) printLog("response:%s" % response) printLog("response.status_code:%s" % response.status_code) printLog("response.text:%s" % response.text) if response.status_code != 200: printLog("OtaClient.postRequest()", u"Response结果码异常。") return False, response else: return True, response def getAppKey(self, area): appKeys = [] url = "" if 1: try: url = self.getParamDict[area] except Exception, e: printLog("OtaClient.addUserId()", u"无法通过传入的area获取url地址。%s"%e) return False # 测试服务器地址 if 0: url = "https://test.uc.qhmoka.com/ota-services/strategy/getParam" post_dict = { "type": SAT_TYPE, "area": area } post_data = json.dumps(post_dict) result, response = self.postRequest(url, post_data, self.headers) if result is False: return False, appKeys resDict = json.loads(response.text) if int(resDict['code']) != 1000: result = False else: result = True appKeys = resDict['data']['appKeys'] printLog("OtaClient.getAppKey()", u"请求返回信息：%s" % resDict['msg']) return result, appKeys def httpAddUserId(self, area, appKey, clientType, projectId, checkVersion, version, userId): url = "" if 1: try: url = self.getParamDict[area] except Exception, e: printLog("OtaClient.addUserId()", u"无法通过传入的area获取url地址。%s"%e) return False # 测试服务器地址 if 0: url = "https://test.uc.qhmoka.com/ota-services/strategy/getParam" post_dict = { "type": SAT_TYPE, "area": area, "appKey": appKey, "clientType": clientType, "projectId": projectId, "checkVersion": checkVersion, "version": version, "userId": userId } post_data = json.dumps(post_dict) result, response = self.postRequest(url, post_data, self.headers) if result is False: return False resDict = json.loads(response.text) if int(resDict['code']) != 1000: result = False else: result = True printLog("OtaClient.addUserId()", u"请求返回信息：%s" % resDict['msg']) return result def httpGetUserId(self, deviceId, clientType, mac, forFormal=True): # 测试环境： testRoot = "http://test.dsp.server.qhmoka.com" # 正式环境： formalRoot = "https://dsp.server.qhmoka.com" # 登录的账号密码 userName = "auto" password = "123456" if forFormal is True: loginUrl = formalRoot + "/login" else: loginUrl = testRoot + "/login" token = ScbcCopyKey.HTTPLogin(loginUrl, userName, password, 1) # token长度不为0, 表示成功获取token值; if token.__len__(): print u"login token = ", token userId = ScbcCopyKey.HTTPGetUserId("http://test.dsp.server.qhmoka.com/api/automation/getUserId", token, clientType, deviceId, mac.replace('-', ':')) # userId == -1表示获取失败; return userId def getDeviceId(self): return ScbcCopyKey.GetDeviceId() def getClientType(self): return ScbcCopyKey.GetClientType() def getMAC(self): return ScbcCopyKey.GetMAC() ''' 从云测试系统随机取一个deviceId，返回 :param forFormal：是否为正式服务器，若为正式，则为True :return deviceId ''' def httpGetDeviceId(self,forFormal=True): # urlTest = "http://10.201.40.65:8088/otaad/deviceId/getDeviceId.do" # 云测试系统地址 # urlFormal = "http://10.201.40.65:8088/otaad/deviceId/getDeviceId.do" urlTest = "http://10.126.16.60:8680/otaad/deviceId/getDeviceId.do" urlFormal = "http://10.126.16.60:8380/otaad/deviceId/getDeviceId.do" post_data = None if forFormal is True: result, response = self.postRequest(urlFormal,post_data,self.headers) else: result, response = self.postRequest(urlTest, post_data,self.headers) # print "result",result # print "response",response resDict = json.loads(response.text) if str(resDict['code']) == "00": return resDict['deviceId'] # 获取成功则返回deviceId else: return None # 获取失败则返回None ''' 返回clientType,deviceId,userId 给云测试平台，将这三个数据绑定在一起 :param clientType,deviceId,userId :return 若删除成功，返回True,否则返回False ''' def httpGetDIdRespond(self,clientType,deviceId,userId=None,forFormal=True): urlTest = "http://10.126.16.60:8680/otaad/deviceId/deleteDeviceId.do" # 云测试系统地址 urlFormal = "http://10.126.16.60:8380/otaad/deviceId/deleteDeviceId.do" supportData = { "deviceId": deviceId, "clientType": clientType, "userId": userId } post_data = json.dumps(supportData) if forFormal is True: result, response = self.postRequest(urlFormal, post_data, self.headers) else: result, response = self.postRequest(urlTest, post_data, self.headers) resDict = json.loads(response.text) if str(resDict['code']) == "00": return True # 删除成功则返回True else: return False # 删除失败则返回False if __name__ == '__main__': otaClient = OtaClient() clientType = "123456" deviceId = "ed5011314ad738e3abaf05868ef168dcae00e094" userId = "987654" forFormal = True # print otaClient.httpGetDeviceId(False) print otaClient.httpGetDIdRespond(clientType,deviceId,userId,False)