fixup! os.test.ListCmd: Add response output convenience symlink

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2023-07-07 11:21:56 +02:00
commit b4bbe7bc9a

View file

@ -5,6 +5,7 @@ import os
import asyncio
import shlex
import traceback
import filecmp
from operator import itemgetter
from functools import total_ordering
from jwutils.log import *
@ -52,17 +53,25 @@ class ListCmd(TestCase): # export
# "needed": [ "dummyd", "v3.23" ]
# "key_": [ feature ]
def check_first_match(self, key_, features):
#level = OFF if self.name != 'root@[rcu_par_gp]' else True
#slog(level, "{}: check_first_match({}, {})".format(self.name, key_, features))
if self.__attribs is None:
#slog(level, "self.__attribs is None")
return False
for feature in features:
for key, val in self.__attribs.items():
#slog(level, "Checking {}={}".format(key, val))
for feature in features:
#slog(level, "Checking feature {}".format(feature))
if type(val) == bool:
#slog(level, "Type is bool")
if key_ == key:
#slog(level, "Returning value {}".format(val))
return val
continue
if type(val) != list:
raise Exception('Found attribute {}="{}" of unexpected value type {}'.format(key, val, type(val)))
if feature in val: # first match
#slog(level, "Returning {} ({} vs {})".format(key == key_, key, key_))
return key == key_
return False
@ -283,7 +292,16 @@ class ListCmd(TestCase): # export
raise Exception('Failed to compile regex ({}): >{}<'.format(e, str(re_str)))
return list(filter(None, [parse_line(l, fields) for l in lines]))
def _eval(self, output, features, header=None, tstamp=None):
def _eval(self, output, features, header=None, tag=None):
def linkify(path, ext, tag):
if tag is None:
link = None
raw = path + ext
else:
link = path + ext
raw = path + '-' + tag + ext
return raw, link
def format_rows(rows, quotes=False):
#def cmp(r1, r2):
@ -305,17 +323,23 @@ class ListCmd(TestCase): # export
else:
ref_lines = self.__read_rows(self.refpath, subject="reference")
# archive raw response
if self.__write_raw_response:
raw_response_path = self.refpath[0] + '.raw'
raw_response_path, raw_link_path = linkify(self.refpath[0], '.raw', tag)
with open(raw_response_path, "w") as f:
slog(INFO, 'Writing raw response to "{}"'.format(raw_response_path))
if header:
f.write(header)
f.write('\n'.join(output))
if raw_link_path is not None:
update_symlink(raw_response_path, raw_link_path)
output = self._filter(output)
# archive filtered response
last_features = set()
if self.__write_response:
response_path = self.refpath[0] + '.last'
response_path, link_path = linkify(self.refpath[0], '.filtered', tag)
if os.path.exists(response_path):
with open(response_path, "r") as f:
for line in f:
@ -328,6 +352,11 @@ class ListCmd(TestCase): # export
if header:
f.write(header)
f.write('\n'.join(output))
if filecmp.cmp(raw_response_path, response_path):
update_symlink(raw_response_path, response_path)
if link_path is not None:
update_symlink(response_path, link_path)
reference = set(self.parse(ref_lines))
actual = set(self.parse(output))
@ -372,13 +401,10 @@ class ListCmd(TestCase): # export
if not len(r):
return None
# archive complaints
if self.__write_response and (len(missing) or len(too_many)):
feature_diff = set(features) - last_features # if len(feature_diff) ?
response_path = self.refpath[0] + '.bad'
link_path = None
if tstamp is not None:
link_path = response_path
response_path += '-' + tstamp
response_path, link_path = linkify(self.refpath[0], '.bad', tag)
feature_diff_str = ', '.join(['"{}"'.format(f) for f in feature_diff])
with open(response_path, "w") as f:
slog(INFO, 'Writing feature diff to "{}"'.format(response_path))
@ -400,6 +426,9 @@ class ListCmd(TestCase): # export
async def _run(self, env, machine, phase):
console = get_console(env)
cmd = self.row_info('cmd')
tag = env.time_stamp
if env.tag is not None:
tag += '-' + env.tag
output = await cmd_exec(console, cmd, act_timeout=self.act_timeout,
total_timeout=self.total_timeout, echo_cmd=False)
if output is None:
@ -409,7 +438,7 @@ class ListCmd(TestCase): # export
else:
header = '# ' + cmd + '\n'
header += '# features: {}\n'.format(' '.join(env.features))
return self._eval(output, env.features, header=header, tstamp=env.time_stamp)
return self._eval(output, env.features, header=header, tag=tag)
def dump(self, prio, *args, **kwargs):
caller = kwargs['caller'] if 'caller' in kwargs.keys() else get_caller_pos(1)