Proposed implementation of a new runtime tests framework

new file mode 100644
index 0000000..82d0491
@@ -0,0 +1,95 @@
+TEST_LOG_DIR ?= "${WORKDIR}/testimage"
+
+DEFAULT_TEST_SUITES = "ping ssh auto"
+DEFAULT_TEST_SUITES_pn-core-image-minimal = "ping"
+DEFAULT_TEST_SUITES_pn-core-image-sato = "ping ssh connman rpm smart dmesg"
+DEFAULT_TEST_SUITES_pn-core-image-sato-sdk = "ping ssh connman rpm smart gcc dmesg"
+
+TEST_SUITES ?= "${DEFAULT_TEST_SUITES}"
+
+python do_testimage() {
+    testimage_main(d)
+}
+addtask testimage
+do_testimage[nostamp] = "1"
+do_testimage[depends] += "qemu-native:do_populate_sysroot"
+do_testimage[depends] += "qemu-helper-native:do_populate_sysroot"
+
+def testimage_main(d):
+    import unittest
+    import os
+    import oeqa.runtime
+    import re
+    from oeqa.oetest import createTestContext
+    from oeqa.utils.sshcontrol import SSHControl
+    from oeqa.utils.qemurunner import QemuRunner
+
+    testdir = d.getVar("TEST_LOG_DIR", True)
+    bb.utils.mkdirhier(testdir)
+    sshlog = os.path.join(testdir, "ssh_target_log.%s" % d.getVar('DATETIME', True))
+    sshloglink = os.path.join(testdir, "ssh_target_log")
+    if os.path.islink(sshloglink):
+        os.unlink(sshloglink)
+    os.symlink(sshlog, sshloglink)
+
+    # testsuite are modules - they are what the user wants (required tests)
+    testsuites = d.getVar("TEST_SUITES", True)
+    # testslist is what we'll run - order matters
+    # perhaps let the user decide if it wants the ssh test or not - it should be the first one though
+    #testslist = [ "ssh" ]
+    testslist = [ x for x in testsuites.split() if x != "auto" ]
+    # if we have auto search for other modules
+    if "auto" in testsuites:
+        for f in os.listdir(os.path.dirname(os.path.abspath(oeqa.runtime.__file__))):
+            if f.endswith('.py') and not f.startswith('_')  and f[:-3] not in testslist:
+                testslist.append(f[:-3])
+
+    testslist = [ "oeqa.runtime." + x for x in testslist ]
+
+    bb.note("Test modules  %s" % testslist)
+
+    class TestContext:
+        def __init__(self):
+            self.d = d
+            self.testslist = testslist
+            self.testsrequired = testsuites.split()
+            self.filesdir = os.path.join(os.path.dirname(os.path.abspath(oeqa.runtime.__file__)),"files")
+
+    # test context
+    tc = TestContext()
+
+    # prepare qemu instance
+    # and boot each supported fs type
+    machine=d.getVar("MACHINE", True)
+    #will handle fs type eventually, stick with ext3 for now
+    rootfs=d.getVar("DEPLOY_DIR_IMAGE", True) + '/' + d.getVar("IMAGE_BASENAME",True) + '-' + machine + '.ext3'
+
+    qemu = QemuRunner(machine, rootfs)
+    qemu.tmpdir = d.getVar("TMPDIR", True)
+    qemu.display = d.getVar("BB_ORIGENV", False).getVar("DISPLAY", True)
+    qemu.logfile = os.path.join(testdir, "qemu_boot_log.%s" % d.getVar('DATETIME', True))
+
+    #catch exceptions when loading or running tests (mostly our own errors)
+    try:
+        if qemu.launch():
+
+            # prepare test suite, loader and runner
+            suite = unittest.TestSuite()
+            runner = unittest.TextTestRunner(verbosity=2)
+            # set more context - ssh instance and qemu - we do this here because we needed the qemu ip
+            tc.qemu = qemu
+            tc.target = SSHControl(host=qemu.ip,logfile=sshlog)
+            # create suite
+            suite = createTestContext(tc)
+            bb.note("Found %s tests" % suite.countTestCases())
+            result = runner.run(suite)
+
+            if result.wasSuccessful():
+                bb.note("All required tests passed")
+            else:
+                raise bb.build.FuncFailed("Some tests failed. You should check the task log and the ssh log. (ssh log is %s" % sshlog)
+
+        else:
+            raise bb.build.FuncFailed("Failed to start qemu. You should check the task log and the qemu boot log (qemu log is %s)" % qemu.logfile)
+    finally:
+        qemu.kill()

