PYTHON+UNITTEST自动发送报告

上一篇 / 下一篇 2018-05-09 17:50:09 / 个人分类：python

查看( 1777 ) / 评论( 0 ) / 评分( 0 / 0 )

1.文件目录

case：存放测试用例格式：test1.py

common：存放公共方法

config：存放配置文件

logs：存放日志文件

test：可以不用管

主要程序：run_main.py

# encoding: utf-8
import sys
reload(sys)
import os
from selenium import webdriver
import unittest
import time
from pip import runner
import HTMLTestRunner
from smtpd import SMTPServer
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import loggerrrr

#import readConfig

cur_path=os.path.dirname(os.path.realpath(__file__))

####################################################
#加载所有用例
####################################################
def add_case(caseName="case",rule="*.py"):
    #加载所有用例
    case_path=os.path.join(cur_path,caseName)

    if not os.path.exists(case_path):os.mkdir(case_path)
    print ("test case path :",case_path)
    #loader=unittest.TestLoader()
    #discover=loader.discover(case_path)
    discover=unittest.defaultTestLoader.discover(case_path, pattern=rule, top_level_dir=None)

    print(discover)
    return discover

####################################################
#生成报告
####################################################
def run_case(all_case,reportName="report"):
    now=time.strftime("%Y_%m_%d_%H_%M_%S_")
    report_path=os.path.join(cur_path,reportName)
    if not os.path.exists(report_path):os.mkdir(report_path)
    report_abspath=os.path.join(report_path,now+"result.html")
    print("report path :",report_abspath)
    fp=open(report_abspath,"wb")

    runner=HTMLTestRunner.HTMLTestRunner(stream=fp,
                                         title=u'自动化测试报告，测试结果如下：',
                                         description=u'用例执行情况：'
        )

    runner.run(all_case)

    fp.close()

####################################################
#获取最新的报告
####################################################
def get_report_file(report_path):
    lists=os.listdir(report_path)
    lists.sort(key=lambda fn:os.path.getatime(os.path.join(report_path,fn)))
    print(u'最新测试生成的报告：'+lists[-1])
    report_file=os.path.join(report_path,lists[-1])
    return report_file

####################################################
#发送**�
####################################################
def send_mail(sender,psw,receiver,smtpserver,report_file,port):
    to_addrs = receiver.split(',')
    print "to_addrs=",to_addrs
    with open(report_file,"rb") as f:
        mail_body=f.read()
    msg = MIMEMultipart()
    body=MIMEText(mail_body,_subtype='html',_charset='utf-8')
    msg['Subject']=u"自动化测试报告"
    msg["from"]=sender
    msg['To'] = ",".join(to_addrs)
    msg.attach(body)

    att=MIMEText(open(report_file,"rb").read(),"base64","utf-8")
    att["Content-Type"]="application/octet-stream"
    att["Content-Disposition"]='attachment;filename="report.html"'
    msg.attach(att)

    try:
        smtp=smtplib.SMTP_SSL(smtpserver,port)
    except:
        smtp=smtplib.SMTP()
        smtp.connect(smtpserver, port)

    smtp.login(sender, psw)
    print to_addrs
    #smtp.sendmail(sender, to_addrs, msg.as_string())
    smtp.quit()
    print ('test report email has send out!')



if __name__=="__main__":
    loggerrrr.Log().info('开始测试')
    all_case=add_case()
    loggerrrr.Log().info('加载完成')
    run_case(all_case)
    loggerrrr.Log().info('开始生成报告')
    report_path=os.path.join(cur_path,"report")
    loggerrrr.Log().info('开始发邮件')
    report_file=get_report_file(report_path)
   sender='****@qq.com'
    psw='pqdddhexlokbbjeh'
    smtp_server='smtp.qq.com'
    port='465'
    #一个人
   receiver='******@qq.com'
    #多个人
   receiver='****@qq.com,****@qq.com,****@163.com,****@163.com'
    send_mail(sender, psw, receiver, smtp_server, report_file, port)

logs——loggerrr

# coding:utf-8
import logging, time
import os
# log_path是存放日志的路径
cur_path = os.path.dirname(os.path.realpath(__file__))
log_path = os.path.join(os.path.dirname(cur_path), 'logs')
# 如果不存在这个logs文件夹，就自动创建一个
if not os.path.exists(log_path):os.mkdir(log_path)

