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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
|
# Copyright (C) 2013 Intel Corporation
#
# Released under the MIT license (see COPYING.MIT)
# Main unittest module used by testimage.bbclass
# This provides the oeRuntimeTest base class which is inherited by all tests in meta/lib/oeqa/runtime.
# It also has some helper functions and it's responsible for actually starting the tests
import os, re, mmap
import unittest
import inspect
import subprocess
import datetime
import bb
from oeqa.utils.decorators import LogResults
def loadTests(tc, type="runtime"):
if type == "runtime":
# set the context object passed from the test class
setattr(oeTest, "tc", tc)
# set ps command to use
setattr(oeRuntimeTest, "pscmd", "ps -ef" if oeTest.hasPackage("procps") else "ps")
# prepare test suite, loader and runner
suite = unittest.TestSuite()
elif type == "sdk":
# set the context object passed from the test class
setattr(oeTest, "tc", tc)
testloader = unittest.TestLoader()
testloader.sortTestMethodsUsing = None
suites = [testloader.loadTestsFromName(name) for name in tc.testslist]
def getTests(test):
'''Return all individual tests executed when running the suite.'''
# Unfortunately unittest does not have an API for this, so we have
# to rely on implementation details. This only needs to work
# for TestSuite containing TestCase.
method = getattr(test, '_testMethodName', None)
if method:
# leaf case: a TestCase
yield test
else:
# Look into TestSuite.
tests = getattr(test, '_tests', [])
for t1 in tests:
for t2 in getTests(t1):
yield t2
# Determine dependencies between suites by looking for @skipUnlessPassed
# method annotations. Suite A depends on suite B if any method in A
# depends on a method on B.
for suite in suites:
suite.dependencies = []
suite.depth = 0
for test in getTests(suite):
methodname = getattr(test, '_testMethodName', None)
if methodname:
method = getattr(test, methodname)
depends_on = getattr(method, '_depends_on', None)
if depends_on:
for dep_suite in suites:
if depends_on in [getattr(t, '_testMethodName', None) for t in getTests(dep_suite)]:
if dep_suite not in suite.dependencies and \
dep_suite is not suite:
suite.dependencies.append(dep_suite)
break
else:
bb.warn("Test %s was declared as @skipUnlessPassed('%s') but that test is either not defined or not active. Will run the test anyway." %
(test, depends_on))
# Use brute-force topological sort to determine ordering. Sort by
# depth (higher depth = must run later), with original ordering to
# break ties.
def set_suite_depth(suite):
for dep in suite.dependencies:
new_depth = set_suite_depth(dep) + 1
if new_depth > suite.depth:
suite.depth = new_depth
return suite.depth
for index, suite in enumerate(suites):
set_suite_depth(suite)
suite.index = index
suites.sort(cmp=lambda a,b: cmp((a.depth, a.index), (b.depth, b.index)))
return testloader.suiteClass(suites)
def runTests(tc, type="runtime"):
suite = loadTests(tc, type)
bb.note("Test modules %s" % tc.testslist)
bb.note("Found %s tests" % suite.countTestCases())
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)
return result
@LogResults
class oeTest(unittest.TestCase):
longMessage = True
@classmethod
def hasPackage(self, pkg):
if re.search(pkg, oeTest.tc.pkgmanifest):
return True
return False
@classmethod
def hasFeature(self,feature):
if feature in oeTest.tc.imagefeatures or \
feature in oeTest.tc.distrofeatures:
return True
else:
return False
class oeRuntimeTest(oeTest):
def __init__(self, methodName='runTest'):
self.target = oeRuntimeTest.tc.target
super(oeRuntimeTest, self).__init__(methodName)
def tearDown(self):
# If a test fails or there is an exception
if (self._resultForDoCleanups.failures or
self._resultForDoCleanups.errors):
self.dump_target_logs()
def dump_target_logs(self):
commands = ["top -bn1", "ps", "free", "df", "_ping", "dmesg", "netstat -a", "ifconfig -a", "_logs"]
dump_dir = "/tmp/oe-saved-tests"
dump_sub_dir = ("%s_%s" % (
datetime.datetime.now().strftime('%Y%m%d%H%M'),
self._testMethodName))
dump_dir = os.path.join(dump_dir, dump_sub_dir)
os.makedirs(dump_dir)
bb.warn("%s failed: getting data from target and "
"saving into %s" % (self._testMethodName, dump_dir))
for command in commands:
# This will ping the host from target
if command == "_ping":
comm = "ping -c3 %s" % self.target.server_ip
# This will get all the logs from /var/log/
elif command == "_logs":
comm = 'find /var/log/ -type f 2>/dev/null '
comm = '%s-exec echo "%s" \\; ' % (comm, '='*20)
comm = '%s-exec echo {} \\; ' % comm
comm = '%s-exec echo "%s" \\; ' % (comm, '='*20)
comm = '%s-exec cat {} \\; -exec echo "" \\;' % comm
else:
comm = command
(status, output) = self.target.run_serial(comm)
filename = command.split()[0]
with open(os.path.join(dump_dir, filename), 'w') as f:
f.write(output)
#TODO: use package_manager.py to install packages on any type of image
def install_packages(self, packagelist):
for package in packagelist:
(status, result) = self.target.run("smart install -y "+package)
if status != 0:
return status
class oeSDKTest(oeTest):
def __init__(self, methodName='runTest'):
self.sdktestdir = oeSDKTest.tc.sdktestdir
super(oeSDKTest, self).__init__(methodName)
@classmethod
def hasHostPackage(self, pkg):
if re.search(pkg, oeTest.tc.hostpkgmanifest):
return True
return False
def _run(self, cmd):
return subprocess.check_output(cmd, shell=True)
def getmodule(pos=2):
# stack returns a list of tuples containg frame information
# First element of the list the is current frame, caller is 1
frameinfo = inspect.stack()[pos]
modname = inspect.getmodulename(frameinfo[1])
#modname = inspect.getmodule(frameinfo[0]).__name__
return modname
def skipModule(reason, pos=2):
modname = getmodule(pos)
if modname not in oeTest.tc.testsrequired:
raise unittest.SkipTest("%s: %s" % (modname, reason))
else:
raise Exception("\nTest %s wants to be skipped.\nReason is: %s" \
"\nTest was required in TEST_SUITES, so either the condition for skipping is wrong" \
"\nor the image really doesn't have the required feature/package when it should." % (modname, reason))
def skipModuleIf(cond, reason):
if cond:
skipModule(reason, 3)
def skipModuleUnless(cond, reason):
if not cond:
skipModule(reason, 3)
|
