七、Python-Unittest、檔案解析、釘釘傳送訊息、自動化測試案例
(一)Unittest前置條件
1、所有用例執行之前,它會執行一次
@classmethod
def setUpClass(cls):
2、所有用例執行完之後,它會執行一次
@classmethod
def tearDownClass(cls):
3、每條測試用例執行之前都會先執行它
def setUp(self)
4、每條測試用例執行之後都會執行它
def tearDown(self)
5、例項如下:
#!/usr/bin/env python # -*- coding:utf-8 -*- # @FileName :用例前置條件.py # @Time :2020\10\28 0028 22:22# @Author :Krystal # @Desc :Testcase import unittest class Test(unittest.TestCase): @classmethod def setUpClass(cls): # 所有用例執行之前,它會執行一次 print('SetUpClass') @classmethod def tearDownClass(cls): # 所有用例執行完之後,它會執行一次 print('tearDownClass')def tearDown(self): # 每條測試用例執行之後都會執行它 print('tearDown') def setUp(self): # 每條測試用例執行之前都會先執行它 print('setUp') def testa(self): print('testa') def testz(self): print('testz') def testb(self): print('testb') def testc(self):print('testc') if __name__ == "__main__": unittest.main()
執行結果如下:
(二)配置檔案:.ini、yaml、yml
1、ini 檔案是Initialization File的縮寫,即初始化檔案。一般可變的東西儘可能的放在配置檔案當中,易於編輯修改。
(1)配置檔案:config.ini
[redis] host = 127.0.0.1 password = 123456 port = 6379 [mysql] host = 127.0.0.1 password = 123456 port = 6379 user = root db = jxz [server] host = 127.0.0.1:8000
(2)解析配置檔案
判斷查詢的節點存不存在:兩種方法
# 1、判斷節點存不存在: c.sections() # 裡面所有的節點 if node in c.sections(): result = dict(c[node]) return result # 2、用try方法 try: result = dict(c[node]) except Exception as e: print("查詢的節點不存在!") else: return result
(3)所有的程式碼:解析配置檔案.py
import configparser import os # with open('config.ini',encoding='utf-8') as fr: # c = configparser.ConfigParser() # c.read_file(fr) # result = dict(c['server']) # print(result) # 定義函式 def parse_ini(node,file_path='config.ini'): if not os.path.exists(file_path): raise Exception('ini檔案不存在!') with open(file_path, encoding='utf-8') as fr: c = configparser.ConfigParser() c.read_file(fr) # 1、判斷節點存不存在: c.sections() # 裡面所有的節點 if node in c.sections(): result = dict(c[node]) return result # 2、用try方法 # try: # result = dict(c[node]) # except Exception as e: # print("查詢的節點不存在!") # else: # return result result1= dict(c[node]) print(result1) if __name__ == "__main__": redis_info=parse_ini('redis') print(redis_info)
2、配置檔案:yaml和yml:用於儲存測試用例的資料
(1)data.yaml
name : 1
port : 3306
names :
- body
- eyes
- hair
(2)解析yaml檔案,需要提前安裝模組:
pip install pyyaml
(3)解析yaml.py
import yaml with open('data.yaml',encoding='utf-8') as fr: print(yaml.load(fr,Loader=yaml.SafeLoader)) if __name__ == "__main__": pass
注:若遇到如下warning警告:
YAMLLoadWarning: calling yaml.load() without Loader=... is deprecated, as the default Loader is unsafe. Please read https://msg.pyyaml.org/load for full details.
print(yaml.load(fr))
需新增下面的程式碼:
yaml.load(fr,Loader=yaml.SafeLoader
(4)定義函式的程式碼:
import yaml def load_yaml(file_path): with open(file_path,encoding='utf-8') as fr: return yaml.load(fr,Loader=yaml.SafeLoader) if __name__ == "__main__": result = load_yaml('data.yaml') print(result)
(三)釘釘傳送訊息:加簽和驗籤
1、新增機器人:群設定-智慧群助手-新增機器人
(1)新增機器人時,安全設定可選擇:自定義關鍵詞 | 加簽 | IP地址
(2)安全設定選擇:自定義關鍵詞,利用postman進行傳送訊息,參考釘釘幫助文件:https://ding-doc.dingtalk.com/doc#/serverapi2/qf2nxq/d535db33
釘釘群獲得的訊息如下截圖:
2、安全設定選擇加簽方式,傳送釘釘訊息
(1)加簽,簽名:把timestamp+"\n"+金鑰當做簽名字串,使用HmacSHA256演算法計算簽名,然後進行Base64 encode,最後再把簽名引數再進行urlEncode,得到最終的簽名(需要使用UTF-8字符集)。
(2)url編碼進行加密和解密
import urllib from urllib.parse import quote,unquote print(quote('hello-123_你好')) secret = 'hello-123_%E4%BD%A0%E5%A5%BD' print(unquote(secret))
執行結果如下:
(3)生成加密,使用postman工具進行傳送訊息
import time import hashlib import base64 import hmac from urllib.parse import quote timestamp = int(time.time() * 1000) secret='SECf2f4947ab1160ccdb6040a88edb91be0427835976aa6dee6eee0969XXXXXXXX' sign_before = '%s\n%s' % (timestamp,secret) hsha256 = hmac.new(secret.encode(),sign_before.encode(),hashlib.sha256) # sha256 = hashlib.sha256(sign_before.encode()) sign_sha256 = hsha256.digest() sign_b64 =base64.b64encode(sign_sha256) sign = quote(sign_b64) print(timestamp,sign)
執行結果如下:
使用postman介面工具進行傳送訊息:
(4) 直接編碼傳送釘釘訊息
import hmac import time import hashlib import base64 from urllib.parse import quote import requests url = "https://oapi.dingtalk.com/robot/send?access_token=bd413385218506104a1903badd88016ba9ae9b7a1738bde9f7a3573aXXXXXXXX" def create_sign(): secret = 'SECf2f4947ab1160ccdb6040a88edb91be0427835976aa6dee6eee09691xxxxxxxx' timestamp = int(time.time() * 1000) sign_before = '%s\n%s' % (timestamp,secret) hsha265 = hmac.new(secret.encode(),sign_before.encode(),hashlib.sha256) sign_sha256 = hsha265.digest() sign_b64 = base64.b64encode(sign_sha256) sign = quote(sign_b64) return {"timestamp":timestamp,"sign":sign} def send_msg_dingding(msg="happy everyday!"): data = { "msgtype": "text", "text": { "content": msg }, "at": { "atMobiles": [ "1312007xxxx" ], "isAtAll": False } } sign = create_sign() r = requests.post(url,params = sign,json=data) print(r.json()) if __name__ == "__main__": send_msg_dingding("好好過好每一天!")
執行結果如下:
(四)寫自動化測試用例:Rainbow
1、彩虹-主架構設計
(1)分別建立目錄,如下圖:
圖A:
圖B:
(2)config_parse.py
import os import configparser import yaml from common.log import Log from config.settings import CONFIG_FILE,CASE_DATA_PATH def parse_ini(node,file_path=CONFIG_FILE): if not os.path.exists(file_path): Log.error("配置檔案不存在,檔案路徑{}",file_path) raise Exception('ini檔案不存在!') with open(file_path, encoding='utf-8') as fr: c = configparser.ConfigParser() c.read_file(fr) if node in c.sections(): result = dict(c[node]) return result Log.warning("配置檔案中[{}]節點不存在",node) def load_yaml(file_name): file_path = os.path.join(CASE_DATA_PATH,file_name) if not os.path.exists(file_path): Log.error("用例資料檔案不存在,檔案路徑{}",file_path) raise Exception('yaml檔案不存在!') with open(file_path,encoding='utf-8') as fr: return yaml.load(fr,Loader = yaml.SaveLoader) if __name__ == "__main__": parse_ini("mysql",'mysql.ini')
(3)log.py
from loguru import logger import sys from config.settings import LOG_FILE,LOG_LEVEL class Log: logger.remove() fmt = '[{time}][{level}][{file.path}:line:{line}:function_name:{function}] || msg={message}' # level file function module time message logger.add(sys.stdout,level=LOG_LEVEL,format=fmt) logger.add(LOG_FILE,level=LOG_LEVEL,format=fmt,encoding='utf-8',enqueue=True,rotation='1 day',retention='10 days') debug = logger.debug info = logger.info warning = logger.warning error = logger.error if __name__ == "__main__": Log.info("日誌測試")
(4)operate_db.py
import pymysql import traceback from common.log import Log class MySQL: def __init__(self,host,user,password,db,charset='utf8',autocommit=True,port=3306): port = int(port) self.conn = pymysql.connect(user=user,host=host,password=password,port=port, db=db,charset=charset,autocommit=autocommit) self.cursor = self.conn.cursor(pymysql.cursors.DictCursor) Log.info("開始連線mysql") def __del__(self): self.__close() def execute(self,sql): try: self.cursor.execute(sql) except Exception: Log.error('sql執行出錯,sql語句是{}',sql) Log.error(traceback.format_exc()) def fetchall(self,sql): self.execute(sql) return self.cursor.fetchall() def fetchone(self,sql): self.execute(sql) return self.cursor.fetchone() def bak_db(self): pass def __close(self): self.cursor.close() self.conn.close() if __name__ == "__main__": pass
(5)config.ini
[mysql] host=118.24.3.xx user=jxz password=123456 db=jxz charset=utf8 [mysql2] host=118.24.3.xxx user=jxz password=123456 db=jxz charset=utf8 port =3306 [redis] host=118.24.3.40 password=xxxx port=6379 [dingding] url = https://oapi.dingtalk.com/robot/send secret = SECf2f4947ab1160xxxxxxxxxxedb91be0427835976aa6dee6eee096xxxxxxxxxx access_token = bd4133852xxxxxxxxxxxx3badd88016ba9ae9b7a1738bde9f7a3573axxxxxxx at = 1312007xxxx [mail] host=smap.qq.com user=127xxxx070@qq.com password=1962xxxxzh to=krystal_xiao@126.com asc[email protected]
(6)settings.py
import os BASE_PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) LOG_FILE = os.path.join(BASE_PATH,'logs','rainbow.log') # 日誌檔案 REPORT_PATH = os.path.join(BASE_PATH,'report') # 報告存放的目錄 CASE_PATH = os.path.join(BASE_PATH,'biz','cases') # 測試用例的目錄 CASE_DATA_PATH = os.path.join(BASE_PATH,'biz','data') # 測試資料的目錄 CONFIG_FILE = os.path.join(BASE_PATH,'config','config.ini') # 配置檔案的目錄 LOG_LEVEL = 'INFO' # 預設日誌級別 if __name__ == "__main__": pass
2、彩虹-MySQL連線
(1)新建一個資料夾:
(2)utils.py
import redis import time import hmac import hashlib import base64 from urllib.parse import quote from common.config_parse import parse_ini from common.operate_db import MySQL ddconfig = parse_ini('dingding') # 取釘釘的配置檔案 secret = ddconfig.get("secret") mysql_conn_mapper = {} #{mysqlNone:Mysql,mysql2None:Mysql2} redis_mapper = {} def get_mysql(node='mysql',db=None): key = '%s%s'%(node,db) if not key in mysql_conn_mapper: mysql_info = parse_ini(node) if db: mysql_info['db']=db mysql = MySQL(**mysql_info) mysql_conn_mapper[key] = mysql else: mysql = mysql_conn_mapper[key] return mysql def get_redis(node='redis',db=None): key = '%s%s'%(node,db) if not key in redis_mapper: redis_info = parse_ini(node) if db: redis_info['db'] = db r = redis.Redis(**redis_info) redis_mapper[key] = r else: r = redis_mapper[key] return r def create_sign(): secret = 'SECf2f4947ab1160ccdbxxxxa88edb91be0427835976aa6dee6eee096912xxxxxx' timestamp = int(time.time() * 1000) sign_before = '%s\n%s' % (timestamp,secret) hsha265 = hmac.new(secret.encode(),sign_before.encode(),hashlib.sha256) sign_sha256 = hsha265.digest() sign_b64 = base64.b64encode(sign_sha256) sign = quote(sign_b64) return {"timestamp":timestamp,"sign":sign} if __name__ == "__main__": c = get_mysql()
3、NbDict
(1)新建一個檔案目錄
(2)custom_class.py
class NbDict(dict): def __getattr__(self, item): # {"login_time"} value = self.get(item) if type(value) == dict: value = NbDict(value) elif isinstance(value,list) or isinstance(value,tuple): value = list(value) for index,obj in enumerate(value): if isinstance(obj,dict): value[index] = NbDict(obj) return value if __name__ == "__main__": d = {"login_time":1} d1 = NbDict(d) print(d1.login_time2)
4、彩虹-封裝http請求類
(2)新建一個檔案目錄:
(2)http_request.py
import requests import traceback from common.log import Log from common.utils import create_sign from common.custom_class import NbDict class Requests: def __init__(self,url,params=None,data=None,headers=None,json=None,files=None): self.url = url self.params = params self.data = data self.headers = headers self.json = json self.files = files def _get_response(self): try: result = self.req.json() except Exception as e: return self.req.text else: result = NbDict(result) return result def get(self): Log.info("開始傳送get請求") Log.info("url:【{}】,params:【{}】,headers:{}",self.url,self.params,self.headers) try: self.req = requests.get(self.url,params=self.params,headers=self.headers,verify=False) except Exception as e: Log.error("http請求傳送錯誤,錯誤資訊:{}",traceback.format_exc()) raise Exception("介面請求不通") else: return self._get_response() def post(self): Log.info("開始傳送post請求") Log.info("url:【{}】,params:【{}】,headers:{},data:{},json:{},files:{}", self.url, self.params, self.headers,self.data,self.json,self.files) try: self.req = requests.post(self.url,params=self.params, data=self.data,json=self.json, files=self.files, # {"key":open("f.py",'rb')} headers=self.headers,verify=False) except Exception as e: Log.error("http請求傳送錯誤,錯誤資訊:{}", traceback.format_exc()) raise Exception("介面請求不通") else: return self._get_response() if __name__ == "__main__": url = "https://oapi.dingtalk.com/robot/send?access_token=bd41xxxxxxxxxxx104a190xxxxx88016ba9ae9b7a1738bde9f7a3573axxxxxxxx" sign = create_sign() msg = "Good Evening!" data = { "msgtype": "text", "text": { "content": msg }, "at": { "atMobiles": [ "13xxxxx1212" ], "isAtAll": False } } r = Requests(url,params=sign,json=data) result = r.post() print(result)
5、彩虹-封裝傳送訊息的方法
(1)新建一個目錄
(2) 程式碼如下:send_msg.py
import yamail from common.http_request import Requests from common.utils import parse_ini,create_sign ddconfig = parse_ini('dingding') # 取釘釘的配置資訊 mail_config =parse_ini('mail') # 取郵件的配置資訊 url = ddconfig.get('url') # 釘釘l access_token = ddconfig.get('access_token') # access_token at= ddconfig.get('at','').split(',') # 釘釘傳送訊息的時候at給誰 def send_dingding(msg): data = { "msgtype": "text", "text": { "content": msg }, "at": { "atMobiles": at, "isAtAll": False } } sign = create_sign() sign['access_token'] = access_token r = Requests(url,params=sign,json=data) r.post() def send_mail(subject,contents,attachments=None): smtp = yamail.SMTP( host=mail_config.get("host"), user = mail_config.get("user"), password =mail_config.get("password") ) smtp.sent(to=mail_config.get("to",'').split(','), # 傳送給誰 subject = subject, # 郵件主題 DD =mail_config.get("xxx",'').split(','), # 抄送 contents=contents, # 郵件正文 attachments=attachments #附件,如果是多個附件,寫list ) smtp.close() if __name__ == "__main__": send_dingding("good evening,guys!")
6、彩虹-完成
(1)在flow資料夾下面新建一個user.py檔案
程式碼如下:user.py
from common.http_request import Requests from biz.support.urls import ServerUrl class UserRequest: @classmethod def login(cls,username,password): ''' 呼叫登入介面的 :param username: 使用者名稱 :param password: 密碼 :return: ''' data = { 'username':username, 'passwd':password } req = Requests(ServerUrl.login_url,data=data) return req.post() @classmethod def register(cls,username,password,cpassword): ''' 註冊 :param username: 使用者名稱 :param password: 密碼 :param cpassword: 確認密碼 :return: ''' data = { 'username': username, 'pwd': password, 'cpwd':cpassword } req = Requests(ServerUrl.register_url,data=data) return req.post() if __name__ == "__main__": result = UserRequest.login('niuhanyang','aAxxxxxx') print(result)
執行結果如下:
(2)在data資料夾下面新建一個login_data.yaml檔案:
yaml檔案資訊如下:
username : niuhanyang
password : xxxxxxx
(3)在cases資料夾下新建3個檔案,分別為:base_case.py、test_login.py、test_open_acc.py
程式碼如下:
base_case.py
import unittest from common.config_parse import load_yaml class BaseCase(unittest.TestCase): data_file_name = None @property def file_data(self): data = load_yaml(self.data_file_name) return data @classmethod def get_token(cls,username): pass if __name__ == "__main__": pass
test_login.py:
import unittest from biz.flow.user import UserRequest from common.utils import get_redis,get_mysql from common.config_parse import load_yaml from biz.cases.base_case import BaseCase class TestLogin(BaseCase): '''登入介面測試用例''' data_file_name = 'login_data.yaml' @classmethod def setUpClass(cls): cls.redis = get_redis() cls.mysql = get_mysql() # cls.file_data = load_yaml('login_data.yaml') def test_normal(self): '''正常登入''' username = self.file_data.get('username') password = self.file_data.get('password') ret = UserRequest.login(username,password) self.assertEqual(0,ret.error_code,'返回的錯誤碼不是0') self.assertIsNotNone(ret.login_info.login_time,msg='logintime為空') redis_key = 'session:%s' % username sessionid = self.redis.get(redis_key) sql = 'select id from app_myuser where username = "%s";' % username db_result = self.mysql.fetchone(sql) user_id = db_result.get('id') self.assertEqual(sessionid,ret.login_info.sign,msg="返回的session和Redis中的不一致") self.assertEqual(user_id,ret.login_info.userId,msg="返回的userId和資料庫中的不一致") if __name__ == "__main__": pass
test_open_acc.py:
import unittest from common.config_parse import load_yaml class BaseCase(unittest.TestCase): data_file_name = None @property def file_data(self): data = load_yaml(self.data_file_name) return data @classmethod def get_token(cls,username): pass if __name__ == "__main__": pass
(4)執行run.py檔案
import unittest import os,time,sys BASE_PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.index(0,BASE_PATH) from config.settings import CASE_PATH,REPORT_PATH,dd_template,mail_template from common import send_msg from common.HTMLTestRunner import HTMLTestRunner def run(): test_suite = unittest.defaultTestLoader.discover(CASE_PATH,'test*.py') file_name = 'report_%s.html' % time.strftime('%Y%m%d%H%M%S') file_abs = os.path.join(REPORT_PATH,file_name) with open(file_abs,'wb') as fw: runner = HTMLTestRunner(stream=fw,title='測試報告標題',description='描述') case_result = runner.run(test_suite) all_count = case_result.failure_count + case_result.success_count dd_msg = dd_template % (all_count,case_result.success_count,case_result.failure_count) mail_msg = mail_template % (all_count, case_result.success_count, case_result.failure_count) send_msg.send_dingding(dd_msg) subject = '天馬座自動化測試報告-%s' % time.strftime('%Y-%m-%d %H:%M:%S') send_msg.send_mail(subject,mail_msg,file_abs) if __name__ == '__main__': run()
最後的執行結果:
7、自動化測試用例流程: