Python 多线程执行测试用例并生成多个报告

为了满足之前的需求,同时运行多个测试用例.

#! /usr/bin/env python 
#coding=utf-8 
import threading 
from multiprocessing import Queue 
from time import ctime,sleep 
from subprocess import Popen,PIPE 
import os,time 
lock=threading.Lock() 
#单个测试用例生成的临时报告,当前目录下result	emp_年月日_时分秒文件目录.html 
#例如 E:pythonseleniumfortest
esult	emp_20160704_102822E_python_selenium_fortest_test_1_test1_py.html 
def resultfile(tempdir,file): 
    name=file.replace('\','_').replace(':','').replace('.','_')+'.html' 
    return os.path.join(tempdir,name) 
class MyThread(threading.Thread): 
    def __init__(self,queue,tempresultdir): 
        threading.Thread.__init__(self) 
        self.queue=queue 
        self.tempresultdir=tempresultdir 
    def run(self): 
        while True: 
            if not self.queue.empty(): 
                filename=self.queue.get() 
                lock.acquire() 
                resultname=resultfile(self.tempresultdir,filename) 
                cmd="python "+filename+" "+resultname 
                #print cmd 
                print 'start time:%s' %ctime() 
                lock.release() 
                p=Popen(cmd,shell=True,stdout=PIPE) 
                #如果不加如下print,不会等待执行完毕 
                print p.stdout.readlines() 
            else: 
                print 'end' 
                break 
#获取路径下test开头的文件夹下以test开头.py结尾的文件 
def getfile(path): 
    paths=[] 
    for p in os.listdir(path): 
        if p[0:4]=='test' and os.path.isdir(p): 
            paths.append(p) 
    file=[] 
    for p in paths: 
        temp=os.path.join(path,p) 
        #print temp 
        files=os.listdir(temp) 
        #print files 
        for f in files: 
            if f[0:4]=='test' and f[-3:]=='.py': 
                file.append(os.path.join(temp,f)) 
    return file 
if __name__=='__main__': 
    print 'main start time:%s' %ctime() 
    tempresultdir=os.path.join(os.getcwd(),"result","temp"+time.strftime("_%Y%m%d_%H%M%S",time.localtime(time.time()))) 
    os.mkdir(tempresultdir) 
    resultreport=os.path.join(os.getcwd(),"result"+time.strftime("_%Y%m%d_%H%M%S",time.localtime(time.time()))) 
    allfile=getfile(os.getcwd()) 
    queue=Queue() 
    for file in allfile: 
        queue.put(file) 
    my_Threads=[] 
    my_Thread=threading.Thread() 
    for i in range(2): 
        my_Thread=MyThread(queue,tempresultdir) 
        my_Thread.deamon=True 
        my_Threads.append(my_Thread) 
        my_Thread.start() 
    for t in my_Threads: 
        t.join()

    reports=os.listdir(tempresultdir) 
    print reports 
    print 'main end time:%s' %ctime()

单个测试用例写法如下,只是示例:

# -*- coding: utf-8 -*- 
import unittest,time,sys 
sys.path.append("..") 
from public import HTMLTestRunner 
class test(unittest.TestCase): 
    def setUp(self): 
        pass 
    def test_login(self): 
        u"""test_1 test1 登录用例login""" 
        time.sleep(5) 
        pass 
    def test_process(self): 
        u"""test_1 test1 处理用例login""" 
        time.sleep(5) 
        assertEqual(1,2) 
    def test_quit(self): 
        u"""test_1 test1 登录用例quit""" 
        time.sleep(5) 
        self.assertEqual(1,2) 
    def tearDown(self): 
        pass 
if  __name__=='__main__': 
    suit=unittest.TestSuite() 
    filename=sys.argv[1] 
    suit.addTest(unittest.makeSuite(test)) 
    f=open(filename,"wb") 
    runner = HTMLTestRunner.HTMLTestRunner( 
             stream=f, 
             title=u'测试报告', 
             description=u'测试结果') 
    runner.run(suit) 

执行的结果:

E:pythonseleniumfortest>thread.py 
main start time:Tue Jul 05 09:31:12 2016 
start time:Tue Jul 05 09:31:12 2016 
start time:Tue Jul 05 09:31:12 2016 
.. 
Time Elapsed: 0:00:04.001000 
[] 
start time:Tue Jul 05 09:31:17 2016 
.EF 
Time Elapsed: 0:00:15.004000 
[] 
start time:Tue Jul 05 09:31:28 2016 
.. 
Time Elapsed: 0:00:04 
.[] 
start time:Tue Jul 05 09:31:32 2016 
. 
Time Elapsed: 0:00:20.002000 
[] 
start time:Tue Jul 05 09:31:37 2016 
.. 
Time Elapsed: 0:00:04 
[] 
end 
.. 
Time Elapsed: 0:00:30.003000 
[] 
end 
['E_python_selenium_fortest_test_1_test1_py.html', 'E_python_selenium_fortest_te 
st_1_test2_py.html', 'E_python_selenium_fortest_test_2_test1_py.html', 'E_python 
_selenium_fortest_test_2_test2_py.html', 'E_python_selenium_fortest_test_3_test1 
_py.html', 'E_python_selenium_fortest_test_3_test2_py.html'] 
main end time:Tue Jul 05 09:32:02 2016 

发现一个问题,如果去掉代码中的 

print 'start time:%s' %ctime()

两个线程都执行完成后才会继续执行

E:pythonseleniumfortest>thread.py 
main start time:Tue Jul 05 09:33:40 2016 
.. 
Time Elapsed: 0:00:04.001000 
.EF 
Time Elapsed: 0:00:15.004000 
[[]]

.. 
Time Elapsed: 0:00:04 
.. 
Time Elapsed: 0:00:20.002000 
[[] 
] 
.. 
Time Elapsed: 0:00:04 
[] 
end 
.. 
Time Elapsed: 0:00:30.004000 
[] 
end 
['E_python_selenium_fortest_test_1_test1_py.html', 'E_python_selenium_fortest_te 
st_1_test2_py.html', 'E_python_selenium_fortest_test_2_test1_py.html', 'E_python 
_selenium_fortest_test_2_test2_py.html', 'E_python_selenium_fortest_test_3_test1 
_py.html', 'E_python_selenium_fortest_test_3_test2_py.html'] 
main end time:Tue Jul 05 09:34:47 2016 

目录结构:

│  testlibmain.py 
│  thread.py 
│  threadtest.py 
│  
├─public 
│      HTMLTestRunner.py 
│      __init__.py 
│      
├─result
│  └─temp_20160704_102848 
│          E_python_selenium_fortest_test_1_test1_py.html 
│          E_python_selenium_fortest_test_1_test2_py.html 
│          E_python_selenium_fortest_test_2_test1_py.html 
│          E_python_selenium_fortest_test_2_test2_py.html 
│          E_python_selenium_fortest_test_3_test1_py.html 
│          E_python_selenium_fortest_test_3_test2_py.html 
│          
├─test_1 
│      test1.py 
│      test2.py 
│      __init__.py 
│      
├─test_2 
│      test1.py 
│      test2.py 
│      __init__.py 
│      
└─test_3 
        test1.py 
        test2.py 
        __init__.py

其中的一份测试报告:
       

上面的代码中没有对webdriver的地址进行参数绑定,后期可以再加入

生成的多个报告文件最终需要合并成一个,后续再写