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
203
204
205
206
207
208
|
#!/usr/bin/env python
import subprocess
import re
import os
from avocado import Test
from avocado import main
SCREENSHOOTER_PKG_NAME='screenshooter'
# All possible paths where screenshooter's rpm can be found.
# name parameter in fetch_asset will assure that a proper one is chosen.
locations=['file:///opt/screenshooter/screenshooter.armv7l.rpm']
class SSH(object):
def __init__(self, ssh_config, ip):
self.ssh_config = ssh_config
self.ip = ip
def run(self, cmd):
"""
cmd -> subprocess.Popen
"""
return subprocess.Popen(["sshpass", "-p", "tizen", "ssh", "-F", self.ssh_config, "root@" + self.ip] + cmd,
stdout=subprocess.PIPE)
def pull(self, src, dest):
"""
src, dest -> returncode (int)
"""
return subprocess.call(["sshpass", "-p", "tizen", "scp", "-F", self.ssh_config,
"root@" + self.ip + ":" + src, dest], stdout=subprocess.PIPE)
def push(self, src, dest):
"""
src, dest -> returncode (int)
"""
return subprocess.call(["sshpass", "-p", "tizen", "scp", "-F", self.ssh_config,
src, "root@" + self.ip + ":" + dest], stdout=subprocess.PIPE)
class SDB(object):
def __init__(self, serial):
self.serial = serial
def run(self, cmd):
"""
cmd -> subprocess.Popen
"""
return subprocess.Popen(["sdb", "-s", self.serial, "shell"] + cmd, stdout=subprocess.PIPE)
def pull(self, src, dest):
"""
src, dest -> returncode (int)
"""
return subprocess.call(["sdb", "-s", self.serial, "pull", src, dest], stdout=subprocess.PIPE)
def push(self, src, dest):
"""
src, dest -> returncode (int)
"""
return subprocess.call(["sdb", "-s", self.serial, "push", src, dest], stdout=subprocess.PIPE)
class Screenshot(object):
def __init__(self, comm, filename, outputdir):
self.comm = comm
self.filename = filename
self.outputdir = outputdir
def take(self):
# False, output - screenshooter failed
# True, output - screenshot taken and downloaded
# False, None - screenshot download failed
remote_path = os.path.join("/tmp", self.filename)
local_path = os.path.join(self.outputdir, self.filename)
ret_scr = self.comm.run(["XDG_RUNTIME_DIR=/run", "screenshooter", "-f", remote_path])
if ret_scr.wait() != 0: # SDB won't
return False, ret_scr.stdout.read()
ret_pull = self.comm.pull(remote_path, local_path)
if ret_pull != 0:
return False, None
return True, ret_scr.stdout.read()
class CommonTest(Test):
default_params = {'timeout': 60.0}
def __init__(self, methodName='test', name=None, params=None,
base_logdir=None, job=None, runner_queue=None):
super(CommonTest, self).__init__(methodName, name, params,
base_logdir, job,
runner_queue)
def setUp(self):
self.log.debug("setUp")
ssh_config = self.params.get('ssh_config', "/*")
ip = self.params.get('ip', "/*")
if ip and ip != 'IP_VALUE':
self.comm = SSH(ssh_config, ip)
return
serial = self.params.get('serial', "/*")
if serial and serial != 'SERIAL_VALUE':
self.comm = SDB(serial)
return
self.error("No communication channel specified")
def test(self):
test = self.params.get('test', "/*/common/*", default=None)
if test == 'service':
self.service_test()
elif test == 'systemctl':
self.systemctl_test()
elif test == 'crash':
self.crash_test()
elif test == 'screenshot':
self.screenshot_test()
else:
self.error("Unknown test: %s" % test)
def screenshot(self, name):
def install(pkg):
# True - screenshooter installed successfully
# False - otherwise
rpm = pkg + '.rpm'
path = self.fetch_asset(name=rpm, locations=locations)
if path == None:
self.error("Asset fetch failed: %s" % rpm)
return False
remote_path = os.path.join("/tmp", rpm)
if self.comm.push(path, remote_path) != 0:
self.log.error("Push failed %s %s" % (path, remote_path))
return False
ret = self.comm.run(["rpm", "-i", remote_path])
if ret.wait() != 0:
self.log.error("Install failed")
self.log.debug(ret.stdout.read())
return False
return True
def is_installed():
# True, pkg - screenshooter is installed
# False, pkg - otherwise
ret = self.comm.run(["uname", "-m"])
self.assertTrue(ret.wait() == 0, "Architecture check failed")
arch = ret.stdout.read().splitlines()[0]
self.assertTrue(arch in ["armv7l", "aarch64", "x86_64", "i686"], "Unknown arch: " + arch)
pkg = SCREENSHOOTER_PKG_NAME + '.' + arch
ret = self.comm.run(["rpm", "-q", SCREENSHOOTER_PKG_NAME])
if ret.wait() == 0 and re.match(SCREENSHOOTER_PKG_NAME, ret.stdout.read()):
return True, pkg
return False, pkg
ret, pkg = is_installed()
if not ret:
self.assertTrue(install(pkg), "Screenshot install failed")
ret, out = Screenshot(self.comm, name + ".png", self.outputdir).take()
self.log.debug("outputdir: %s" % self.outputdir)
self.log.debug(out)
self.assertTrue(ret, "Screenshot failed")
def screenshot_test(self):
self.log.debug("screeenshot_test")
name = self.params.get('name', "/*/common/*", default="initial")
self.screenshot(name)
def systemctl_test(self):
self.log.debug("systemctl_test")
ref_cnt = self.params.get('ref_cnt', "/*/common/*", default=-1)
ref_services = self.params.get('ignored_services', "/*/common/*", default=[])
state = self.params.get('state', "/*/common/*", default='failed')
ret = self.comm.run(["systemctl", "--state=" + state])
self.assertTrue(ret.wait() == 0, "Command failed")
out = ret.stdout.read()
fail_num = re.search(r'(\d+) loaded units listed', out)
# Save to whiteboard
self.whiteboard = out
int_failed_num = int(fail_num.group(1))
services = re.findall(r'([\w_-]+)\.service', out)
for service in services:
# SDB issue workaround
if service.startswith('31m'):
service = re.sub('31m', '', service)
self.assertTrue(service in ref_services, "Unexpected service failed (%s)" % service)
if ref_cnt != -1:
self.assertTrue(int_failed_num <= ref_cnt, "More services failed (%d)" % int_failed_num)
def service_test(self):
self.log.debug("service_test")
service = self.params.get('service', "/*/common/*", default=[])
ret = self.comm.run(["systemctl", "show", "--property", "SubState", service])
self.assertTrue(ret.wait() == 0, "Commnad failed.")
self.assertEqual(ret.stdout.read(), "SubState=running\n", "Service (%s) not running" % service)
def crash_test(self):
self.log.debug("crash_test")
ret = self.comm.run(["ls", "/opt/usr/share/crash/dump"])
self.assertTrue(ret.wait() == 0, "Command failed")
out = ret.stdout.read()
# Save to whiteboard
self.whiteboard = out
# Get apps allowed/expected to generate data in /opt/share/crash
apps = self.params.get('ignored_apps', "/*/common/*", default=[])
# Get reference number of generated files
ref_cnt = self.params.get('ref_cnt', "/*/common/*", default=0)
crashed = re.findall(r'\w+_\d+_\d+', out)
cnt = len(crashed)
self.assertTrue(cnt <= ref_cnt, "More crashfiles were generated (%d)" % cnt)
for match in crashed:
self.assertTrue(match.split('_')[0] in apps)
if __name__ == "__main__":
main()
|
