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

 1 #! /usr/bin/env python 
 2 #coding=utf-8 
 3 import threading 
 4 from multiprocessing import Queue 
 5 from time import ctime,sleep 
 6 from subprocess import Popen,PIPE 
 7 import os,time 
 8 lock=threading.Lock() 
 9 #单个测试用例生成的临时报告,当前目录下result\temp_年月日_时分秒\文件目录.html 
10 #例如 E:\python\selenium\fortest\result\temp_20160704_102822\E_python_selenium_fortest_test_1_test1_py.html 
11 def resultfile(tempdir,file): 
12     name=file.replace(‘\\‘,‘_‘).replace(‘:‘,‘‘).replace(‘.‘,‘_‘)+‘.html‘ 
13     return os.path.join(tempdir,name) 
14 class MyThread(threading.Thread): 
15     def __init__(self,queue,tempresultdir): 
16         threading.Thread.__init__(self) 
17         self.queue=queue 
18         self.tempresultdir=tempresultdir 
19     def run(self): 
20         while True: 
21             if not self.queue.empty(): 
22                 filename=self.queue.get() 
23                 lock.acquire() 
24                 resultname=resultfile(self.tempresultdir,filename) 
25                 cmd="python "+filename+" "+resultname 
26                 #print cmd 
27                 print ‘start time:%s‘ %ctime() 
28                 lock.release() 
29                 p=Popen(cmd,shell=True,stdout=PIPE) 
30                 #如果不加如下print,不会等待执行完毕 
31                 print p.stdout.readlines() 
32             else: 
33                 print ‘end‘ 
34                 break 
35 #获取路径下test开头的文件夹下以test开头.py结尾的文件 
36 def getfile(path): 
37     paths=[] 
38     for p in os.listdir(path): 
39         if p[0:4]==‘test‘ and os.path.isdir(p): 
40             paths.append(p) 
41     file=[] 
42     for p in paths: 
43         temp=os.path.join(path,p) 
44         #print temp 
45         files=os.listdir(temp) 
46         #print files 
47         for f in files: 
48             if f[0:4]==‘test‘ and f[-3:]==‘.py‘: 
49                 file.append(os.path.join(temp,f)) 
50     return file 
51 if __name__==‘__main__‘: 
52     print ‘main start time:%s‘ %ctime() 
53     tempresultdir=os.path.join(os.getcwd(),"result","temp"+time.strftime("_%Y%m%d_%H%M%S",time.localtime(time.time()))) 
54     os.mkdir(tempresultdir) 
55     resultreport=os.path.join(os.getcwd(),"result"+time.strftime("_%Y%m%d_%H%M%S",time.localtime(time.time()))) 
56     allfile=getfile(os.getcwd()) 
57     queue=Queue() 
58     for file in allfile: 
59         queue.put(file) 
60     my_Threads=[] 
61     my_Thread=threading.Thread() 
62     for i in range(2): 
63         my_Thread=MyThread(queue,tempresultdir) 
64         my_Thread.deamon=True 
65         my_Threads.append(my_Thread) 
66         my_Thread.start() 
67     for t in my_Threads: 
68         t.join()
69 
70     reports=os.listdir(tempresultdir) 
71     print reports 
72     print ‘main end time:%s‘ %ctime()

 1 # -*- coding: utf-8 -*- 
 2 import unittest,time,sys 
 3 sys.path.append("..") 
 4 from public import HTMLTestRunner 
 5 class test(unittest.TestCase): 
 6     def setUp(self): 
 7         pass 
 8     def test_login(self): 
 9         u"""test_1 test1 登录用例login""" 
10         time.sleep(5) 
11         pass 
12     def test_process(self): 
13         u"""test_1 test1 处理用例login""" 
14         time.sleep(5) 
15         assertEqual(1,2) 
16     def test_quit(self): 
17         u"""test_1 test1 登录用例quit""" 
18         time.sleep(5) 
19         self.assertEqual(1,2) 
20     def tearDown(self): 
21         pass 
22 if  __name__==‘__main__‘: 
23     suit=unittest.TestSuite() 
24     filename=sys.argv[1] 
25     suit.addTest(unittest.makeSuite(test)) 
26     f=open(filename,"wb") 
27     runner = HTMLTestRunner.HTMLTestRunner( 
28              stream=f, 
29              title=u‘测试报告‘, 
30              description=u‘测试结果‘) 
31     runner.run(suit) 

E:\python\selenium\fortest>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:\python\selenium\fortest>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