compatible with python 3.6

Tests/locustfile.py

@@ -2,9 +2,12 @@
 from locust import Locust, TaskSet, events, task
 import sys
 import os
-sys.path.append(os.path.abspath(os.path.dirname(__file__)))
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir)))
-from test_demo import test_demo
+
+from Tests.test_demo import test_demo
+
+#sys.path.append(os.path.abspath(os.path.dirname(__file__)))
+#sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir)))
+
 
 
 class HttpClient(object):

Tests/run_locust.py

@@ -1,13 +1,16 @@
 # coding:utf-8
 import sys
 import os
+
+from Util.locustTool.locustUtil import locustUtil
+
 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir)))
 from Util import *
 import multiprocessing
 
 
 def start_slave(locust_file, master_port):
-    print 'start slave pid:\t{0}'.format(os.getpid())
+    print ('start slave pid:\t{0}'.format(os.getpid()))
     os.system('locust -f {0} --slave --master-port {1}'.format(locust_file, master_port))
 
 
@@ -30,12 +33,12 @@ def start_slave(locust_file, master_port):
             process.start()
             record.append(process)
 
-        print 'start master pid:\t{0}'.format(os.getpid())
+        print ('start master pid:\t{0}'.format(os.getpid()))
         cmd = 'locust -f {0} {1}'.format(locust_file, locust_command)
-        print 'cmd:\t{0}'.format(cmd)
+        print ('cmd:\t{0}'.format(cmd))
         os.system(cmd)
     else:
         # 单例模式
         cmd = 'locust -f {0} {1}'.format(locust_file, locust_command)
-        print 'cmd:\t{0}'.format(cmd)
+        print ('cmd:\t{0}'.format(cmd))
         os.system(cmd)

Tests/test_Selenium.py

@@ -2,14 +2,15 @@
 
 import sys
 import os
-reload(sys)
-sys.setdefaultencoding("utf-8")
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir)))
-
-from Util import *
+#reload(sys)
+#sys.setdefaultencoding("utf-8")
+#sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir)))
+from Util.commonTool.configUtil import ConfigUtil
 from selenium import webdriver
 import time
 
+from Util.seleniumTool.loginPageUtil import loginPage
+
 selenium_yml = '/config/selenium.yml'
 
 
@@ -36,7 +37,7 @@ def tearDownClass(cls):
     @staticmethod
     def login(self, user, passwd, tips=''):
         # 定义通用login方法
-        print 'Login user: %s ,passwd: %s' % (user, passwd)
+        print ('Login user: %s ,passwd: %s' % (user, passwd))
         # 声明loginPage对象
         login_page = loginPage(self.driver, self.url, self.title)
         # 打开页面
@@ -56,7 +57,7 @@ def login(self, user, passwd, tips=''):
         if tips:
             # 登陆失败校验提示信息
             fail_tips = login_page.show_span()
-            print 'Login Failed: %s' % fail_tips
+            print ('Login Failed: %s' % fail_tips)
             assert tips == fail_tips, 'Check Login Error Tips Failed!'
         else:
             # 登陆成功校验UserID
@@ -65,7 +66,7 @@ def login(self, user, passwd, tips=''):
             login_page.click_logout()
             time.sleep(1)
 
-            print 'Login UserID: %s' % login_userID
+            print ('Login UserID: %s' % login_userID)
             assert user + self.suffix == login_userID, 'Check UserID Failed!'
 
     def test_BVT(self):

Tests/test_baiduSearch.py

@@ -1,34 +1,47 @@
 # encoding: utf-8
+import nose
 import requests
 import re
 from nose.tools import *
 from functools import partial
 
+
 class check_response():
     @staticmethod
     def check_title(response, key):
         # 校验key是否匹配返回页面title
         expect_title = key + u'_百度搜索'
         re_title = re.compile('<title>(.*)</title>')  # 搜索页面title正则表达式
         title = re.search(re_title, response).groups()[0]
-        print 'Search Result Title:%s' % title
-        eq_(title, expect_title, 'Title Check Error!%s != %s' % (title, expect_title))
+        print('Search Result Title:%s' % title)
+        eq_(title, expect_title, 'Title Check Error!%s != %s' %
+            (title, expect_title))
 
     @staticmethod
     def check_results(response, key):
         # 校验key是否匹配搜索结果的名称或者URL
         re_name = re.compile('>(.*)</a></h3>')  # 搜索结果name正则表达式
