diff --git a/moler/cmd/unix/bzip2.py b/moler/cmd/unix/bzip2.py new file mode 100644 index 000000000..1dcc980a6 --- /dev/null +++ b/moler/cmd/unix/bzip2.py @@ -0,0 +1,108 @@ +# -*- coding: utf-8 -*- +""" +Bzip2 command module. + +""" + +__author__ = 'Marcin Usielski' +__copyright__ = 'Copyright (C) 2024, Nokia' +__email__ = 'marcin.usielski@nokia.com' + + +import re + +from moler.cmd.unix.genericunix import GenericUnixCommand +from moler.exceptions import CommandFailure, ParsingDone + + +class Bzip2(GenericUnixCommand): + + def __init__(self, connection, files, options=None, prompt=None, newline_chars=None, + runner=None): + """ + Command to work with bzip2 command. + + :param connection: moler connection to device, terminal when command is executed. + :param options: options of bzip2 command. + :param files: files to work on. String with file(s) or a list of files. + :param prompt: prompt on which command has to stop working. + :param newline_chars: characters to split lines. + :param runner: Runner to run command. + """ + super(Bzip2, self).__init__(connection=connection, prompt=prompt, newline_chars=newline_chars, runner=runner) + # Parameters defined by calling the command + self.options = options + self.files = files + self.ret_required = False + + def build_command_string(self): + cmd = "bzip2" + if self.options is not None: + cmd = f"{cmd} {self.options}" + if self.files is not None: + if isinstance(self.files, str): + cmd = f"{cmd} {self.files}" + else: + cmd = f"{cmd} {' '.join(self.files)}" + return cmd + + def on_new_line(self, line, is_full_line): + print(f"on_new_line: {is_full_line}-> '{line}'") + try: + if is_full_line: + self._parse_error_via_output_line(line) + self._parse_file(line) + except ParsingDone: + pass # line has been fully parsed by one of above parse-methods + return super(Bzip2, self).on_new_line(line, is_full_line) + + _re_error_line = re.compile(r"(?P.*(Can't open input file|No such file or directory|already has\s+\.bz2 suffix).*)") + + def _parse_error_via_output_line(self, line): + if self._cmd_output_started and self._regex_helper.search_compiled(Bzip2._re_error_line, line): + self.set_exception(CommandFailure(self, f"ERROR: {self._regex_helper.group('error')}")) + raise ParsingDone() + + # a.bz2: done + _re_parse_file = re.compile(r"(?P\S+|\S.*\S)\s*:\s*(done|no data compressed|\d+)") + + def _parse_file(self, line: str) -> None: + if self._regex_helper.search_compiled(Bzip2._re_parse_file, line): + if 'files' not in self.current_ret: + self.current_ret['files'] = [] + self.current_ret['files'].append(self._regex_helper.group('name')) + raise ParsingDone() + + +COMMAND_OUTPUT_decompress = """bzip2 -dkfv *.bz2 + a.bz2: done + b.bz2: done +moler_bash# """ + + +COMMAND_RESULT_decompress = { + 'files': ['a.bz2', 'b.bz2'] +} + + +COMMAND_KWARGS_decompress = { + "options": "-dkfv", + "files": "*.bz2", +} + + +COMMAND_OUTPUT_compress = """bzip2 -zkfv a b + a: 0.914:1, 8.750 bits/byte, -9.38% saved, 96 in, 105 out. + b: 0.941:1, 8.500 bits/byte, -6.25% saved, 176 in, 187 out. +moler_bash# """ + + +COMMAND_RESULT_compress = { + 'files': ['a', 'b'] +} + + +COMMAND_KWARGS_compress = { + "options": "-zkfv", + "files": ["a", "b"], +} diff --git a/moler/cmd/unix/gunzip.py b/moler/cmd/unix/gunzip.py index f36ff69ff..9874aac12 100644 --- a/moler/cmd/unix/gunzip.py +++ b/moler/cmd/unix/gunzip.py @@ -3,9 +3,9 @@ Gunzip command module. """ -__author__ = "Adrianna Pienkowska" -__copyright__ = "Copyright (C) 2018, Nokia" -__email__ = "adrianna.pienkowska@nokia.com" +__author__ = "Adrianna Pienkowska, Marcin Usielski" +__copyright__ = "Copyright (C) 2018-2024, Nokia" +__email__ = "adrianna.pienkowska@nokia.com, marcin.usielski@nokia.com" import re @@ -60,6 +60,7 @@ def on_new_line(self, line, is_full_line): self._asks_to_overwrite(line) self._create_dictionary_at_l_option(line) self._command_failure(line) + self._parse_verbose(line) except ParsingDone: pass return super(Gunzip, self).on_new_line(line, is_full_line) @@ -118,6 +119,24 @@ def _command_failure(self, line): ) raise ParsingDone + # a.gz: 0.0% -- created a + _re_verbose = re.compile(r"(?P\S+):.*\s+(?P[\d\.]+%)\s+\S+\s+\S+\s+(?P\S+)") + + def _parse_verbose(self, line): + if self._regex_helper.search_compiled(Gunzip._re_verbose, line): + if 'files' not in self.current_ret: + self.current_ret['files'] = [] + self.current_ret['details'] = [] + file_desc = { + 'compressed_file': self._regex_helper.group("COMPRESSED_FILE"), + 'ratio': self._regex_helper.group("RATIO"), + 'decompressed_file': self._regex_helper.group("DECOMPRESSED_FILE") + } + self.current_ret['files'].append(self._regex_helper.group("DECOMPRESSED_FILE")) + self.current_ret['details'].append(file_desc) + + raise ParsingDone() + COMMAND_OUTPUT_without_options = """ xyz@debian:~$ gunzip new.gz @@ -201,3 +220,28 @@ def _command_failure(self, line): } ] } + + +COMMAND_OUTPUT_verbose = """gunzip -fkv a.gz b.gz +a.gz: 0.0% -- created a +b.gz: 0.0% -- created b +moler_bash# """ + +COMMAND_KWARGS_verbose = {"archive_name": ["a.gz", "b.gz"], "options": "-fkv"} + +COMMAND_RESULT_verbose = { + "files": ["a", "b"], + "details": [ + { + 'compressed_file': 'a.gz', + 'ratio': '0.0%', + 'decompressed_file': 'a', + }, + { + 'compressed_file': 'b.gz', + 'ratio': '0.0%', + 'decompressed_file': 'b', + } + ], + 'RESULT': [] +} diff --git a/moler/cmd/unix/seven_z.py b/moler/cmd/unix/seven_z.py new file mode 100644 index 000000000..9211c4940 --- /dev/null +++ b/moler/cmd/unix/seven_z.py @@ -0,0 +1,251 @@ +# -*- coding: utf-8 -*- +""" +7zip command module. +To get command instance from device use cmd_name 7z like: +dev.get_cmd(cmd_name='7z'...) +""" + +__author__ = 'Marcin Usielski' +__copyright__ = 'Copyright (C) 2024, Nokia' +__email__ = 'marcin.usielski@nokia.com' + + +import re + +from moler.cmd.unix.genericunix import GenericUnixCommand +from moler.exceptions import CommandFailure, ParsingDone +from moler.helpers import ClassProperty, convert_to_int + + +class SevenZ(GenericUnixCommand): + + def __init__(self, connection, options, archive_file, files=None, prompt=None, newline_chars=None, + runner=None): + """ + Command to work with 7z command. + + :param connection: moler connection to device, terminal when command is executed. + :param options: options of 7z command. + :param archive_file: archive file to work on. + :param files: files to work on. String with file(s) or list of files. + :param prompt: prompt on which command has to stop working. + :param newline_chars: characters to split lines. + :param runner: Runner to run command. + """ + super(SevenZ, self).__init__(connection=connection, prompt=prompt, newline_chars=newline_chars, runner=runner) + # Parameters defined by calling the command + self.options = options + self.archive_file = archive_file + self.files = files + self.ret_required = False + self._all_sent = False + + def build_command_string(self): + cmd = "7z" + if self.options: + cmd = f"{cmd} {self.options}" + if self.archive_file: + cmd = f"{cmd} {self.archive_file}" + if self.files is not None: + if isinstance(self.files, str): + cmd = f"{cmd} {self.files}" + else: + cmd = f"{cmd} {' '.join(self.files)}" + return cmd + + def on_new_line(self, line, is_full_line): + try: + self._parse_all(line=line) + if is_full_line: + self._parse_error_via_output_line(line) + self._parse_file(line) + except ParsingDone: + pass # line has been fully parsed by one of above parse-methods + if is_full_line: + self._all_sent = False + return super(SevenZ, self).on_new_line(line, is_full_line) + + _re_error_line = re.compile(r'(?P.*(: No more files|ERROR:).*)') + + def _parse_error_via_output_line(self, line): + if self._cmd_output_started and self._regex_helper.search_compiled(SevenZ._re_error_line, line): + self.set_exception(CommandFailure(self, f"ERROR: {self._regex_helper.group('error')}")) + raise ParsingDone() + + # ? (Y)es / (N)o / (A)lways / (S)kip all / A(u)to rename all / (Q)uit? A + _re_all = re.compile(r"\(Y\)es / \(N\)o / \(A\)lways / \(S\)kip all / A\(u\)to rename all / \(Q\)uit\?") + + def _parse_all(self, line: str) -> None: + if self._all_sent is False and self._regex_helper.search_compiled(SevenZ._re_all, line): + self.connection.send("a\n") + self._all_sent = True + raise ParsingDone() + + # 2024-07-26 12:04:29 ....A 0 0 b + _re_parse_file = re.compile(r"^(?P\d{4}-\d{2}-\d{2})\s+(?P