"""调用python自带的base64，直接对返回的响应做解密，即可得到解密后的响应。封装对于不同算法的处理方法。"""import requestsimport base64import json

def test_encode():    url = "http://127.0.0.1:9999/demo.txt"    r = requests.get(url)    print(r.text)    # content 获取一个二进制的响应结果，对加密内容进行解密，使用json.loads()优化格式    res = json.loads(base64.b64decode(r.content))    print(res)

"""调用python自带的base64，直接对返回的响应做解密，即可得到解密后的响应。封装对于不同算法的处理方法。"""import requestsimport base64import json

class ApiRequest:    def send(self, data:dict):        res = requests.request(data["method"], data["url"], headers=data["headers"])        if data["encoding"] == "base64":            return json.loads(base64.b64decode(res.content))
        # 把加密过后的响应值发给第三方服务，让第三方去做解密然后返回解密过后的信息，private是自定义的，不是固定        elif data["encoding"] == "private":            return requests.post("url", data=res.content)

from interface import test_api_request
class TestApiRequest:    res_data = {        "method": "get",        "url": "http://127.0.0.1:9999/demo.txt",        "headers": None,        "encoding": "base64"    }
    def test_send(self):        # 实例化对象        ar = test_api_request.ApiRequest()        print(ar.send(self.res_data))

# mac设置环境变量，尽量不直接使用env，env可能会和底层环境变量重名，可以使用interface_env等export interface_env=dev# windows 设置环境变量set interface_env=dev

# 测试环境的配置文件base_url: https://httpbin.org/

# 开发环境的配置文件base_url: https://httpbin.ceshiren.com/

import osimport requestsimport yaml

# 设置临时环境变量# Mac： export interface_env="test"# windows：set interface_env="test"# window如果使用的是powershell，需要使用$env:interface_env="test" 命令设置环境变量class TestMulitiEnv:    def setup_class(self):        # 在接口用例中只指定path，不指定url        # 从yaml文件读取数据，env读取出来是一个字典        # 优化点：第一种方式：从环境变量读取名称为inter_env的配置环境，default给默认值        path_env = os.getenv("inter_env", default="test")        env = yaml.safe_load(open(f"{path_env}.yaml", encoding="utf-8"))        # env = yaml.safe_load(open("dev.yaml", encoding="utf-8"))        self.base_url = env["base_url"]
    def test_devenv(self):        """        验证是否为开发环境        :return:        """        path = "get"        r = requests.get(self.base_url + path)        # print(r.json())        # 假设httpbin.ceshiren.com是开发环境，则断言当前请求是否是向"开发环境"发起的        assert r.json()["headers"]["Host"] == "httpbin.ceshiren.com"
    def test_testenv(self):        """        验证是否为测试环境        :return:        """        path = "get"        r = requests.get(self.base_url + path)        # r = requests.get("https://httpbin.org/get")        assert r.json()["headers"]["Host"] == "httpbin.org"

# 设置开发环境变量export interface_env="dev"pytest test_muliti_env.py 
# 设置测试环境变量export interface_env="test"pytest test_muliti_env.py 

# 声明变量 global_envglobal_env = {}

# hook 函数，是添加命令行参数使用def pytest_addoption(parser):    # group 将下面所有的 option都展示在这个group下。    mygroup = parser.getgroup("hogwarts")    # 注册一个命令行选项    mygroup.addoption("--env",    # 参数的默认值    default='test',    # 存储的变量    dest='env',    # 参数的描述信息    help='设置接口自动化测试默认的环境'     )

# 获取设置对命令行参数，并传递给一个变量global_envdef pytest_configure(config):    default_ev = config.getoption("--env")    tmp = {"env": default_ev}    global_env.update(tmp)

import requestsimport yamlfrom interface.conftest import global_env

