1

lixi
1 parent f960ecf0
Showing 2 changed files with 83 additions and 70 deletions
commons/MySession.py
main.py
 #!/usr/bin/python
 # -*- coding: UTF-8 -*-
 import requests
+import re
 from commons import common as com
 from commons.Logging import log
 from commons.clientSession import cliSession
@@ -36,16 +37,17 @@ class mysession(requests.Session):
     body = "userName=sg_wenze&password=111111"
     body_client= {"userName":"sg_wenze","password":"111111"}
-    def __init__(self):
+    def __init__(self,host=None):
         "如下代码，可以通过配置文件来控制测试环境和灰度环境，http和https"
         super().__init__()
+        self.user={}
         self.url = mysession.url.replace("http://test.", com.get_global_config("global_data", "environment", "en"))
         self.header = mysession.header
         self.body = mysession.body
         self.url_client = mysession.url_client.replace("http://test.", com.get_global_config("global_data", "environment", "en"))
         self.header_client = mysession.header_client
         self.body_client = mysession.body_client
-        self.timeout = (6,6)
+        self.timeout = (10,10)
         self.max_retries = 3
         self.keep_alive = False
         self.ssl_verify = False
@@ -55,36 +57,48 @@ class mysession(requests.Session):
         self.allow_redirects = False
         self.firmid={"group":"1","hd":"2","cd":"3","qqhe":"4","mdj":"5","gy":"6","cc":"7","sg":"8","sy":"9"}
         self.market={"sy":"沈阳","heb":"哈尔滨","sg":"寿光","gy":"贵阳","cc":"长春","hs":"杭水","hg":"杭果"}
-        self.user={}
+        self.host= None if host==None else eval(com.get_global_config("global_data", "host_ip", host))
+
     def cliLogin(self,user="sy_userName_01"):
         self.webHeaders, self.clientHeaders, self.userInfo = cliSession().loginUser(user=user)
         return self
-    def get_session(self, account, **kwargs):
-        "如下代码，可以通过配置文件来控制登录的账户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])
-        self.se = requests.session()
-        co = requests.cookies.RequestsCookieJar()
-        #加入UAP_firmId属性
-        firm=account.split("_")[0]
-        co.set("UAP_firmId", self.firmid[firm])
-        self.se.cookies.update(co)
-        # 进行登录请求
-        re = self.se.post(url=self.url, headers=self.header, data=self.body, proxies=self.proxies, **kwargs)
-        self.UAP_accessToken=self.se.cookies["UAP_accessToken"]
-        self.UAP_refreshToken=self.se.cookies["UAP_refreshToken"]
-        return self.se
+    # def get_session(self, account, **kwargs):
+    #     "如下代码，可以通过配置文件来控制登录的账户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])
+    #     self.se = requests.session()
+    #     co = requests.cookies.RequestsCookieJar()
+    #     #加入UAP_firmId属性
+    #     firm=account.split("_")[0]
+    #     co.set("UAP_firmId", self.firmid[firm])
+    #     self.se.cookies.update(co)
+    #     # 进行登录请求
+    #     re = self.se.post(url=self.url, headers=self.header, data=self.body, proxies=self.proxies, **kwargs)
+    #     self.UAP_accessToken=self.se.cookies["UAP_accessToken"]
+    #     self.UAP_refreshToken=self.se.cookies["UAP_refreshToken"]
+    #     return self.se
+    #
+    # def get_login_info(self, account, **kwargs):
+    #     "用于获取用户信息"
+    #     self.body_client.update({"userName":com.get_global_config("global_data", "account", account).split("&")[0]})
+    #     self.body_client.update({"password":pwdCry(com.get_global_config("global_data", "account", account).split("&")[1])})
+    #     tmp = requests.post(url=self.url_client, headers=self.header_client, json=self.body_client, proxies=self.proxies, **kwargs)
+    #     return tmp
-    def get_login_info(self, account, **kwargs):
-        "用于获取用户信息"
-        self.body_client.update({"userName":com.get_global_config("global_data", "account", account).split("&")[0]})
-        self.body_client.update({"password":pwdCry(com.get_global_config("global_data", "account", account).split("&")[1])})
-        tmp = requests.post(url=self.url_client, headers=self.header_client, json=self.body_client, proxies=self.proxies, **kwargs)
-        return tmp
+    def url_pro(self, url, host):
+        # url = url.replace(" ", "")
+        if host!=None:
+            if "http:" in url:
+                d1 = re.match(r"http://(.+?)/", url).group(1).split(":")[0]
+                url = re.sub(r"http://(.+?)/", r"http://" + host[d1] + "/", url)
+            elif "https:" in url:
+                d1 = re.match(r"https://(.+?)/", url).group(1).split(":")[0]
+                url = re.sub(r"https://(.+?)/", r"https://" + host[d1] + "/", url)
+        return url
     def get_session_client(self, account, **kwargs):
         "get_session和get_session_client的方法只能用一个"