new file mode 100644
index 0000000..e69de29

94 lines not on this page · raw message

new file mode 100644
index 0000000..e69de29

100 lines not on this page · raw message

new file mode 100644
index 0000000..cdebac1
@@ -0,0 +1,49 @@
+import socket
+import time
+import re
+from telnetlib import Telnet
+
+class oeTelnet(Telnet):
+
+    """
+    Override Telnet class to use unix domain sockets,
+    Telnet uses AF_INET for socket, we don't want that.
+    Also, provide a read_all variant with timeout, that
+    returns whatever output there is.
+    """
+
+    def __init__(self, stream=None, logfile=None):
+
+        Telnet.__init__(self, host=None)
+        self.stream = stream
+        self.logfile = logfile
+        if stream is not None:
+            self.open(stream)
+
+    def log(self, msg):
+        if self.logfile:
+            with open(self.logfile, "a") as f:
+                f.write("%s\n" % msg)
+
+    def open(self, stream):
+
+        self.eof = 0
+        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self.sock.connect(stream)
+
+    def read_all_timeout(self, match, timeout=120):
+        """Read until EOF or until timeout or until match.
+        """
+        ret = False
+        self.process_rawq()
+        endtime = time.time() + timeout
+        while not self.eof and time.time() < endtime:
+            self.fill_rawq()
+            self.process_rawq()
+            if re.search(match, self.cookedq):
+                ret = True
+                break
+        buf = self.cookedq
+        self.cookedq = ''
+        self.log(buf)
+        return (ret, buf)

new file mode 100644
index 0000000..e201e25
@@ -0,0 +1,160 @@
+import subprocess
+import optparse
+import sys
+import os
+import time
+import signal
+import re
+import bb
+from oeqa.utils.oetelnetlib import oeTelnet
+
+class QemuRunner:
+
+    def __init__(self, machine, rootfs, display = None, tmpdir = None, logfile = None):
+        # Popen object
+        self.runqemu = None
+
+        self.machine = machine
+        self.rootfs = rootfs
+
+        self.streampath = '/tmp/qemuconnection.%s' % os.getpid()
+        self.qemuparams = 'bootparams="console=ttyS0" qemuparams="-snapshot -serial unix:%s,server,nowait"' % self.streampath
+        self.qemupid = None
+        self.ip = None
+
+        self.display = display
+        self.tmpdir = tmpdir
+        self.logfile = logfile
+
+    def launch(self, qemuparams = None):
+
+        if qemuparams:
+            self.qemuparams = self.qemuparams[:-1] + " " + qemuparams + " " + '\"'
+
+        if self.display:
+            os.environ["DISPLAY"] = self.display
+        if not os.path.exists(self.rootfs):
+            bb.error("Invalid rootfs %s" % self.rootfs)
+            return False
+        if not os.path.exists(self.tmpdir):
+            bb.error("Invalid TMPDIR path %s" % self.tmpdir)
+            return False
+        else:
+            os.environ["OE_TMPDIR"] = self.tmpdir
+
+        launch_cmd = 'runqemu %s %s %s' % (self.machine, self.rootfs, self.qemuparams)
+        self.runqemu = subprocess.Popen(launch_cmd,shell=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,preexec_fn=os.setpgrp)
+
+        bb.note("runqemu started, pid is %s" % self.runqemu.pid)
+        # wait at most 30 seconds until qemu pid appears
+        bb.note("waiting at most 30 seconds for qemu pid")
+        endtime = time.time() + 30
+        while not self.is_alive() and time.time() < endtime:
+            time.sleep(0.5)
+
+        if self.is_alive():
+            bb.note("qemu started - qemu procces pid is %s" % self.qemupid)
+
+            console = oeTelnet(self.streampath, self.logfile)
+            bb.note("Waiting at most 120 seconds for login banner")
+            (match, text) = console.read_all_timeout("login:", 120)
+
+            if match:
+                bb.note("Reached login banner")
+                console.write("root\n")
+                (index, match, text) = console.expect([r"(root@[\w-]+:~#)"],10)
+                if not match:
+                    bb.note("Couldn't get prompt, all I got was:\n%s" % match.group(0))
+                    return False
+                console.write("ip addr show eth0 | sed -n '3p' | awk '{ print $2 }' | cut -f 1 -d \"/\"\n")
+                (index, match, text) = console.expect([r"((?:[0-9]{1,3}\.){3}[0-9]{1,3})"],10)
+                console.close()
+                if match:
+                    self.ip = match.group(0)
+                    bb.note("Ip found: %s" % self.ip)
+                else:
+                    bb.note("Couldn't determine ip, all I got was:\n%s" % text)
+                    return False
+            else:
+                console.close()
+                bb.note("Target didn't reached login boot in 120 seconds")
+                lines = "\n".join(text.splitlines()[-5:])
+                bb.note("Last 5 lines of text:\n%s" % lines)
+                bb.note("Check full boot log: %s" % self.logfile)
+                return False
+        else:
+            bb.note("Qemu pid didn't appeared in 30 seconds")
+            self.runqemu.terminate()
+            self.runqemu.kill()
+            bb.note("Output from runqemu: %s " % self.runqemu.stdout.read())
+            self.runqemu.stdout.close()
+            return False
+
+        return self.is_alive()
+
+
+    def kill(self):
+        if self.runqemu:
+            os.kill(-self.runqemu.pid,signal.SIGTERM)
+        self.qemupid = None
+        self.ip = None
+        if os.path.exists(self.streampath):
+            os.remove(self.streampath)
+
+    def restart(self, qemuparams = None):
+        if self.is_alive():
+            self.kill()
+        bb.note("Qemu Restart required...")
+        return self.launch(qemuparams)
+
+    def is_alive(self):
+        qemu_child = self.find_child(str(self.runqemu.pid))
+        if qemu_child:
+            self.qemupid = qemu_child[0]
+            return os.path.exists("/proc/" + str(self.qemupid))
+        return False
+
+    def find_child(self,parent_pid):
+        #
+        # Walk the process tree from the process specified looking for a qemu-system. Return its [pid'cmd]
+        #
+        ps = subprocess.Popen(['ps', 'axww', '-o', 'pid,ppid,command'], stdout=subprocess.PIPE).communicate()[0]
+        processes = ps.split('\n')
+        nfields = len(processes[0].split()) - 1
+        pids = {}
+        commands = {}
+        for row in processes[1:]:
+            data = row.split(None, nfields)
+            if len(data) != 3:
+                continue
+            if data[1] not in pids:
+                pids[data[1]] = []
+
+            pids[data[1]].append(data[0])
+            commands[data[0]] = data[2]
+
+        if parent_pid not in pids:
+            sys.stderr.write("No children found matching %s\n" % parent_pid)
+            return []
+
+        parents = []
+        newparents = pids[parent_pid]
+        while newparents:
+            next = []
+            for p in newparents:
+                if p in pids:
+                    for n in pids[p]:
+                        if n not in parents and n not in next:
+                            next.append(n)
+                if p not in parents:
+                    parents.append(p)
+                    newparents = next
+        #print "Children matching %s:" % str(parents)
+        for p in parents:
+            # Need to be careful here since runqemu-internal runs "ldd qemu-system-xxxx"
+            # Also, old versions of ldd (2.11) run "LD_XXXX qemu-system-xxxx"
+            basecmd = commands[p].split()[0]
+            basecmd = os.path.basename(basecmd)
+            if "qemu-system" in basecmd and "192.168" in commands[p]:
+                return [int(p),commands[p]]
+