class TestMulitiEnvByOption:    def setup_class(self):        path_env = global_env.get("env")        env = yaml.safe_load(open(f"{path_env}.yaml", encoding="utf-8"))        self.base_url = env["base_url"]
    def test_devenv(self):        """        验证是否为开发环境        :return:        """        path = "get"        r = requests.get(self.base_url + path)        # print(r.json())        # 假设httpbin.ceshiren.com是开发环境，则断言当前请求是否是向"开发环境"发起的        assert r.json()["headers"]["Host"] == "httpbin.ceshiren.com"
    def test_testenv(self):        """        验证是否为测试环境        :return:        """        path = "get"        r = requests.get(self.base_url + path)        # r = requests.get("https://httpbin.org/get")        assert r.json()["headers"]["Host"] == "httpbin.org"

pytest test_muliti_by_option_env.py --env=testpytest test_muliti_by_option_env.py --env=dev

import requestsimport xmltodict

def test_xml_to_dict():    res = requests.get("https://www.nasa.gov/rss/dyn/lg_image_of_the_day.rss")    # 注意，打印res的texts属性    print(res.text)    # 转换为python的dict格式    dict_res = xmltodict.parse(res.text)    print(dict_res)

"""多协议封装：对响应值做二次封装，可以使用统一提取方式完成断言"""import requestsimport xmltodictfrom requests import Response

def test_response_to_dict():    # res = requests.get("https://www.nasa.gov/rss/dyn/lg_image_of_the_day.rss")    res = requests.get("https://httpbin.ceshiren.com/get")    final_res = response_to_dict(res)    # 断言响应值是否为dict类型的格式    assert isinstance(final_res, dict)

def response_to_dict(response: Response):    # 获取text属性    res_text = response.text    # 判断响应文本信息是否以 <?xml 开头    if res_text.startswith("<?xml"):        final_dict = xmltodict.parse(res_text)    elif res_text.startswith("<!DOCTYPE html>"):        final_dict = "html"    # 如果是json 则返回 json 格式    else:        final_dict = response.json()    return final_dict

import requestsfrom intermediate.apis.base_api import BaseApi

# 优化1：接口里面直接使用了 requests# 解决方案：在base_api 中添加公共的 send 方法，通过继承来调用# 优化2：大量重复的base_url# 在构造函数，实例化base_url， 并且在send方法， 就直接做拼接，就无需重复编写self.base_url# 优化3：token的获取也放在 BaseApi 的构造函数中class Goods(BaseApi):
    def create(self, goods_data):        create_url = "admin/goods/create"        # r = requests.post(create_url, json=goods_data, headers={"X-Litemall-Admin-Token": self.token})        r = self.send("post", create_url, json=goods_data)        return r
    def list(self, goods_name, order="desc", sort="add_time"):        goods_list_url = "admin/goods/list"        goods_list_data = {            "name": goods_name,            "order": order,            "sort": sort        }        # r = requests.get(goods_list_url, params=goods_list_data, headers={"X-Litemall-Admin-Token": self.token})        r = self.send("get", goods_list_url, params=goods_list_data)        return r
    def detail(self, goods_id):        goods_detail_url = "admin/goods/detail"        # r = requests.get(goods_detail_url, params={"id": goods_id}, headers={"X-Litemall-Admin-Token": self.token})        r = self.send("get", goods_detail_url, params={"id": goods_id})        return r
    def detele(self, goods_id):        delete_url = "admin/goods/delete"        data = {"id": goods_id}        r = self.send("post", delete_url, json=data)        return r

from intermediate.apis.base_api import BaseApi

class Cart(BaseApi):    def add(self, goods_id, product_id):        cart_url = "wx/cart/add"        cart_data = {"goodsId": goods_id, "number": 1, "productId": product_id}        r = self.send("post", cart_url, json=cart_data)        return r

import jsonimport requestsfrom intermediate.utils.log_utils import logger

