common_func.py 17.99 KiB
import base64
import json
import requests
from airtest.core.api import *
from common.confop import confOP
from common.dubboUtils import GetDubboService2, InvokeDubboApi
from common.rw import Rw
workspace = os.path.abspath(".")
business_path = workspace + os.sep + "data" + os.sep
class commonFuc(object):
    def find_path(self, module=""):
        if module == "":
            return business_path
        else:
            return business_path + module + os.sep
    def get_business_data(self, module, key, *args):
        env = os.environ['ENV']
        data_list = confOP().getBusiYamlValue(self.find_path(module), "data")
        if args is not None and len(args) > 0:
            if isinstance(data_list[key], dict):
                result = json.loads(json.dumps(data_list[key]) % args)
            else:
                if str(data_list[key]) == "":
                    result = data_list[key]
                else:
                    result = data_list[key] % args
        else:
            result = data_list[key]
        if "_sit_" in str(result) and "_on_" in str(result):
            if env == "on":
                return result["_on_"]
            else:
                return result["_sit_"]
        else:
            return result
    def get_message(self, module, key):
        message_list = confOP().getBusiYamlValue(self.find_path(module), "message")
        return message_list[key]
    def get_openauth_url(self):
        return "http://openauth.meitun.com/tools/sign/create"
    def get_api_url(self):
        """
        接口类型的请求
        :return:
        """
        env = os.environ['ENV']
        print(env)
        if env == 'on':
            url = 'https://m.meitun.com'
        elif env == 'pre':
            url = 'http://pre-m.meitun.com'
        elif env == 'sita':
            url = 'http://sita-m.meitun.com'
        elif env == "spd3":  # spd3.0
            # url = 'http://sit-m.meitun.com'
            url = 'http://spddev.cmic.com.cn'
        elif env == "spd2":  # spd2.0
            url = ""
        elif env == "od":  # 骨科
            url = ""
7172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140
elif env == "bs": # 电商 url = "https://service-slb.cmic.com.cn" else: url = 'http://spddev.cmic.com.cn' return url def get_token(self, module, enc_user_id="enc_user_id"): enc_user_id = self.get_business_data(module, enc_user_id) token = self.check_token(enc_user_id) return token def check_token(self, enc_user_id='u779700044448'): """ 多个case同一个token不用一直查数据库 :param enc_user_id: :return: """ env = os.environ['ENV'] result = Rw().r_token(enc_user_id, env) if result is None: return Rw().w_token(enc_user_id, env) else: return result["token"] def http_get(self, url, headers="", params=""): """ 一个get请求,返回json """ result = requests.request("GET", url, headers=headers, params=params) result = json.loads(result.text) return result def http_post(self, url, postdata=None, header=None): """ 一个post请求,返回json """ # result = requests.post(url, data=postdata, headers=header) result = requests.post(url, data=json.dumps(postdata), headers=header) result = json.loads(result.content) return result def check_result(self, check_dict, result): """ 结果检查,要检查的字段及值放在字典里, 结果是完全匹配 result 是json """ assert_not_equal(result, [], "只想看下result的值") for k, v in check_dict.items(): actual_value = self.analysis_json(k, result) assert_equal(str(v).replace(" ", ""), str(actual_value).replace(" ", "")) def analysis_json(self, key, result): """ 解析json """ res = None if self.typeof(result) == 'dict': if key in result.keys(): return result[key] else: for k, v in result.items(): res = self.analysis_json(key, v) if res is not None: break elif self.typeof(result) == 'list': for value in result: res = self.analysis_json(key, value) if res is not None: break