1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
# Copyright (C) 2017 Intel Corporation
# Released under the MIT license (see COPYING.MIT)
import threading
import multiprocessing
from unittest.suite import TestSuite
from oeqa.core.loader import OETestLoader
from oeqa.core.runner import OEStreamLogger
class OETestLoaderThreaded(OETestLoader):
def __init__(self, tc, module_paths, modules, tests, modules_required,
filters, process_num=0, *args, **kwargs):
super(OETestLoaderThreaded, self).__init__(tc, module_paths, modules,
tests, modules_required, filters, *args, **kwargs)
self.process_num = process_num
def discover(self):
suite = super(OETestLoaderThreaded, self).discover()
if self.process_num <= 0:
self.process_num = min(multiprocessing.cpu_count(),
len(suite._tests))
suites = []
for _ in range(self.process_num):
suites.append(self.suiteClass())
def _search_for_module_idx(suites, case):
"""
Cases in the same module needs to be run
in the same thread because PyUnit keeps track
of setUp{Module, Class,} and tearDown{Module, Class,}.
"""
for idx in range(self.process_num):
suite = suites[idx]
for c in suite._tests:
if case.__module__ == c.__module__:
return idx
return -1
def _search_for_depend_idx(suites, depends):
"""
Dependency cases needs to be run in the same
thread, because OEQA framework look at the state
of dependant test to figure out if skip or not.
"""
for idx in range(self.process_num):
suite = suites[idx]
for case in suite._tests:
if case.id() in depends:
return idx
return -1
def _get_best_idx(suites):
sizes = [len(suite._tests) for suite in suites]
return sizes.index(min(sizes))
def _fill_suites(suite):
idx = -1
for case in suite:
if isinstance(case, TestSuite):
_fill_suites(case)
else:
idx = _search_for_module_idx(suites, case)
depends = {}
if 'depends' in self.tc._registry:
depends = self.tc._registry['depends']
if idx == -1 and case.id() in depends:
case_depends = depends[case.id()]
idx = _search_for_depend_idx(suites, case_depends)
if idx == -1:
idx = _get_best_idx(suites)
suites[idx].addTest(case)
_fill_suites(suite)
suites_tmp = suites
suites = []
for suite in suites_tmp:
if len(suite._tests) > 0:
suites.append(suite)
return suites
class OEStreamLoggerThreaded(OEStreamLogger):
_lock = threading.Lock()
buffers = {}
def write(self, msg):
tid = threading.get_ident()
if not tid in self.buffers:
self.buffers[tid] = ""
if msg:
self.buffers[tid] += msg
def finish(self):
tid = threading.get_ident()
self._lock.acquire()
self.logger.info('THREAD: %d' % tid)
self.logger.info('-' * 70)
for line in self.buffers[tid].split('\n'):
self.logger.info(line)
self._lock.release()
|