class BaseApi:    # 如果有多角色，就做判断，如果没有就不做，即默认传None    def __init__(self, base_url, role=None):        self.base_url = base_url        # 获取对应的角色信息        if role:            self.role = role
        # # ======管理端token获取        # login_b_url = "admin/auth/login"        # login_b_data = {"username": "hogwarts", "password": "test12345", "code": ""}        # r = self.send("post", login_b_url, json=login_b_data)        # # 获取B端 token，定义为实例变量        # # self.token 的声明要在用例之前完成，使用 setup_class 提前完成变量的声明        # self.token_b = r["data"]["token"]        #        # # ======用户端token获取        # login_c_url = "wx/auth/login"        # login_c_data = {"username": "user123", "password": "user123"}        # login_c_r = self.send("post", login_c_url, json=login_c_data)        # # 获取C端 token，定义为实例变量        # self.token_c = login_c_r["data"]["token"]
    def __set_token(self, request_infos):        """        优化点：大部分接口都需要设置token        解决方案：        1、在发起接口请求之前，就获取token信息        2、获取token信息之后，塞入到请求信息之中            除了 method 和 url 之外，所有到其他信息，包括 header、params 等其他的参数，都会塞入至 kwargs 不定长参数中            cart_api（发起请求） —> 调用send方法 —> （send方法控制kwargs）获取 kwargs —> 在 kwargs 中塞入 header（鉴权）信息
        优化点：有两个token如何解决            不同的角色有不同的token信息        :return:        """        # ======管理端token获取        admin_url = "admin/auth/login"        admin_data = {"username": "hogwarts", "password": "test12345", "code": ""}        admin_r = requests.request("post", self.base_url + admin_url, json=admin_data)        # 获取B端 token，定义为实例变量        # B端和C端token的key值不一样，可以将 self.token_b 变为键值对，加入相应的key        self.token = {"X-Litemall-Admin-Token": admin_r.json()["data"]["token"]}
        # ======用户端token获取        client_url = "wx/auth/login"        client_data = {"username": "user123", "password": "user123"}        client_r = requests.request("post", self.base_url + client_url, json=client_data)        # 获取C端 token，定义为实例变量        self.client_token = {"X-Litemall-Admin-Token": client_r.json()["data"]["token"]}
        # 如果是 admin，那么就塞入 admin 的 token，如果是其他，那么就塞入其他的 token        # 在 test_cart 测试用例文件中，实例化时，需要传一个角色的值，确认是否是admin角色还是其他角色        if self.role == "admin":            self.final_token = self.token        else:            self.final_token = self.client_token        # 获取headers，如果请求本身有头信息，那么就把token信息更新（添加）进去        # if request_infos.get("headers"):        #     request_infos["headers"]["X-Litemall-Admin-Token"] = self.token_b        # else:        #     request_infos["headers"] = {"X-Litemall-Admin-Token": self.token_b}        if request_infos.get("headers"):            request_infos["headers"].update(self.final_token)        else:            request_infos["headers"] = self.final_token
        return request_infos
    def send(self, method, url, **kwargs):        self.__set_token(kwargs)        r = requests.request(method, self.base_url + url, **kwargs)        logger.debug(f"{url}接口的响应值为：{json.dumps(r.json(), indent=2, ensure_ascii=False)}")        return r.json()

import pytestfrom intermediate.apis.admin.goods import Goodsfrom intermediate.apis.wx.cart import Cart

@pytest.mark.parametrize("goods_name", ["0529goods01", "0529goods02"])class TestCart:    def setup_class(self):        self.goods = Goods("https://litemall.hogwarts.ceshiren.com/", "admin")        self.cart = Cart("https://litemall.hogwarts.ceshiren.com/", "client")
    def test_add_cart(self, goods_name):        """            添加购物车的步骤：            1、上架商品            2、获取商品列表            3、获取商品详情            4、添加购物车        """        goods_data = {"goods": {"picUrl": "", "gallery": [], "isHot": False, "isNew": True, "isOnSale": True,                                "goodsSn": "23052901", "name": goods_name},                      "specifications": [{"specification": "规格", "value": "标准", "picUrl": ""}],                      "products": [{"id": 0, "specifications": ["标准"], "price": "59", "number": "59", "url": ""}],                      "attributes": []}
        self.goods.create(goods_data)        goods_list_r = self.goods.list(goods_name)        self.goods_id = goods_list_r["data"]["list"][0]["id"]        goods_detail_r = self.goods.detail(self.goods_id)        product_id = goods_detail_r["data"]["products"][0]["id"]        res = self.cart.add(self.goods_id, product_id)        # 将delete放在用例中执行：1、调用方便；2、测试类中的每个方法并非都会添加goods数据；        # 如果部分用例不添加，则相当于这个步骤，不能在每个测试用例执行完成之后执行，只能在单个用例执行完之后执行        self.goods.detele(self.goods_id)        # 断言        assert res["errmsg"] == "成功"