index f2eb2e1..406092b 100755
@@ -156,7 +156,7 @@ while true; do
             serial_option=`expr "$SCRIPT_QEMU_EXTRA_OPT" : '.*\(-serial\)'`
             kvm_option=`expr "$SCRIPT_QEMU_EXTRA_OPT" : '.*\(-enable-kvm\)'`
             [ ! -z "$serial_option" -o ! -z "$kvm_option" ] && \
-                error "Please use simplified serial or kvm options instead"
+                echo "Please use simplified serial or kvm options instead"
             ;;
         "bootparams="*)
             SCRIPT_KERNEL_OPT="$SCRIPT_KERNEL_OPT ${arg##bootparams=}"

new file mode 100644
index 0000000..4b7a53d
@@ -0,0 +1,40 @@
+from oeqa.oetest import *
+
+class skipIfFailure(object):
+
+    def __init__(self,testcase):
+        self.testcase = testcase
+
+    def __call__(self,f):
+        def wrapped_f(*args):
+            if self.testcase in oeRuntimeTest.testFailures:
+                raise unittest.SkipTest("Testcase dependency not met: %s" % self.testcase)
+            f(*args)
+        wrapped_f.__name__ = f.__name__
+        return wrapped_f
+
+class skipIfSkipped(object):
+
+    def __init__(self,testcase):
+        self.testcase = testcase
+
+    def __call__(self,f):
+        def wrapped_f(*args):
+            if self.testcase in oeRuntimeTest.testSkipped:
+                raise unittest.SkipTest("Testcase dependency not met: %s" % self.testcase)
+            f(*args)
+        wrapped_f.__name__ = f.__name__
+        return wrapped_f
+
+class skipUnlessPassed(object):
+
+    def __init__(self,testcase):
+        self.testcase = testcase
+
+    def __call__(self,f):
+        def wrapped_f(*args):
+            if self.testcase in (oeRuntimeTest.testSkipped, oeRuntimeTest.testFailures, oeRuntimeTest.testErrors):
+                raise unittest.SkipTest("Testcase dependency not met: %s" % self.testcase)
+            f(*args)
+        wrapped_f.__name__ = f.__name__
+        return wrapped_f