-        re_url = re.compile('style="text-decoration:none;">(.*)</a><div')  # 搜索结果url正则表达式
+        re_url = re.compile(
+            'style="text-decoration:none;">(.*)</a><div')  # 搜索结果url正则表达式
         names = re.findall(re_name, response)
         urls = re.findall(re_url, response)
 
         for name, url in zip(names, urls):
             # name,url简单处理，去除特殊符号
             name = name.replace('</em>', '').replace('<em>', '')
-            url = url.replace('<b>', '').replace('</b>', '').replace('&nbsp;', '').replace('...', '')
-            print 'Search Results Name:%s\tURL:%s' % (name, url)
+            url = url.replace(
+                '<b>',
+                '').replace(
+                '</b>',
+                '').replace(
+                '&nbsp;',
+                '').replace(
+                '...',
+                '')
+            print('Search Results Name:%s\tURL:%s' % (name, url))
             if key.lower() not in (name + url).lower():
-                assert False, 'Search Results Check Error!%s not in %s' % (key, name + url)
+                assert False, 'Search Results Check Error!%s not in %s' % (
+                    key, name + url)
         return True
 
 
@@ -52,8 +65,9 @@ def search(wd):
     def test_BVT(self):
         # 校验输入不同类型的wd时，百度是否均可正常搜索返回结果
         # wd分类：中文，英文，数字
-        wd_list = [u'lovesoo', u'自动化测试', u'12345']
+        wd_list = ['lovesoo', '自动化测试','12345']
         for wd in wd_list:
             f = partial(test_baiduSearch.search, wd)
-            f.description = wd.encode('utf-8')
-            yield (f,)
+            f.description = wd
+#            f.description = wd.encode('utf-8')
+            yield (f,)

Tests/test_demo.py

@@ -2,12 +2,24 @@
 
 import sys
 import os
-reload(sys)
-sys.setdefaultencoding("utf-8")
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir)))
+#reload(sys)
+#sys.setdefaultencoding("utf-8")
+#sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir)))
+from nose.tools import nottest
+from collections import OrderedDict
 from Util import *
 from functools import partial
-
+from pyhessian import protocol
+
+from Util.DBTool.dbUtil import DBUtil
+from Util.OATool.OAT import OAT
+from Util.checkTool.resultCheck import *
+from Util.hessianTool.hessianUtil import HessianUtil
+from Util.httpTool.httpParam import *
+from Util.httpTool.httpUtil import *
+from Util.redisTool.redisUtil import RedisUtil
+from Util.webserviceTool.webServiceUtil import WebServiceUtil
+from Util.securityTool.securityUtil import Security
 
 class test_demo(object):
     """接口测试demo"""
@@ -69,8 +81,8 @@ def test_webservice(self):
     @nottest
     def test_db(self):
         # 数据库操作demo
-        print DBUtil.execute('select * from user_info;')
-        print DBUtil.execute('select * from user_info;', confSection='Sqlserver')
+        print (DBUtil.execute('select * from user_info;'))
+        print (DBUtil.execute('select * from user_info;', confSection='Sqlserver'))
 
     @nottest
     def test_OA(self):
@@ -96,37 +108,37 @@ def test_OA(self):
                              ('B', ['B1']),
                              ('C', ['C1'])])
 
-        print json.dumps(oat.genSets(case1))
-        print json.dumps(oat.genSets(case2))
-        print json.dumps(oat.genSets(case3))
-        print json.dumps(oat.genSets(case4))
-        print json.dumps(oat.genSets(case4, 1, 0))
-        print json.dumps(oat.genSets(case4, 1, 1))
-        print json.dumps(oat.genSets(case4, 1, 2))
-        print json.dumps(oat.genSets(case4, 1, 3))
+        print (json.dumps(oat.genSets(case1)))
+        print (json.dumps(oat.genSets(case2)))
+        print (json.dumps(oat.genSets(case3)))
+        print (json.dumps(oat.genSets(case4)))
+        print (json.dumps(oat.genSets(case4, 1, 0)))
+        print (json.dumps(oat.genSets(case4, 1, 1)))
+        print (json.dumps(oat.genSets(case4, 1, 2)))
+        print (json.dumps(oat.genSets(case4, 1, 3)))
 
     @nottest
     def test_redis(self):
         # redis/redis cluster操作 demo
