scbc_repos/ssat_sdk/client/ota_client.py

#-*- 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 OtaClient():
    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)