requests_method.py 3.6 KB
Newer Older
fyx_admin committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79
# -*-coding:utf-8 -*- #
# ---------------------------------------------------------------------------
# ProjectName:   test64
# FileName:      requests_method.py
# Author:       lao_zhao
# Datetime:     2024/12/26 14:21
# Description:
# 
# ---------------------------------------------------------------------------
import requests

from APIAutoTest_v3.common.db import DB
from APIAutoTest_v3 import log_decorator
from APIAutoTest_v3.common.read_basic_ini import ReadBasicIni


class RequestsMethod:
    @log_decorator
    def __init__(self):
        """管理被测系统的登录状态"""
        self.db = DB()
        self.bpm_session = requests.sessions.Session()
        login_url = ReadBasicIni().get_host("bpm") + "/auth"
        login_data = {"username":"admin","password":"hxWNSKUPo8BKFLKduQ/yp1eSkefdSd9u1oKwY4mksY8K3zQ8olzp8mbHX61GDRrmdtEqrsnfjm7IKPrz7n1MFDKN0jGsU2xZ7buyJKMjYJj5mZUDcIp85T70ijjpEisQQGnH+75d6R3SLM3LTSRWX9ziB35PYJUHT5KWjYd6i8Q="}
        self.bpm_session.headers["Authorization"] = "Bearer "+self.bpm_session.post(url=login_url, json=login_data).json().get("token")
        print(self.bpm_session.headers["Authorization"])

    @log_decorator
    def close(self):
        self.db.close()

    @log_decorator
    def request_all(self, req_url, req_mime, req_method, case_data, sql_type, sql_data, update_key):
        """封装一个公共的请求方法,先执行sql语句再发生请求。"""
        # 先执行删除sql语句
        if sql_type == "delete":
            self.db.delete(sql_data)
        # 再执行查询的sql语句,将查询结果更新到用例数据中
        elif sql_type == "select":
            select_result = self.db.select(sql_data)
            case_data[update_key] = select_result
        # 判断sql语句是否为delete|select或者为select|delete
        elif sql_type == "delete|select" or sql_type == "select|delete":
            # 先执行删除
            self.db.delete(sql_data["delete"])
            # 再执行查询,将查询结果更新到用例数据中
            select_result = self.db.select(sql_data["select"])
            case_data[update_key] = select_result


        # 发送请求
        # requests关心的是传参的关键字,不关心请求方法
        # 判断媒体类型是否为query|json或者json|query表示请求体使用json传参,同时也在地址栏传参
        if req_mime == "query|json" or req_mime == "json|query":
            # requests支持地址栏和请求体同时传参,请求体中每次只能使用一个关键字传参。
            return self.bpm_session.request(method=req_method, url=req_url, json=case_data["json"], params=case_data["query"])

        elif req_mime == "application/json" or req_mime == "json":
            return self.bpm_session.request(method=req_method, url=req_url, json=case_data)

        # requests如果在请求体中传参,除了json数据使用json关键字以外,传递的所有数据全部使用data
        elif req_mime == "application/x-www-form-urlencoded" or req_mime == "form":
            return self.bpm_session.request(method=req_method, url=req_url, data=case_data)

        elif req_mime == "multipart/form-data" or req_mime == "form-data":
            return self.bpm_session.request(method=req_method, url=req_url, files=case_data)

        elif req_mime == "query":
            return self.bpm_session.request(method=req_method, url=req_url, params=case_data)

        elif req_mime is None:
            return self.bpm_session.request(method=req_method, url=req_url)

        else:
            raise ValueError("媒体类型的值错误")


if __name__ == '__main__':
    r = RequestsMethod()