Add avocado testscript: tct/run/common_test.py

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

diff --git a/tct/run/common_test.py b/tct/run/common_test.py
new file mode 100755 (executable)
index 0000000..a13a2a7
--- /dev/null
+++ b/tct/run/common_test.py
@@ -0,0 +1,88 @@
+#!/usr/bin/env python
+
+import subprocess
+import re
+
+from avocado import Test
+from avocado import main
+
+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)
+
+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', "/*")
+        self.ssh = SSH(ssh_config, ip)
+
+    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()
+        else:
+            self.error("Unknown test:  %s" % test)
+
+    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=[])
+        ret = self.ssh.run(["systemctl", "--failed"])
+        self.assertTrue(ret.wait() == 0, "Command failed")
+        out = ret.stdout.read()
+        # 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:
+            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")
+
+    def crash_test(self):
+        self.log.debug("crash_test")
+        ret = self.ssh.run(["tree", "-L", "2", "/opt/share/crash"])
+        self.assertTrue(ret.wait() == 0)
+        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)
+        for match in crashed:
+            self.assertTrue(match.split('_')[0] in apps)
+
+if __name__ == "__main__":
+    main()