Add SDB support to tct/run/common_test.py

Change-Id: Ic5a135a2763762cacbd0d756e1dea077d545204d
Signed-off-by: Aleksander Mistewicz <a.mistewicz@samsung.com>

1 files changed, 29 insertions, 8 deletions

diff --git a/tct/run/common_test.py b/tct/run/common_test.py
index a13a2a7..fae82a5 100755
--- a/tct/run/common_test.py
+++ b/tct/run/common_test.py
@@ -18,6 +18,16 @@ class SSH(object):
         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}
 
@@ -31,7 +41,14 @@ class CommonTest(Test):
         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)
@@ -48,14 +65,18 @@ class CommonTest(Test):
         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)
@@ -63,14 +84,14 @@ class CommonTest(Test):
     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
@@ -80,7 +101,7 @@ class CommonTest(Test):
         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)