return subprocess.Popen(["sshpass", "-p", "tizen", "ssh", "-F", self.ssh_config, "root@" + self.ip] + cmd,
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)
+
class CommonTest(Test):
default_params = {'timeout': 60.0}
self.log.debug("setUp")
ssh_config = self.params.get('ssh_config', "/*")
ip = self.params.get('ip', "/*")
- self.ssh = SSH(ssh_config, 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)
self.log.debug("systemctl_test")
ref_cnt = self.params.get('ref_cnt', "/*/common/*", default=-1)
ref_services = self.params.get('ignored_services', "/*/common/*", default=[])
- ret = self.ssh.run(["systemctl", "--failed"])
+ ret = self.comm.run(["systemctl", "--failed"])
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.ssh.run(["systemctl", "show", "--property", "SubState", service])
- self.assertTrue(ret.wait() == 0)
- self.assertEqual(ret.stdout.read(), "SubState=running\n", "Service not running")
+ 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.ssh.run(["tree", "-L", "2", "/opt/share/crash"])
- self.assertTrue(ret.wait() == 0)
+ ret = self.comm.run(["tree", "-L", "2", "/opt/share/crash"])
+ self.assertTrue(ret.wait() == 0, "Command failed")
out = ret.stdout.read()
# Save to whiteboard
self.whiteboard = out
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)
+ self.assertTrue(cnt <= ref_cnt, "More crashfiles were generated (%d)" % cnt)
for match in crashed:
self.assertTrue(match.split('_')[0] in apps)