janware/jw-devtest
1
mirror of ssh://git.janware.com/srv/git/janware/proj/jw-devtest synced 2026-01-15 02:22:56 +01:00
Code Releases Activity

test.expect(): Add support for cmd keyword argument

Browse source 
expect now takes a cmd argument, which defaults to None, and which
runs the command over the given connection before it takes to waiting
for a matching regex to be sent.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2022-12-09 16:07:27 +01:00
commit 6239c0546a
1 changed files with 23 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

29
src/python/devtest/os/test/test.py
View file

@ -8,11 +8,15 @@ from ..misc import *
from ..Connection import Connection
from ..Connections import Connections
async def expect(conn, regex=None, subject=None, act_timeout=0.1, total_timeout=None): # export
async def expect(conn, cmd=None, regex=None, subject=None, act_timeout=0.1, total_timeout=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 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
@ -66,16 +70,29 @@ async def cmds_exec(console, cmds, log_act=None, echo_cmd=False): # export
cmdline = c[0]
e = None if len(c) < 2 else c[1]
timeout = 200 if len(c) < 3 else c[2]
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):
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