MySession.py 9.5 KB
#!/usr/bin/python
# -*- coding: UTF-8 -*-
import requests
import json
from commons import common as com
from commons.Logging import Logger
from commons.clientSession import cliSession
log=Logger()

class mysession():
    "封装了requests的基类,以供后期统一使用"

    url = "http://test.uap.diligrp.com/login/login.action"
    header = {
        "Host": "test.uap.diligrp.com",
        "Connection": "keep-alive",
        "Content-Length": "33",
        "Cache-Control": "max-age=0",
        "Upgrade-Insecure-Requests": "1",
        "Origin": "http://test.uap.diligrp.com",
        "Content-Type": "application/x-www-form-urlencoded",
        "User-Agent": "Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/90.0.4430.212Safari/537.36",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9",
        "Referer": "http://test.uap.diligrp.com/login/index.html",
        "Accept-Encoding": "gzip,deflate",
        "Accept-Language": "zh-CN,zh-TW;q=0.9,zh;q=0.8,en;q=0.7",
        "Cookie": "UAP_accessToken=;UAP_refreshToken=;UAP_loginPath="}
    body = "userName=sg_wenze&password=111111"

    def __init__(self):
        "如下代码,可以通过配置文件来控制测试环境和灰度环境,http和https"
        self.url = mysession.url.replace("http://test.", com.get_global_config("global_data", "environment", "en"))
        self.header = mysession.header
        self.body = mysession.body
        self.timeout = (5, 5)
        self.max_retries = 3
        self.keep_alive = False
        self.ssl_verify = False
        self.proxies = None