new file mode 100644
index 0000000..e69de29

new file mode 100644
index 0000000..835e135
@@ -0,0 +1,34 @@
+import unittest
+from oeqa.oetest import oeRuntimeTest, skipModule
+from oeqa.utils.decorators import *
+
+def setUpModule():
+    if not oeRuntimeTest.hasPackage("connman"):
+        skipModule("No connman package in image")
+
+
+class ConnmanTest(oeRuntimeTest):
+
+    @skipUnlessPassed('test_ssh')
+    def test_connmand_help(self):
+        (status, output) = self.target.run('/usr/sbin/connmand --help')
+        self.assertEqual(status, 0, msg="status and output: %s and %s" % (status,output))
+
+
+    @skipUnlessPassed('test_connmand_help')
+    def test_connmand_running(self):
+        status = self.target.run('ls -l `which ps` | grep busybox')[0]
+        if status == 0:
+            oeRuntimeTest.pscmd = 'ps'
+        else:
+            oeRuntimeTest.pscmd = 'ps -ef'
+        (status, output) = self.target.run(oeRuntimeTest.pscmd + ' | grep [c]onnmand')
+        self.assertEqual(status, 0, msg="no connmand process, ps output: %s" % self.target.run(oeRuntimeTest.pscmd)[1])
+
+    @skipUnlessPassed('test_connmand_running')
+    def test_connmand_unique(self):
+        self.target.run('/usr/sbin/connmand')
+        output = self.target.run(oeRuntimeTest.pscmd + ' | grep -c [c]onnmand')[1]
+        self.assertEqual(output, "1", msg="more than one connmand running in background, ps output: %s\n%s" % (output, self.target.run(oeRuntimeTest.pscmd)[1]))
+
+

new file mode 100644
index 0000000..3acf09e
@@ -0,0 +1,10 @@
+import unittest
+from oeqa.oetest import oeRuntimeTest
+from oeqa.utils.decorators import *
+
+class DmesgTest(oeRuntimeTest):
+
+    @skipUnlessPassed('test_ssh')
+    def test_dmesg(self):
+        (status, output) = self.target.run('dmesg | grep -v mmci-pl18x | grep -i error')
+        self.assertEqual(status, 1, msg = "Error messages in dmesg log: %s" % output)

new file mode 100644
index 0000000..397d075
@@ -0,0 +1,14 @@
+import unittest
+from oeqa.oetest import oeRuntimeTest, skipModule
+
+def setUpModule():
+    multilibs = oeRuntimeTest.tc.d.getVar("MULTILIBS", True) or ""
+    if "multlib:lib32" not in multilibs:
+        skipModule("this isn't a multilib:lib32 image")
+
+
+class MultilibFileTest(oeRuntimeTest):
+
+    def test_file_connman(self):
+        (status, output) = self.target.run('file -L /usr/sbin/connmand | grep "ELF 32-bit LSB executable"')
+        self.assertEqual(status, 0, msg="status and output : %s and %s" % (status,output))

new file mode 100644
index 0000000..d6a0c28
@@ -0,0 +1,11 @@
+import subprocess
+import unittest
+import sys
+from oeqa.oetest import oeRuntimeTest
+
+class PingTest(oeRuntimeTest):
+
+    def test_ping(self):
+        status = subprocess.call("ping -w 30 -c 1 %s" % oeRuntimeTest.tc.qemu.ip, shell=True, stdout=subprocess.PIPE)
+        self.assertEqual(status, 0)
+