class Log():
    def __init__(self):
        # 文件的命名
        self.logname = os.path.join(log_path, '%s.log'%time.strftime('%Y_%m_%d'))
        self.logger = logging.getLogger()
        self.logger.setLevel(logging.DEBUG)
        # 日志输出格式
        self.formatter = logging.Formatter('[%(asctime)s] - %(filename)s] - %(levelname)s: %(message)s')

    def __console(self, level, message):
        # 创建一个FileHandler，用于写到本地
        fh = logging.FileHandler(self.logname, 'a') # 追加模式这个是python2的
        # fh = logging.FileHandler(self.logname, 'a', encoding='utf-8') # 这个是python3的
        fh.setLevel(logging.DEBUG)
        fh.setFormatter(self.formatter)
        self.logger.addHandler(fh)

        # 创建一个StreamHandler,用于输出到控制台
        ch = logging.StreamHandler()
        ch.setLevel(logging.DEBUG)
        ch.setFormatter(self.formatter)
        self.logger.addHandler(ch)

        if level == 'info':
            self.logger.info(message)
        elif level == 'debug':
            self.logger.debug(message)
        elif level == 'warning':
            self.logger.warning(message)
        elif level == 'error':
            self.logger.error(message)
        # 这两行代码是为了避免日志输出重复问题
        self.logger.removeHandler(ch)
        self.logger.removeHandler(fh)
        # 关闭打开的文件
        fh.close()

def debug(self, message):
self.__console('debug', message)

def info(self, message):
self.__console('info', message)

def warning(self, message):
self.__console('warning', message)

def error(self, message):
self.__console('error', message)

if __name__ == "__main__":
   log = Log()
   log.info("---测试开始----")
   log.info("操作步骤1,2,3")
   log.info("操作步骤hahah")
   log.warning("----测试结束----")

config：readconfigdd.py

# encoding: utf-8
import os
import ConfigParser
from run_main import cur_path, sender

cur_path=os.path.dirname(os.path.realpath(__file__))
configPath=os.path.join(cur_path,"cfg.ini")
conf=ConfigParser.ConfigParser()
conf.read(configPath)

smtp_server=conf.get("email", "smtp_server")
sender=conf.get("email","sender")
psw=conf.get("email","psw")
receiver=conf.get("email","receiver")
sender=conf.get("email","sender")
port=conf.get("email","port")

case:test.py

# encoding: utf-8
#如果填写中文需要加上这句话
import unittest
import loggerrrr
from requests.api import request
import requests

import sys

global accessToken1
#用来存放令牌=
userid=""
username=""
accessToken1="eFljQmFsZXZjeXVUZ0hEcXZW"
class test_log(unittest.TestCase):



    def test_a_addAuthUser(self):
        #loggerrrr.Log().info("begin addAuthUser")
        url1="http://train.test.*****100.com/train-gateway/admin/api/authUser/addAuthUser"
        data1 = {
            "accessToken":accessToken1,
            "account":"test03",
            "username":"test03",
            "password":"c4ca4238a0b923820dcc509a6f75849b",
            "phoneNo":13640281725,
            "authRoleId":1,
            "areaId":1
        }
        username=data1['account']
        #loggerrrr.Log().info("addAuthUser调用接口")
        web1 = requests.post(url=url1, params=data1)
        loggerrrr.Log().info(web1.text)
        if web1.status_code == 200:
                loggerrrr.Log().info( data1['account'])
                loggerrrr.Log().info("end addAuthUser")

        else:
                print web1.status_code,web1.text

    def test_e_deleteAuthUser(self):
        #loggerrrr.Log().info("findAuthUserPage begin")
        url2="http://train.test.*****100.com/train-gateway/admin/api/authUser/findAuthUserPage"
        data2 = {
            "accessToken":accessToken1,
            "username":username,
            "status":"",
            "areaId":"",
            "pageIndex":1,
            "pageSize":10,
            "isTrusted":"true"
        }
        web2 = requests.post(url=url2, params=data2)
            #get id
        if web2.status_code == 200:
                ty=web2.text
                c=web2.json()
                for value in c.values():
                    if isinstance(value, dict):
                        for value2 in value.values():
                            if isinstance(value2, list):
                                loggerrrr.Log().info(value2[0])
                                userid=value2[0]['id']
                                #loggerrrr.Log().info("findAuthUserPage end")

        else:
                print web2.status_code,web2.text

        loggerrrr.Log().info("deleteUser删除开始")
        url3="http://train.test.*****100.com/train-gateway/admin/api/authUser/deleteAuthUser"

        data3 = {
            "accessToken":accessToken1,
            "id":userid
        }
        loggerrrr.Log().info(userid)
        web3 = requests.post(url=url3, params=data3)

        if web3.status_code == 200:
                print "delete success.delete user",username
                loggerrrr.Log().info("deleteUser删除完成")

        else:
                print web3.status_code,web3.text




if __name__=='__main__':
    unittest.main()

收藏举报

TAG:

查看全部评论

PYTHON+UNITTEST自动发送报告

用户菜单

我的栏目

标题搜索

日历

我的存档

数据统计

RSS订阅