#         self.proxies={'http': 'http://localhost:8888', 'https': 'http://localhost:8888'}
        self.allow_redirects = False
        self.firmid={"group":"1","hd":"2","cd":"3","qqhe":"4","mdj":"5","gy":"6","cc":"7","sg":"8","sy":"9"}


    def cliLogin(self):
        self.webHeaders, self.clientHeaders, self.userInfo = cliSession().loginUser()

    def get_session(self, account, **kwargs):
        # print("get_session")
        "如下代码,可以通过配置文件来控制登录的账户session"
        self.body = self.body.replace("sg_wenze",
                                      com.get_global_config("global_data", "account", account).split("&")[0])
        self.body = self.body.replace("111111", com.get_global_config("global_data", "account", account).split("&")[1])
        # requests.session()会话保持,比如使用session成功的登录了某个网站,
        # 则在再次使用该session对象求求该网站的其他网页都会默认使用该session之前使用的cookie等参数
        self.se = requests.session()
        #更新cookie加入UAP_firmId
        firm=account.split("_")[0]
        co = requests.cookies.RequestsCookieJar()
        co.set("UAP_firmId", self.firmid[firm])
        self.se.cookies.update(co)
        # 使用session对象的方法POST/GET等
        re = self.se.post(url=self.url, headers=self.header, data=self.body, proxies=self.proxies, **kwargs)
        #获取关键信息供其他接口header使用
        self.UAP_accessToken=self.se.cookies["UAP_accessToken"]
        self.UAP_refreshToken=self.se.cookies["UAP_refreshToken"]
        # 返回session对象,供其他接口使用
        return self.se

    def close_session(self):
        "关闭session"
        self.se.close()

    def check_login(self, account, **kwargs):
        "验证登录接口"
        self.body = self.body.replace("sg_wenze",
                                      com.get_global_config("global_data", "account", account).split("&")[0])
        self.body = self.body.replace("111111", com.get_global_config("global_data", "account", account).split("&")[1])
        # POST请求,
        re = self.se.post(url=self.url, headers=self.header, data=self.body, proxies=self.proxies,
                          allow_redirects=False, **kwargs)
        # 判断请求
        assert "UAP_accessToken" in re.headers["Set-Cookie"]
        assert "UAP_refreshToken" in re.headers["Set-Cookie"]
        assert "UAP_loginPath" in re.headers["Set-Cookie"]
        print("登录接口验证成功")
        return re

    def useHeadersRequests(self, method=None, url=None, headers=None, data=None, **kwargs):
        """
        模拟客户端接口操作,使用headers进行cookies传递,调用requests库进行接口访问
        :param method:接口请求方式,POST,GET等
        :param url:接口路径
        :param data:接口数据
        :kwargs:其他requests.request()支持参数可以直接传递
        """
        print(url)
        # print(headers)
        print(data)
        log.info("{0:=^86}".format(''))
        log.info("{}\n{}\n{}\n".format(url, data, kwargs))
        if "gateway" in url:
            # 判断接口路径,通过接口路径钟是否包含gateway来判定接口是否是由客户端进行访问,来判定headers使用
            self.clientHeaders = dict(self.clientHeaders, **headers)
            res = requests.request(method=method, url=url, data=data, headers=self.clientHeaders, **kwargs)
        else:
            self.webHeaders = dict(self.webHeaders, **headers)
            res = requests.request(method=method, url=url, data=data, headers=self.webHeaders, **kwargs)
        return res

    def get(self, url, **kwargs):
        """Sends a GET request. Returns :class:`Response` object.

        :param url: URL for the new :class:`Request` object.
        :param \*\*kwargs: Optional arguments that ``request`` takes.
        :rtype: requests.Response
        """
        # 记录日志
        log.info("{0:=^86}".format(''))
        log.info("{}\n{}\n".format(url, kwargs))
        # 进行请求
        re = self.se.get(url, **kwargs, proxies=self.proxies, timeout=self.timeout)
        return re

    def post(self, url, data=None, json=None, **kwargs):
        """Sends a POST request. Returns :class:`Response` object.
        :param url: URL for the new :class:`Request` object.
        :param data: (optional) Dictionary, list of tuples, bytes, or file-like
            object to send in the body of the :class:`Request`.
        :param json: (optional) json to send in the body of the :class:`Request`.
        :param \*\*kwargs: Optional arguments that ``request`` takes.
        :rtype: requests.Response
        """
        # 记录日志
        log.info("{0:=^86}".format(''))
        log.info("{}\n{}\n{}\n{}".format(url, data, json, kwargs))
        # 进行请求
        re = self.se.post(url, data=data, json=json, proxies=self.proxies, **kwargs, timeout=self.timeout)
        return re

    def options(self, url, **kwargs):
        """Sends a OPTIONS request. Returns :class:`Response` object.
        :param url: URL for the new :class:`Request` object.
        :param \*\*kwargs: Optional arguments that ``request`` takes.
        :rtype: requests.Response
        """
        # 记录日志
        log.info(url, kwargs)
        # 进行请求
        re = self.se.options(url, **kwargs)
        return re

    def head(self, url, **kwargs):
        """Sends a HEAD request. Returns :class:`Response` object.

        :param url: URL for the new :class:`Request` object.
        :param \*\*kwargs: Optional arguments that ``request`` takes.
        :rtype: requests.Response
        """
        # 记录日志
        log.info(url, kwargs)
        # 进行请求
        re = self.se.head(url, **kwargs)
        return re

    def put(self, url, data=None, **kwargs):
        """Sends a PUT request. Returns :class:`Response` object.
        :param url: URL for the new :class:`Request` object.
        :param data: (optional) Dictionary, list of tuples, bytes, or file-like
            object to send in the body of the :class:`Request`.
        :param \*\*kwargs: Optional arguments that ``request`` takes.
        :rtype: requests.Response
        """
        # 记录日志
        log.info(url, data, kwargs)
        # 进行请求
        re = self.se.put(url, data, **kwargs)
        return re

    def delete(self, url, **kwargs):
        """Sends a DELETE request. Returns :class:`Response` object.
        :param url: URL for the new :class:`Request` object.
        :param \*\*kwargs: Optional arguments that ``request`` takes.
        :rtype: requests.Response
        """
        # 记录日志
        log.info(url, kwargs)
        # 进行请求
        re = self.se.delete(url, **kwargs)
        return re

    def set_mark(self):
        "用户自定义优先级方法"
        mark_list = eval(com.get_global_config("global_data", "mark", "list"))
        print("预设运行标记:", mark_list)
        global _global_mark
        if type(mark_list) == type([]) and len(mark_list) != 0:
            _global_mark = mark_list
            return _global_mark
        elif type(mark_list) == type([]) and len(mark_list) == 0:
            _global_mark = False
            return _global_mark
        else:
            raise Exception("error,pls check mark data")

    def mark(self, m=None):
        try:
            # 没有输入任何预设值,默认跑全部
            if _global_mark == False:
                return True
            # 输入预设值且未标记用例,默认跑全部
            elif _global_mark != False and (m in _global_mark):
                return True
            else:
                return False
        except Exception as e:
            return False


my = mysession()
my.set_mark()
my.cliLogin()
s1 = my.get_session("sg_user01")
s2 = my.get_session("sg_user02")
# sg1=my.get_session("sg_user01")
# sy1=my.get_session("sy_user01")