´úÂë¼ûͼ

ÏÈ´´½¨Á¬½Ó£¬»ñÈ¡Êý¾ÝºóתÂë½ØÈ¡×Ö·û´®¡£

Ö÷ÒªÓÐÈý¸ö·½·¨

getLogData£¨£©ÓÃÓÚ»ñÈ¡´íÎóÈÕÖ¾ÐÅÏ¢²¢´æÈëÎĵµ

run£¨£©ÊÇÖ÷³ÌÐò£¬ÓÃÓÚÂÖѯÇëÇó

eachFile£¨£©ÓÃÓÚ±éÀúÎļþ¼Ð

²é¿´(799) ÆÀÂÛ(0) ÊÕ²Ø ·ÖÏí ¹ÜÀí

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()