@@ -151,6 +165,8 @@ class mysession(requests.Session):
         """
         # 记录日志
         log.info("{0:=^86}".format(''))
+        log.info(url)
+        url=self.url_pro(url,self.host)
         log.info("{}\n{}\n".format(url, kwargs))
         # 进行请求
         re = super().request(method , url, **kwargs,timeout=self.timeout)
@@ -262,15 +278,15 @@ class mysession(requests.Session):
 my = mysession()
 my.set_mark()
-
 # 沈阳客户端session
 sessionSy = mysession().cliLogin("sy_userName_01")
 # 哈尔滨客户端session
 sessionHeb = mysession().cliLogin("hd_userName_01")
-# print(sessionSy.userInfo)
-# print(sessionHeb.userInfo)
-
-sy1=mysession().get_session_client("sy_user01")
-heb=mysession().get_session_client("heb_user01")
-# sg=mysession().get_session_client("sg_user01")
-
+#获取对应市场session
+sy1=mysession("host1").get_session_client("sy_user01")
+heb=mysession("host1").get_session_client("heb_user01")
+hg=mysession("host2").get_session_client("hg_user01")
+# 检测登录接口
+sy1.check_login("sy_user01")
+# 检测登录接口
+hg.check_login("hg_user01")
 \ No newline at end of file
 #!/usr/bin/python
 # -*- coding: UTF-8 -*-
-import os,time
+import os, time
 import sys
 import unittest
 import re
@@ -13,72 +13,69 @@ from commons.MySession import my
 from commons.scripts import delReport
-
 def Create_Testcase_suite(path=""):
     '''创建测试套件'''
-    if path!="":
-        path+="/"
+    if path != "":
+        path += "/"
     print(path)
-    testunit=unittest.TestSuite()
+    testunit = unittest.TestSuite()
     test_Loader = DiscoveringTestLoader()
-    discover=test_Loader.discover("./testcase/{}".format(path),pattern='test_*.py',top_level_dir=None)
+    discover = test_Loader.discover("./testcase/{}".format(path), pattern='test_*.py', top_level_dir=None)
     # print(discover)
     for test_suite in discover:
         testunit.addTests(test_suite)
-#     print(testunit)
+    #     print(testunit)
     return testunit
-    
-def Run_Testcase(testsuit,head=""):
+
+def Run_Testcase(testsuit, head=""):
     '''运行测试用例并生成报告'''
     if (head in my.market.keys()):
-        head="["+ my.market[head] + "]"
-    now = time.strftime("%Y-%m-%d %H_%M_%S",time.localtime())
-    path=os.path.dirname(os.path.abspath(sys.argv[0]))
-    report_file=path+"/report/"+now+"_result.html"
-    #创建报告文件
-    fp=open(report_file,'wb')
+        head = "[" + my.market[head] + "]"
+    now = time.strftime("%Y-%m-%d %H_%M_%S", time.localtime())
+    path = os.path.dirname(os.path.abspath(sys.argv[0]))
+    report_file = path + "/report/" + now + "_result.html"
+    # 创建报告文件
+    fp = open(report_file, 'wb')
-    runner=HTMLTestRunner_cn.HTMLTestRunner(
+    runner = HTMLTestRunner_cn.HTMLTestRunner(
         stream=fp,
         title=u'进门收费{}--接口测试报告'.format(head),
         description=u'用例简要执行情况如下：(注:报告详细信息需要下载report.html并用浏览器打开)',
-        verbosity = 2)
-    #执行用例
+        verbosity=2)
+    # 执行用例
     runner.run(testsuit)
-    #关闭文件
+    # 关闭文件
     fp.close()
     delReport.delReport(path)
     return report_file
-    
+
+
 def Send_email(filename):
     '''判断邮件发送逻辑'''
-    l=eval(com.get_global_config("global_data", "email","to_list").lower())
-    c=eval(com.get_global_config("global_data", "email","cc_list").lower())
-    if type(l)!=type([]):
+    l = eval(com.get_global_config("global_data", "email", "to_list").lower())
+    c = eval(com.get_global_config("global_data", "email", "cc_list").lower())
+    if type(l) != type([]):
         raise Exception("error,pls input list type send-email address")
-    elif len(l)==0:
+    elif len(l) == 0:
         print("\n To_list of send-email is null,won't send email!")
-    elif len(l)!=0:
+    elif len(l) != 0:
         for i in l:
             print("\n check send-email format : {}".format(i))
-            if re.match(r'^[0-9a-zA-Z_]{1,19}@[0-9a-zA-Z]{1,13}\.[com,cn,net]{1,3}$',i):
+            if re.match(r'^[0-9a-zA-Z_]{1,19}@[0-9a-zA-Z]{1,13}\.[com,cn,net]{1,3}$', i):
                 pass
             else:
                 raise Exception("error,pls check your send-email format")
-        #发送邮件
-        em.send_email(filename,send_to=l,cc_to=c)
+        # 发送邮件
+        em.send_email(filename, send_to=l, cc_to=c)
     else:
         print("\n Haven't sent the email,pls check send-email address!")
-        
 if __name__ == "__main__":
-    #检测数据库
+    # 检测数据库
     db.mysql_conn_test()
-    #检测登录接口
-    my.check_login("sy_user01")
-    #创建测试套,运行测试用例,生成报告
-    report=Run_Testcase(Create_Testcase_suite("sy"))
-    #发送邮件
-    Send_email(report)
+    # 创建测试套,运行测试用例,生成报告
+    report = Run_Testcase(Create_Testcase_suite())
+    # 发送邮件
+    Send_email(report)
 \ No newline at end of file