new file mode 100644
index 0000000..57101b0
@@ -0,0 +1,25 @@
+import unittest
+from oeqa.oetest import oeRuntimeTest, skipModule
+from oeqa.utils.decorators import *
+
+def setUpModule():
+    if not oeRuntimeTest.hasFeature("package-management"):
+            skipModule("rpm module skipped: target doesn't have package-management in IMAGE_FEATURES")
+    if "package_rpm" != oeRuntimeTest.tc.d.getVar("PACKAGE_CLASSES", True).split()[0]:
+            skipModule("rpm module skipped: target doesn't have rpm as primary package manager")
+
+
+class RpmHelpTest(oeRuntimeTest):
+
+    @skipUnlessPassed('test_ssh')
+    def test_rpm_help(self):
+        (status, output) = self.target.run('rpm --help')
+        self.assertEqual(status, 0, msg="status and output: %s and %s" % (status,output))
+
+class RpmQueryTest(oeRuntimeTest):
+
+    @skipUnlessPassed('test_rpm_help')
+    def test_rpm_query(self):
+        (status, output) = self.target.run('rpm -q rpm')
+        self.assertEqual(status, 0, msg="status and output: %s and %s" % (status,output))
+

new file mode 100644
index 0000000..b7d4e3e
@@ -0,0 +1,23 @@
+import unittest
+from oeqa.oetest import oeRuntimeTest
+from oeqa.utils.decorators import *
+
+def setUpModule():
+    if not oeRuntimeTest.hasFeature("package-management"):
+        skipModule("Image doesn't have package management feature")
+    if not oeRuntimeTest.hasPackage("smart"):
+        skipModule("Image doesn't have smart installed")
+
+class SmartHelpTest(oeRuntimeTest):
+
+    def test_smart_help(self):
+        status = self.target.run('smart --help')[0]
+        self.assertEqual(status, 0)
+
+class SmartQueryTest(oeRuntimeTest):
+
+    @skipUnlessPassed('test_smart_help')
+    def test_smart_query(self):
+        (status, output) = self.target.run('smart query rpm')
+        self.assertEqual(status, 0, msg="status and output : %s and %s" % (status,output))
+

new file mode 100644
index 0000000..8c96020
@@ -0,0 +1,16 @@
+import subprocess
+import unittest
+import sys
+from oeqa.oetest import oeRuntimeTest, skipModule
+from oeqa.utils.decorators import *
+
+def setUpModule():
+    if not (oeRuntimeTest.hasPackage("dropbear") or oeRuntimeTest.hasPackage("openssh")):
+        skipModule("No ssh package in image")
+
+class SshTest(oeRuntimeTest):
+
+    @skipUnlessPassed('test_ping')
+    def test_ssh(self):
+        (status, output) = self.target.run('uname -a')
+        self.assertEqual(status, 0, msg="SSH Test failed: %s" % output)

new file mode 100644
index 0000000..ca1844e
@@ -0,0 +1,5 @@
+test: test.o
+	gcc -o test test.o -lm
+test.o: test.c
+	gcc -c test.c
+

new file mode 100644
index 0000000..b63badd
@@ -0,0 +1,36 @@
+import unittest
+import os
+from oeqa.oetest import oeRuntimeTest, skipModule
+from oeqa.utils.decorators import *
+
+def setUpModule():
+    if not oeRuntimeTest.hasFeature("tools-sdk"):
+        skipModule("Image doesn't have tools-sdk in IMAGE_FEATURES")
+
+
+class GccCompileTest(oeRuntimeTest):
+
+    @classmethod
+    def setUpClass(self):
+        oeRuntimeTest.tc.target.copy_to(os.path.join(oeRuntimeTest.tc.filesdir, "test.c"), "/tmp/test.c")
+        oeRuntimeTest.tc.target.copy_to(os.path.join(oeRuntimeTest.tc.filesdir, "testmakefile"), "/tmp/testmakefile")
+
+    def test_gcc_compile(self):
+        (status, output) = self.target.run('gcc /tmp/test.c -o /tmp/test -lm')
+        self.assertEqual(status, 0, msg="gcc compile failed, output: %s" % output)
+        (status, output) = self.target.run('/tmp/test')
+        self.assertEqual(status, 0, msg="running compiled file failed, output %s" % output)
+
+    def test_gpp_compile(self):
+        (status, output) = self.target.run('g++ /tmp/test.c -o /tmp/test -lm')
+        self.assertEqual(status, 0, msg="g++ compile failed, output: %s" % output)
+        (status, output) = self.target.run('/tmp/test')
+        self.assertEqual(status, 0, msg="running compiled file failed, output %s" % output)
+
+    def test_make(self):
+        (status, output) = self.target.run('cd /tmp; make -f testmakefile')
+        self.assertEqual(status, 0, msg="running make failed, output %s" % output)
+
+    @classmethod
+    def tearDownClass(self):
+        oeRuntimeTest.tc.target.run("rm /tmp/test.c /tmp/test.o /tmp/test /tmp/testmakefile")