) -> None:
self._get_client = get_client
self._verbose = verbose
+ self._print_on_fail: list[str] = []
self._idx = 0
def expand_parsed(marker: str, parse_into: Callable, **kwargs) -> None:
and not line.startswith(_MARK_COMMENT)]
expand_inserts()
expand_parsed(_MARK_LOGSRVRMSG, split_server_put_and_log)
- if self._verbose:
- title_idx = 'line number(s)'
- title_cmdname = 'command'
- self._colwidths: dict[str, int] = {}
- for idx_str, line in self._lines_t:
- self._colwidths['idx'] = max(
- len(title_idx),
- len(idx_str),
- self._colwidths.get('idx', 0))
- cmd_name, args = self._cmdname_and_args_from(line)
- self._colwidths['cmdname'] = max(
- len(title_cmdname),
- len(cmd_name),
- self._colwidths.get('cmdname', 0))
- self._colwidths['midargs'] = max(
- len(self._args_verbose(args[:-1])),
- self._colwidths.get('midargs', 0))
- self._print_verbose_columns(
- (title_idx, title_cmdname, 'arguments'))
+ title_idx = 'line number(s)'
+ title_cmdname = 'command'
+ self._colwidths: dict[str, int] = {}
+ for idx_str, line in self._lines_t:
+ self._colwidths['idx'] = max(
+ len(title_idx),
+ len(idx_str),
+ self._colwidths.get('idx', 0))
+ cmd_name, args = self._cmdname_and_args_from(line)
+ self._colwidths['cmdname'] = max(
+ len(title_cmdname),
+ len(cmd_name),
+ self._colwidths.get('cmdname', 0))
+ self._colwidths['midargs'] = max(
+ len(self._args_verbose(args[:-1])),
+ self._colwidths.get('midargs', 0))
+ self._log_verbose_columns((title_idx, title_cmdname, 'arguments'))
@staticmethod
def _args_verbose(args: tuple[str, ...]) -> str:
def _current_line(self) -> tuple[str, str]:
return self._lines_t[self._idx]
- def _print_verbose_columns(self, items: tuple[str, ...]) -> None:
+ def _log_verbose_columns(self, items: tuple[str, ...]) -> None:
to_print = []
for idx, colname in enumerate(('idx', 'cmdname', 'midargs'
)[:len(items) - 1]):
msg = items[idx]
to_print += [msg + ' ' * (self._colwidths[colname] - len(msg))]
to_print += [items[-1]]
- print(*to_print)
+ if self._verbose:
+ print(*to_print)
+ else:
+ self._print_on_fail += [' '.join(to_print)]
def ensure_has_started(self) -> None:
'Check if still at beginning, and if so, play till at next log line.'
) -> None:
'Call f with checks before and after against playbook, play it on.'
cmd_name, args = self._cmdname_and_args_from(self._current_line[1])
- if test_before:
- test_before(cmd_name, args)
- self._idx += 1
- next_idx_before = self._idx
- ret = f(*f_args, **f_kwargs)
- if test_after:
- test_after(cmd_name, args, ret)
- if self._idx == next_idx_before: # f may have called ._play_till_test
- self._play_till_test() # so we avoid jumping over next test
+ try:
+ if test_before:
+ test_before(cmd_name, args)
+ self._idx += 1
+ next_idx_before = self._idx
+ ret = f(*f_args, **f_kwargs)
+ if test_after:
+ test_after(cmd_name, args, ret)
+ if self._idx == next_idx_before: # f may call ._play_till_test, so
+ self._play_till_test() # only call if that didn't happen
+ except AssertionError as e:
+ for line in self._print_on_fail:
+ print(line)
+ raise e
def _play_till_test(self) -> None:
while True:
self._lines_t += [('', '> /quit'), ('', 'log 0 <..')]
idx_info, line = self._current_line
cmd_name, args = self._cmdname_and_args_from(line)
- if self._verbose:
- self._print_verbose_columns((idx_info, cmd_name,
- self._args_verbose(args[:-1]),
- self._args_verbose(args[-1:])))
+ self._log_verbose_columns((idx_info, cmd_name,
+ self._args_verbose(args[:-1]),
+ self._args_verbose(args[-1:])))
if cmd_name == _MARK_PROMPT:
assert self.put_keypress is not None
for c in args[0]: