os.test.main.py: Rename to os.test.test.py

Signed-off-by: Jan Lindemann <jan@janware.com>

src/python/devtest/os/test/test.py

@@ -0,0 +1,191 @@
+# -*- coding: utf-8 -*-
+
+import asyncio
+import time
+import re
+from jwutils.log import *
+from ..misc import *
+from ..Connection import Connection
+from ..Connections import Connections
+
+async def expect(conn, cmd=None, regex=None, subject=None, act_timeout=0.1, total_timeout=None, log_act=None): # export
+    # regex is a keyword-argument so that other arguments to support other
+    # checks can be added later
+    if regex is None:
+        raise Exception("passed empty pattern to test.expect()")
+    if log_act is None:
+        log_act = INFO
+    if cmd is not None:
+        if isinstance(cmd, str) and not cmd.endswith('\n'):
+            cmd += '\n'
+        await conn.write(cmd)
+    deadline = None
+    if total_timeout is not None:
+        deadline = time.time() + total_timeout
+    if subject == None:
+        subject = ">" + regex + "<"
+    while True:
+        flags = Connection.Flags.ReadStripNewline|Connection.Flags.ReadDecodeToString|Connection.Flags.ReadOnlyPrintable
+        timeout = act_timeout
+        now = time.time()
+        if deadline is not None:
+            timeout = min(deadline - now, timeout)
+        buf = await conn.read(timeout, flags=flags)
+        if buf is None:
+            if deadline is not None and now > deadline:
+                slog(INFO, "reading from console timed out after %d seconds total" % total_timeout)
+            else:
+                slog(INFO, "reading from console timed out after %d seconds of inactivity" % act_timeout)
+            return None
+        slog(log_act, "read from console: >%s<" % buf, only_printable=True)
+        if len(buf) == 0:
+            raise Exception('connection "{}" reset while waiting on {}'.format(conn, subject))
+        if re.search(regex, buf):
+            slog(NOTICE, "found %s" % subject)
+            return buf
+    raise Exception("never reached")
+
+async def cmd_expect(conn, cmd, subject=None, regex=None, log_act=None, act_timeout=0.1, echo_cmd=None): # export
+    slog(NOTICE, 'sending command "{}"'.format(str(cmd).rstrip("\n")))
+    res = await cmd_exec(conn, cmd, echo_cmd=echo_cmd)
+    if not isinstance(res, list):
+        msg = 'failed to send command over connection {}'.format(conn)
+        slog(ERR, 'FAIL: ' + msg)
+        return msg
+    if await expect(conn, regex=regex, subject=subject, act_timeout=act_timeout, log_act=log_act):
+        slog(NOTICE, 'PASS: found pattern "{}" in command output'.format(regex))
+        return None
+    r = "FAIL: failed to find {} in command output".format(regex)
+    slog(ERR, r)
+    return r
+
+# takes a list of [ cmd, regex-expect (or None), expect_timeout (or None)]
+async def cmds_exec(console, cmds, log_act=None, echo_cmd=False): # export
+
+    def find_match(response, pattern):
+        for s in response:
+            if re.search(pattern, s) is not None:
+                return True
+        return False
+
+    for c in cmds:
+        cmdline = c[0]
+        e = None if len(c) < 2 else c[1]
+        timeout = 200 if len(c) < 3 else c[2]
+        if cmdline is None:
+            if e is None:
+                continue
+            slog(DEBUG, 'Waiting for: "{}"'.format(e))
+            rr = await expect(console, regex=e, act_timeout=timeout)
+            if rr is None:
+                return 'Timed out waiting for: "{}"'.format(e)
+        elif isinstance(cmdline, str):
+            rr = await cmd_exec(console, cmdline, act_timeout=1, log_act=log_act, echo_cmd=echo_cmd)
+            if rr is None:
+                return 'Failed to run command "{}"'.format(cmdline)
+            if e is not None:
+                if find_match(rr, e):
+                    continue
+                slog(DEBUG, 'Waiting for: "{}"'.format(e))
+                rr = await expect(console, regex=e, act_timeout=timeout)
+                if rr is None:
+                    return 'Timed out waiting for: "{}"'.format(e)
+        else:
+            rr = cmdline(console, *c[1:])
+            if ret is not None:
+                return ret
+
+    return None
+
+async def grep_log(conn, regex, log_glob='/var/log/messages', act_timeout=0.1, total_timeout=None): # export
+    cmd = "ls {} | tail -1".format(log_glob)
+    res = await cmd_exec(conn, cmd, act_timeout=act_timeout, total_timeout=total_timeout)
+    if res is None or len(res) < 2:
+        raise Exception("failed to get log file {}".format(log_glob))
+    log = res[1]
+    slog(DEBUG, 'found log file path "{}"'.format(log))
+    cmd = 'grep "{}" {}'.format(regex, log)
+    res = await cmd_exec(conn, cmd, act_timeout=act_timeout, total_timeout=total_timeout)
+    if res is None or len(res) < 2:
+        raise Exception("{} failed".format(cmd))
+    r = []
+    # remove everything up to and including "grep .* log" from array, assume last line is a prompt
+    for line in reversed(res[:-1]):
+        if line.find(log) != -1:
+            return r
+        r.insert(0, line)
+    for line in res:
+        slog(ERR, "read >{}<".format(line))
+    raise Exception("{} returned garbage".format(cmd))
+
+def diff_iterables(reference, actual, ignore = None, unneeded = None, fail_unknown = False, fail_missing = True): # export
+
+    reference = set(reference)
+    actual    = set(actual)
+    needed    = reference
+    if ignore is not None:
+        ignore = set(ignore)
+        needed   -= ignore
+        actual   -= ignore
+    if unneeded is not None:
+        unneeded = set(unneeded)
+        needed   -= unneeded
+
+    both      = needed | actual
+    only_n    = needed - actual
+    only_a    = actual - needed
+
+    needed = sorted(list(needed))
+    actual = sorted(list(actual))
+    only_n = sorted(list(only_n))
+    only_a = sorted(list(only_a))
+
+    slog_m(NOTICE, "--- needed:\n", sorted(list(needed)))
+    if ignore is not None:
+        slog_m(NOTICE, "--- ignored:\n", sorted(list(ignore)))
+    slog_m(NOTICE, "--- actual:\n", sorted(list(actual)))
+
+    r = []
+    if fail_missing and len(only_n):
+        slog_m(ERR,    "--- should be there but are not:\n", only_n)
+        r.append("missing (" + ', '.join(only_n) + ')')
+    if fail_unknown and len(only_a):
+        slog_m(ERR,    "--- too many:\n", only_a)
+        r.append("too many (" + ', '.join(only_a) + ')')
+    if not len(r):
+        return None
+    return ' and '.join(r)
+
+def get_console(env, name=None, require=True): # export
+    # need to copy this because otherwise the iterator is not reset
+    connections = Connections(env.connections)
+    for c in connections:
+        if name is not None and hasattr(c.info, 'name') and c.info.name == name:
+            return c.instance
+        if c.info.proto == Connection.Proto.Console:
+            return c.instance
+        # FIXME: This whole proto == Console thing is BS: "Console" is not a protocol
+        if c.info.proto == Connection.Proto.Serial:
+            return c.instance
+    if require:
+        raise Exception("no console connection among connections: {}".format(', '.join(
+            [ '{}'.format(c.instance) for c in env.connections ])))
+    return None
+
+def get_connection(env, name, require=True): # export
+    # need to copy this because otherwise the iterator is not reset
+    connections = Connections(env.connections)
+    for c in connections:
+        if hasattr(c.info, 'name') and c.info.name == name:
+            return c.instance
+    if require:
+        raise Exception('no connection named "{}" among connections: {}'.format(
+            name, ', '.join([ '{}'.format(c.instance) for c in env.connections ])))
+    return None
+
+# untested
+def get_connections(env, proto=None, require=True): # export
+    r = env.connections if proto is None else [c.instance for c in env.connections if c.info.proto == proto]
+    if require and len(r) == 0:
+        raise Exception("No connection of %s type" % proto.name if proto else "any")
+    return r