-        print RedisUtil.execute('hexists', 'Search:HotWord', u'刘德华')
-        print RedisUtil.execute("get", "userSession:%s", "12345", confSection='Redis_Cluster')
+        print (RedisUtil.execute('hexists', 'Search:HotWord', u'刘德华'))
+        print (RedisUtil.execute("get", "userSession:%s", "12345", confSection='Redis_Cluster'))
 
     @nottest
     def test_security(self):
         # 加密方法使用demo
         import string
         sec = Security()
-        key_8 = string.lowercase[:8]
-        key_16 = string.lowercase[:16]
+        key_8 = str.lower()
+        key_16 = str.lower()
         data = 'Taffy is a Test Automation Framework based on nosetests.'
 
-        print 'DES:', sec.getDES(key_8, data).encode('hex')  # des
-        print 'Decode DES:', sec.decodeDES(key_8, sec.getDES(key_8, data))  # decode des
-        print 'DES3:', sec.getDES3(key_16, data).encode('hex')  # desc3
-        print 'Decode DES3:', sec.decodeDES3(key_16, sec.getDES3(key_16, data))  # decode desc3
-        print 'HMAC_SHA1:', sec.getHMAC_SHA1(key_8, data)  # sha1
-        print 'SHA:', sec.getSHA(data)  # sha
-        print 'MD5:', sec.getMD5(data)  # md5
-        print 'AES:', sec.getAES(key_16, data).encode('hex')  # aes
-        print'Base64:', sec.getBase64(data)  # base64
-        print'Decode Base64:', sec.decodeBase64(sec.getBase64(data))  # decode base64
+        print ('DES:', sec.getDES(key_8, data).encode('hex') ) # des
+        print ('Decode DES:', sec.decodeDES(key_8, sec.getDES(key_8, data)))  # decode des
+        print ('DES3:', sec.getDES3(key_16, data).encode('hex'))  # desc3
+        print ('Decode DES3:', sec.decodeDES3(key_16, sec.getDES3(key_16, data)))  # decode desc3
+        print ('HMAC_SHA1:', sec.getHMAC_SHA1(key_8, data))  # sha1
+        print ('SHA:', sec.getSHA(data))  # sha
+        print ('MD5:', sec.getMD5(data))  # md5
+        print ('AES:', sec.getAES(key_16, data).encode('hex'))  # aes
+        print('Base64:', sec.getBase64(data))  # base64
+        print('Decode Base64:', sec.decodeBase64(sec.getBase64(data)))  # decode base64

Tests/test_doubanSearch.py

@@ -2,8 +2,8 @@
 
 import sys
 import os
-reload(sys)
-sys.setdefaultencoding("utf-8")
+#reload(sys)
+#sys.setdefaultencoding("utf-8")
 import requests
 import json
 from datetime import datetime as dt
@@ -42,7 +42,7 @@ def send_mail():
         # 退出
         s.quit()
     except Exception as e:
-        print "Exceptioin ", e
+        print ("Exceptioin ", e)
 
 
 class check_response():
@@ -101,10 +101,10 @@ class test_doubanSearch(object):
     def search(params, expectNum=None):
         url = 'https://api.douban.com/v2/movie/search'
         r = requests.get(url, params=params)
-        print 'Search Params:\n', json.dumps(params, ensure_ascii=False)
-        print 'Search Response:\n', json.dumps(r.json(), ensure_ascii=False, indent=4)
-        code = r.json().get('code')
-        if code > 0:
+        print ('Search Params:\n', json.dumps(params, ensure_ascii=False))
+        print ('Search Response:\n', json.dumps(r.json(), ensure_ascii=False, indent=4))
+        code = r.status_code
+        if code != 200:
             assert False, 'Invoke Error.Code:\t{0}'.format(code)
         else:
             if params.get('start') is not None or params.get('count') is not None:
@@ -120,7 +120,7 @@ def test_q(self):
         for q in qs:
             params = dict(q=q)
             f = partial(test_doubanSearch.search, params)
-            f.description = json.dumps(params, ensure_ascii=False).encode('utf-8')
+            f.description = json.dumps(params, ensure_ascii=False)
             yield (f,)
 
     def test_tag(self):
@@ -129,7 +129,7 @@ def test_tag(self):
         for tag in tags:
             params = dict(tag=tag)
             f = partial(test_doubanSearch.search, params)
-            f.description = json.dumps(params, ensure_ascii=False).encode('utf-8')
+            f.description = json.dumps(params, ensure_ascii=False)
             yield (f,)
 
     def test_param_combination(self):
@@ -141,10 +141,10 @@ def test_param_combination(self):
         for params in params_list:
             if isinstance(params, tuple):
                 f = partial(test_doubanSearch.search, params[0], params[1])
-                f.description = json.dumps(params[0], ensure_ascii=False).encode('utf-8')
+                f.description = json.dumps(params[0], ensure_ascii=False)
             else:
                 f = partial(test_doubanSearch.search, params)
-                f.description = json.dumps(params, ensure_ascii=False).encode('utf-8')
+                f.description = json.dumps(params, ensure_ascii=False)
             yield (f,)
 
     def test_page(self):
@@ -155,7 +155,7 @@ def test_page(self):
             start = page * count
             params = dict(q=q, start=start, count=count)
             f = partial(test_doubanSearch.search, params)
-            f.description = json.dumps(params, ensure_ascii=False).encode('utf-8')
+            f.description = json.dumps(params, ensure_ascii=False)
             yield (f,)
 
     def test_param_q(self):
@@ -165,11 +165,11 @@ def test_param_q(self):
             if isinstance(q, tuple):
                 params = dict(q=q[0])
                 f = partial(test_doubanSearch.search, params, q[1])
-                f.description = json.dumps(params, ensure_ascii=False).encode('utf-8')
+                f.description = json.dumps(params, ensure_ascii=False)
             else:
                 params = dict(q=q)
                 f = partial(test_doubanSearch.search, params)
-                f.description = json.dumps(params, ensure_ascii=False).encode('utf-8')
+                f.description = json.dumps(params, ensure_ascii=False)
             yield (f,)
 
     def test_param_tag(self):
@@ -179,11 +179,11 @@ def test_param_tag(self):
             if isinstance(tag, tuple):
                 params = dict(tag=tag[0])
                 f = partial(test_doubanSearch.search, params, tag[1])
-                f.description = json.dumps(params, ensure_ascii=False).encode('utf-8')
+                f.description = json.dumps(params, ensure_ascii=False)
             else:
                 params = dict(tag=tag)
                 f = partial(test_doubanSearch.search, params)
-                f.description = json.dumps(params, ensure_ascii=False).encode('utf-8')
+                f.description = json.dumps(params, ensure_ascii=False)
             yield (f,)
 
     def test_param_start(self):
@@ -192,7 +192,7 @@ def test_param_start(self):
         for start in start_list:
             params = dict(q=q, start=start)
             f = partial(test_doubanSearch.search, params)
-            f.description = json.dumps(params, ensure_ascii=False).encode('utf-8')
+            f.description = json.dumps(params, ensure_ascii=False)
             yield (f,)
 
     def test_param_count(self):
@@ -201,7 +201,7 @@ def test_param_count(self):
         for count in count_list:
             params = dict(tag=tag, count=count)
             f = partial(test_doubanSearch.search, params)
-            f.description = json.dumps(params, ensure_ascii=False).encode('utf-8')
+            f.description = json.dumps(params, ensure_ascii=False)
             yield (f,)
 
 
@@ -220,9 +220,9 @@ def test_param_count(self):
     report_file = 'TestReport.html'
 
     # 运行nosetests进行自动化测试并生成测试报告
-    print 'Run Nosetests Now...'
+    print ('Run Nosetests Now...')
     os.system('nosetests -v test_doubanSearch.py:test_doubanSearch --with-html --html-file={0}'.format(report_file))
 
     # 发送测试报告邮件
-    print 'Send Test Report Mail Now...'
+    print ('Send Test Report Mail Now...')
     send_mail()