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

Add test.cmds_exec()

Browse source 
test.cmds_exec() takes a list of lists with commands to be sent to a
console, returns None on success and an error message on failure.

The list can consist of up to three entries, 2 + 3 are optional:

  [ cmd, expect-regex, timeout]

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

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

@ -39,20 +39,45 @@ async def expect(conn, regex=None, subject=None, act_timeout=0.1, total_timeout=
return buf
raise Exception("never reached")
async def cmd_expect(conn, cmd, subject=None, regex=None, act_timeout=0.1): # export
async def cmd_expect(conn, cmd, subject=None, regex=None, log_act=None, act_timeout=0.1): # export
slog(NOTICE, 'sending command "{}"'.format(str(cmd).rstrip("\n")))
res = await cmd_exec(conn, cmd)
if not isinstance(res, list):
msg = 'failed to send command over connection {}'.format(conn)
slog(ERR, 'FAIL: ' + msg)
return msg
if expect(conn, regex=regex, subject=subject, act_timeout=act_timeout):
if 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]
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)
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)