# 配置日志import loggingimport os
from logging.handlers import RotatingFileHandler
# # 绑定绑定句柄到logger对象# logger = logging.getLogger(__name__)# # 获取当前工具文件所在的路径# root_path = os.path.dirname(os.path.abspath(__file__))# # 拼接当前要输出日志的路径# log_dir_path = os.sep.join([root_path, '..', f'/logs'])# if not os.path.isdir(log_dir_path):#     os.mkdir(log_dir_path)# # 创建日志记录器，指明日志保存路径,每个日志的大小，保存日志的上限# file_log_handler = RotatingFileHandler(os.sep.join([log_dir_path, 'log.log']), maxBytes=1024 * 1024, backupCount=10)# # 设置日志的格式# date_string = '%Y-%m-%d %H:%M:%S'# formatter = logging.Formatter(#     '[%(asctime)s] [%(levelname)s] [%(filename)s]/[line: %(lineno)d]/[%(funcName)s] %(message)s ', date_string)# # 日志输出到控制台的句柄# stream_handler = logging.StreamHandler()# # 将日志记录器指定日志的格式# file_log_handler.setFormatter(formatter)# stream_handler.setFormatter(formatter)# # 为全局的日志工具对象添加日志记录器# # 绑定绑定句柄到logger对象# logger.addHandler(stream_handler)# logger.addHandler(file_log_handler)# # 设置日志输出级别# logger.setLevel(level=logging.INFO)
# 日志配置import logging# 创建logger实例logger = logging.getLogger('simple_example')# 设置日志级别logger.setLevel(logging.DEBUG)# 流处理器ch = logging.StreamHandler()ch.setLevel(logging.DEBUG)# 日志打印格式formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')# 添加格式配置ch.setFormatter(formatter)# 添加日志配置logger.addHandler(ch)

class GoodsDomain:    """    抽象的概念，不做具体实现    代表是某一种业务模型    """    def detele_by_name(self, name):        # 如果子类 Goods 没有重写父类方法，是调不通的        # 调用子类方法        goods_list_r = self.list(name)        goods_id = goods_list_r["data"]["list"][0]["id"]        self.detele(goods_id)

from intermediate.apis.wx.cart import Cart

# 不使用 self.goods.detele(self.goods_id)方法# 使用 self.goods.detele_by_name(goods_name) 方法调用delete接口@pytest.mark.parametrize("goods_name", ["0529goods01", "0529goods02"])class TestCart:    def setup_class(self):        self.goods = Goods("https://litemall.hogwarts.ceshiren.com/", "admin")        self.cart = Cart("https://litemall.hogwarts.ceshiren.com/", "client")
    def test_add_cart(self, goods_name):        """            添加购物车的步骤：            1、上架商品            2、获取商品列表            3、获取商品详情            4、添加购物车        """        goods_data = {"goods": {"picUrl": "", "gallery": [], "isHot": False, "isNew": True, "isOnSale": True,                                "goodsSn": "23052901", "name": goods_name},                      "specifications": [{"specification": "规格", "value": "标准", "picUrl": ""}],                      "products": [{"id": 0, "specifications": ["标准"], "price": "59", "number": "59", "url": ""}],                      "attributes": []}
        self.goods.create(goods_data)        goods_list_r = self.goods.list(goods_name)        self.goods_id = goods_list_r["data"]["list"][0]["id"]        goods_detail_r = self.goods.detail(self.goods_id)        product_id = goods_detail_r["data"]["products"][0]["id"]        res = self.cart.add(self.goods_id, product_id)        # 将delete放在用例中执行：1、调用方便；2、测试类中的每个方法并非都会添加goods数据；        # 如果部分用例不添加，则相当于这个步骤，不能在每个测试用例执行完成之后执行，只能在单个用例执行完之后执行        # self.goods.detele(self.goods_id)        self.goods.detele_by_name(goods_name)        # 断言        assert res["errmsg"] == "成功"