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

test.ListCmd: Add support for refpath lists

Browse source 
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2022-12-19 15:19:01 +01:00
commit e0b3cc41ce
1 changed files with 21 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

33
src/python/devtest/os/test/ListCmd.py
View file

@ -172,7 +172,7 @@ class ListCmd(TestCase): # export
# ------------------------------------- class ListCmd methods # ------------------------------------- class ListCmd methods
def __init__(self, refpath, act_timeout=2, total_timeout=None, write_response=None): def __init__(self, refpath, act_timeout=2, total_timeout=None, write_response=None):
self.refpath = refpath self.refpath = [refpath] if isinstance(refpath, str) else refpath
self.act_timeout = act_timeout self.act_timeout = act_timeout
self.total_timeout = total_timeout self.total_timeout = total_timeout
self.__decisive = None self.__decisive = None
@ -192,6 +192,18 @@ class ListCmd(TestCase): # export
else: else:
raise Exception('Invalid value "{}" of environment variable "{}"'.format(val, key)) raise Exception('Invalid value "{}" of environment variable "{}"'.format(val, key))
def __read_rows(self, path, subject=None):
if isinstance(path, list):
ret = []
for entry in path:
ret.extend(self.__read_rows(entry, subject))
return ret
if subject is None:
subject = 'rows'
slog(INFO, 'Reading {} from "{}"'.format(subject, path))
with open(path, "r") as f:
return f.readlines()
# override this # override this
def _row_info(self): def _row_info(self):
return { return {
@ -273,15 +285,13 @@ class ListCmd(TestCase): # export
#return [row.to_str(fields=['cmp-fields', 'fields'], only_values=True, quotes=quotes) for row in sorted(rows, key=itemgetter(*sort_keys))] #return [row.to_str(fields=['cmp-fields', 'fields'], only_values=True, quotes=quotes) for row in sorted(rows, key=itemgetter(*sort_keys))]
return [row.to_str(fields=['cmp-fields', 'fields'], only_values=True, quotes=quotes) for row in sorted(rows)] return [row.to_str(fields=['cmp-fields', 'fields'], only_values=True, quotes=quotes) for row in sorted(rows)]
if self.__write_response and not os.path.exists(self.refpath): if self.__write_response and not os.path.exists(self.refpath[0]):
ref_lines = [] ref_lines = []
else: else:
slog(INFO, 'Reading reference from "{}"'.format(self.refpath)) ref_lines = self.__read_rows(self.refpath, subject="reference")
with open(self.refpath, "r") as f:
ref_lines = f.readlines()
if self.__write_raw_response: if self.__write_raw_response:
raw_response_path = self.refpath + '.raw' raw_response_path = self.refpath[0] + '.raw'
with open(raw_response_path, "w") as f: with open(raw_response_path, "w") as f:
slog(INFO, 'Writing raw response to "{}"'.format(raw_response_path)) slog(INFO, 'Writing raw response to "{}"'.format(raw_response_path))
if header: if header:
@ -290,7 +300,7 @@ class ListCmd(TestCase): # export
output = self._filter(output) output = self._filter(output)
last_features = set() last_features = set()
if self.__write_response: if self.__write_response:
response_path = self.refpath + '.last' response_path = self.refpath[0] + '.last'
if os.path.exists(response_path): if os.path.exists(response_path):
with open(response_path, "r") as f: with open(response_path, "r") as f:
for line in f: for line in f:
@ -342,7 +352,7 @@ class ListCmd(TestCase): # export
feature_diff = set(features) - last_features feature_diff = set(features) - last_features
if self.__write_response and len(feature_diff): if self.__write_response and len(feature_diff):
response_path = self.refpath + '.bad' response_path = self.refpath[0] + '.bad'
feature_diff_str = ', '.join(['"{}"'.format(f) for f in feature_diff]) feature_diff_str = ', '.join(['"{}"'.format(f) for f in feature_diff])
with open(response_path, "w") as f: with open(response_path, "w") as f:
slog(INFO, 'Writing feature diff to "{}"'.format(response_path)) slog(INFO, 'Writing feature diff to "{}"'.format(response_path))
@ -351,11 +361,11 @@ class ListCmd(TestCase): # export
if len(missing): if len(missing):
f.write("# --- missing {}\n".format(feature_diff)) f.write("# --- missing {}\n".format(feature_diff))
for row in missing: for row in missing:
f.write(row.line + ' # "needed": [{}], "bad": ["default"]\n'.format(feature_diff_str)) f.write(row.line.strip() + ' # "needed": [{}], "bad": ["default"]\n'.format(feature_diff_str))
if len(too_many): if len(too_many):
f.write("# --- too many {}\n".format(feature_diff)) f.write("# --- too many {}\n".format(feature_diff))
for row in too_many: for row in too_many:
f.write(row.line + ' # "bad" [{}]\n'.format(feature_diff_str)) f.write(row.line.strip() + ' # "bad" [{}]\n'.format(feature_diff_str))
return ' and '.join(r) return ' and '.join(r)
async def _run(self, env, machine, phase): async def _run(self, env, machine, phase):
@ -374,8 +384,7 @@ class ListCmd(TestCase): # export
def dump(self, prio, *args, **kwargs): def dump(self, prio, *args, **kwargs):
caller = kwargs['caller'] if 'caller' in kwargs.keys() else get_caller_pos(1) caller = kwargs['caller'] if 'caller' in kwargs.keys() else get_caller_pos(1)
with open(self.refpath, "r") as f: reference = self.__read_rows(self.refpath, subject="reference")
reference = self.parse(f.readlines())
for l in reference: for l in reference:
slog(NOTICE, "{}".format(l), caller=caller) slog(NOTICE, "{}".format(l), caller=caller)