diff --git a/doc/source/conf.py b/doc/source/conf.py index 019efaf..be6a612 100644 --- a/doc/source/conf.py +++ b/doc/source/conf.py @@ -4,6 +4,7 @@ For the full list of built-in configuration values, see the documentation: https://www.sphinx-doc.org/en/master/usage/configuration.html """ + from pathlib import Path import sys @@ -53,7 +54,7 @@ autodoc_default_options = { "show-inheritance": True, "members": True, - "undoc-members": True + "undoc-members": True, } autoclass_content = "both" autodoc_preserve_defaults = True diff --git a/example/build_flex.py b/example/build_flex.py index 955aacb..b620b9a 100755 --- a/example/build_flex.py +++ b/example/build_flex.py @@ -23,14 +23,14 @@ f"git clone --depth 1 --branch {FLEX_VERSION} " f"https://github.com/westes/flex.git {FLEX_VERSION}", live_stdout=True, - live_stderr=True + live_stderr=True, ) sl.log( "Run `autogen`.", "./autogen.sh", cwd=Path.cwd() / FLEX_VERSION, live_stdout=True, - live_stderr=True + live_stderr=True, ) measure = ["cpu", "memory", "disk"] sl.log( @@ -39,7 +39,7 @@ cwd=Path.cwd() / FLEX_VERSION, live_stdout=True, live_stderr=True, - measure=measure + measure=measure, ) sl.log( "Build `libcompat.la`.", @@ -47,7 +47,7 @@ cwd=Path.cwd() / f"{FLEX_VERSION}/lib", live_stdout=True, live_stderr=True, - measure=measure + measure=measure, ) sl.log( "Build & install flex.", @@ -55,7 +55,7 @@ cwd=Path.cwd() / FLEX_VERSION, live_stdout=True, live_stderr=True, - measure=measure + measure=measure, ) sl.finalize() print(f"Open {sl.html_file} to view the log.") diff --git a/example/hello_world_html.py b/example/hello_world_html.py index c923886..d332bae 100755 --- a/example/hello_world_html.py +++ b/example/hello_world_html.py @@ -21,7 +21,7 @@ sl.log( "Tell everyone who you are, but from a different directory.", "whoami", - cwd=Path.cwd().parent + cwd=Path.cwd().parent, ) sl.finalize() print(f"Open {sl.html_file} to view the log.") diff --git a/example/hello_world_html_and_console.py b/example/hello_world_html_and_console.py index bb173dd..0001c76 100755 --- a/example/hello_world_html_and_console.py +++ b/example/hello_world_html_and_console.py @@ -13,8 +13,7 @@ from shell_logger import ShellLogger sl = ShellLogger( - "Hello World HTML and Console", - Path.cwd() / f"log_{Path(__file__).stem}" + "Hello World HTML and Console", Path.cwd() / f"log_{Path(__file__).stem}" ) sl.print( "This example demonstrates logging information both to the HTML log file " @@ -24,14 +23,14 @@ "Greet everyone to make them feel welcome.", "echo 'Hello World'", live_stdout=True, - live_stderr=True + live_stderr=True, ) sl.log( "Tell everyone who you are, but from a different directory.", "whoami", cwd=Path.cwd().parent, live_stdout=True, - live_stderr=True + live_stderr=True, ) sl.finalize() print(f"Open {sl.html_file} to view the log.") diff --git a/example/hello_world_html_with_stats.py b/example/hello_world_html_with_stats.py index 59f32b8..29f1469 100755 --- a/example/hello_world_html_with_stats.py +++ b/example/hello_world_html_with_stats.py @@ -13,8 +13,7 @@ from shell_logger import ShellLogger sl = ShellLogger( - "Hello World HTML with Stats", - Path.cwd() / f"log_{Path(__file__).stem}" + "Hello World HTML with Stats", Path.cwd() / f"log_{Path(__file__).stem}" ) sl.print( "This example demonstrates logging information solely to the HTML log " @@ -24,13 +23,13 @@ sl.log( "Greet everyone to make them feel welcome.", "echo 'Hello World'", - measure=measure + measure=measure, ) sl.log( "Tell everyone who you are, but from a different directory.", "whoami", cwd=Path.cwd().parent, - measure=measure + measure=measure, ) sl.finalize() print(f"Open {sl.html_file} to view the log.") diff --git a/shell_logger/abstract_method.py b/shell_logger/abstract_method.py index a11a52b..24411b9 100644 --- a/shell_logger/abstract_method.py +++ b/shell_logger/abstract_method.py @@ -24,7 +24,7 @@ def __init__(self): implemented for the class to be concrete. """ class_name = ( - inspect.stack()[1].frame.f_locals['self'].__class__.__name__ + inspect.stack()[1].frame.f_locals["self"].__class__.__name__ ) method_name = inspect.stack()[1].function super().__init__(f"`{class_name}` must implement `{method_name}()`.") diff --git a/shell_logger/html_utilities.py b/shell_logger/html_utilities.py index 67abdd9..948231d 100644 --- a/shell_logger/html_utilities.py +++ b/shell_logger/html_utilities.py @@ -42,8 +42,7 @@ def nested_simplenamespace_to_dict( return namespace elif isinstance(namespace, Mapping): return { - k: nested_simplenamespace_to_dict(v) - for k, v in namespace.items() + k: nested_simplenamespace_to_dict(v) for k, v in namespace.items() } # yapf: disable elif isinstance(namespace, Iterable): return [nested_simplenamespace_to_dict(x) for x in namespace] @@ -64,7 +63,7 @@ def get_human_time(milliseconds: float) -> str: A string representation of the date and time. """ seconds = milliseconds / 1000.0 - return datetime.fromtimestamp(seconds).strftime('%Y-%m-%d %H:%M:%S.%f') + return datetime.fromtimestamp(seconds).strftime("%Y-%m-%d %H:%M:%S.%f") def opening_html_text() -> str: @@ -98,8 +97,7 @@ def append_html(*args: Union[str, Iterator[str]], output: Path) -> None: """ def _append_html( - f: TextIO, - *inner_args: Union[str, bytes, Iterable] + f: TextIO, *inner_args: Union[str, bytes, Iterable] ) -> None: # yapf: disable """ Write some text to the given HTML log file. @@ -158,8 +156,7 @@ def flatten(element: Union[str, bytes, Iterable]) -> Iterator[str]: def parent_logger_card_html( - name: str, - *args: List[Iterator[str]] + name: str, *args: List[Iterator[str]] ) -> Iterator[str]: # yapf: disable """ Generate the HTML for the card corresponding to the parent @@ -176,9 +173,9 @@ def parent_logger_card_html( The header, followed by all the contents of the :class:`ShellLogger`, and then the footer. """ - header, indent, footer = split_template(parent_logger_template, - "parent_body", - name=name) + header, indent, footer = split_template( + parent_logger_template, "parent_body", name=name + ) yield header for arg in flatten(args): yield textwrap.indent(arg, indent) @@ -210,9 +207,7 @@ def child_logger_card(log) -> Iterator[str]: def child_logger_card_html( - name: str, - duration: str, - *args: Union[Iterator[str], List[Iterator[str]]] + name: str, duration: str, *args: Union[Iterator[str], List[Iterator[str]]] ) -> Iterator[str]: # yapf: disable """ Generate the HTML for a card corresponding to the child @@ -234,10 +229,9 @@ def child_logger_card_html( * Should we replace the ``for`` loop with the one found in :func:`parent_logger_card_html`? """ - header, indent, footer = split_template(child_logger_template, - "child_body", - name=name, - duration=duration) + header, indent, footer = split_template( + child_logger_template, "child_body", name=name, duration=duration + ) yield header for arg in args: if isinstance(arg, str): @@ -249,8 +243,7 @@ def child_logger_card_html( def command_card_html( - log: dict, - *args: Iterator[Union[str, Iterable]] + log: dict, *args: Iterator[Union[str, Iterable]] ) -> Iterator[str]: # yapf: disable """ Generate the HTML for a card corresponding to a command that was @@ -267,13 +260,15 @@ def command_card_html( The header, followed by all the contents of the command card, and then the footer. """ - header, indent, footer = split_template(command_template, - "more_info", - cmd_id=log["cmd_id"], - command=fixed_width(log["cmd"]), - message=log["msg"], - return_code=log["return_code"], - duration=log["duration"]) + header, indent, footer = split_template( + command_template, + "more_info", + cmd_id=log["cmd_id"], + command=fixed_width(log["cmd"]), + message=log["msg"], + return_code=log["return_code"], + duration=log["duration"], + ) yield header for arg in args: if isinstance(arg, str): @@ -300,21 +295,21 @@ def html_message_card(log: dict) -> Iterator[str]: """ timestamp = ( log["timestamp"] - .replace(' ', '_') - .replace(':', '-') - .replace('/', '_') - .replace('.', '-') + .replace(" ", "_") + .replace(":", "-") + .replace("/", "_") + .replace(".", "-") ) # yapf: disable header, indent, footer = split_template( html_message_template, "message", title=log["msg_title"], - timestamp=timestamp + timestamp=timestamp, ) text = html_encode(log["msg"]) - text = "
" + text.replace('\n', "
") + "
" + text = "
" + text.replace("\n", "
") + "
" yield header - yield textwrap.indent(text, indent) + '\n' + yield textwrap.indent(text, indent) + "\n" yield footer @@ -333,9 +328,9 @@ def message_card(log: dict) -> Iterator[str]: """ header, indent, footer = split_template(message_template, "message") text = html_encode(log["msg"]) - text = "
" + text.replace('\n', "
") + "
" + text = "
" + text.replace("\n", "
") + "
" yield header - yield textwrap.indent(text, indent) + '\n' + yield textwrap.indent(text, indent) + "\n" yield footer @@ -354,9 +349,9 @@ def command_detail_list(cmd_id: str, *args: Iterator[str]) -> Iterator[str]: The header, followed by each of the details associated with the command that was run, and then the footer. """ - header, indent, footer = split_template(command_detail_list_template, - "details", - cmd_id=cmd_id) + header, indent, footer = split_template( + command_detail_list_template, "details", cmd_id=cmd_id + ) yield header for arg in args: if isinstance(arg, str): @@ -365,10 +360,7 @@ def command_detail_list(cmd_id: str, *args: Iterator[str]) -> Iterator[str]: def command_detail( - cmd_id: str, - name: str, - value: str, - hidden: bool = False + cmd_id: str, name: str, value: str, hidden: bool = False ) -> str: """ Create the HTML snippet for a detail associated with a command that @@ -387,9 +379,7 @@ def command_detail( """ if hidden: return hidden_command_detail_template.format( - cmd_id=cmd_id, - name=name, - value=value + cmd_id=cmd_id, name=name, value=value ) else: return command_detail_template.format(name=name, value=value) @@ -428,7 +418,7 @@ def command_card(log: dict, stream_dir: Path) -> Iterator[str]: command_detail(cmd_id, "Group", log["group"], hidden=True), command_detail(cmd_id, "Shell", log["shell"], hidden=True), command_detail(cmd_id, "umask", log["umask"], hidden=True), - command_detail(cmd_id, "Return Code", log["return_code"]) + command_detail(cmd_id, "Return Code", log["return_code"]), ), output_block_card("stdout", stdout_path, cmd_id, collapsed=False), output_block_card("stderr", stderr_path, cmd_id, collapsed=False), @@ -455,7 +445,7 @@ def command_card(log: dict, stream_dir: Path) -> Iterator[str]: "/var/log", "/var/log/audit", "/boot", - "/boot/efi" + "/boot/efi", ] disk_stats = { x: y @@ -472,9 +462,7 @@ def command_card(log: dict, stream_dir: Path) -> Iterator[str]: def time_series_plot( - cmd_id: str, - data_tuples: List[Tuple[float, float]], - series_title: str + cmd_id: str, data_tuples: List[Tuple[float, float]], series_title: str ) -> Iterator[str]: # yapf: disable """ Create the HTML for a plot of time series data. @@ -495,9 +483,7 @@ def time_series_plot( def disk_time_series_plot( - cmd_id: str, - data_tuples: Tuple[float, float], - volume_name: str + cmd_id: str, data_tuples: Tuple[float, float], volume_name: str ) -> Iterator[str]: # yapf: disable """ Create the HTML for a plot of the disk usage time series data for a @@ -524,10 +510,7 @@ def disk_time_series_plot( def stat_chart_card( - labels: List[str], - data: List[float], - title: str, - identifier: str + labels: List[str], data: List[float], title: str, identifier: str ) -> Iterator[str]: """ Create the HTML for a two-dimensional plot. @@ -542,18 +525,12 @@ def stat_chart_card( A HTML snippet for the chart with all the details filled in. """ yield stat_chart_template.format( - labels=labels, - data=data, - title=title, - id=identifier + labels=labels, data=data, title=title, id=identifier ) def output_block_card( - title: str, - output: Union[Path, str], - cmd_id: str, - collapsed: bool = True + title: str, output: Union[Path, str], cmd_id: str, collapsed: bool = True ) -> Iterator[str]: # yapf: disable """ Given the output from a command, generate a corresponding HTML card @@ -571,16 +548,12 @@ def output_block_card( The header, followed by each line of the output, and then the footer. """ - name = title.replace(' ', '_').lower() + name = title.replace(" ", "_").lower() template = ( output_card_collapsed_template if collapsed else output_card_template ) header, indent, footer = split_template( - template, - "output_block", - name=name, - title=title, - cmd_id=cmd_id + template, "output_block", name=name, title=title, cmd_id=cmd_id ) yield header for line in output_block(output, name, cmd_id): @@ -589,9 +562,7 @@ def output_block_card( def output_block( - output: Union[Path, str], - name: str, - cmd_id: str + output: Union[Path, str], name: str, cmd_id: str ) -> Iterator[str]: # yapf: disable """ Given the output from a command, generate the HTML equivalent for @@ -631,9 +602,7 @@ def diagnostics_card(cmd_id: str, *args: Iterator[str]) -> Iterator[str]: and then the footer. """ header, indent, footer = split_template( - diagnostics_template, - "diagnostics", - cmd_id=cmd_id + diagnostics_template, "diagnostics", cmd_id=cmd_id ) yield header for arg in args: @@ -646,9 +615,7 @@ def diagnostics_card(cmd_id: str, *args: Iterator[str]) -> Iterator[str]: def output_block_html( - lines: Union[TextIO, str], - name: str, - cmd_id: str + lines: Union[TextIO, str], name: str, cmd_id: str ) -> Iterator[str]: # yapf: disable """ Given the output of a command, generate its HTML equivalent for @@ -665,12 +632,9 @@ def output_block_html( output, and then the footer. """ if isinstance(lines, str): - lines = lines.split('\n') + lines = lines.split("\n") header, indent, footer = split_template( - output_block_template, - "table_contents", - name=name, - cmd_id=cmd_id + output_block_template, "table_contents", name=name, cmd_id=cmd_id ) yield header line_no = 0 @@ -681,9 +645,7 @@ def output_block_html( def split_template( - template: str, - split_at: str, - **kwargs + template: str, split_at: str, **kwargs ) -> Tuple[str, str, str]: # yapf: disable """ Take a templated HTML snippet and split it into a header and footer, @@ -722,8 +684,7 @@ def split_template( """ fmt = {k: v for k, v in kwargs.items() if k != split_at} pattern = re.compile( - f"(.*\\n)(\\s*)\\{{{split_at}\\}}\\n(.*)", - flags=re.DOTALL + f"(.*\\n)(\\s*)\\{{{split_at}\\}}\\n(.*)", flags=re.DOTALL ) before, indent, after = pattern.search(template).groups() return before.format(**fmt), indent, after.format(**fmt) @@ -757,10 +718,7 @@ def html_encode(text: str) -> str: The encoded text. """ return sgr_to_html( - text - .replace('&', "&") - .replace('<', "<") - .replace('>', ">") + text.replace("&", "&").replace("<", "<").replace(">", ">") ) # yapf: disable @@ -779,7 +737,7 @@ def sgr_to_html(text: str) -> str: while text.find("\x1b[") >= 0: start = text.find("\x1b[") finish = text.find("m", start) - sgrs = text[start + 2:finish].split(';') + sgrs = text[start + 2 : finish].split(";") span_string = "" if len(sgrs) == 0: span_string += "" * span_count @@ -802,7 +760,7 @@ def sgr_to_html(text: str) -> str: span_count += 1 span_string += sgr_4bit_color_and_style_to_html(sgrs[0]) sgrs = sgrs[1:] - text = text[:start] + span_string + text[finish + 1:] + text = text[:start] + span_string + text[finish + 1 :] return text @@ -873,17 +831,17 @@ def sgr_8bit_color_to_html(sgr_params: List[str]) -> str: """ sgr_256 = int(sgr_params[2]) if len(sgr_params) > 2 else 0 if sgr_256 < 0 or sgr_256 > 255 or not sgr_params: - return '' + return "" if 15 < sgr_256 < 232: - red_6cube = (sgr_256-16) // 36 - green_6cube = (sgr_256 - (16 + red_6cube*36)) // 6 - blue_6cube = (sgr_256-16) % 6 + red_6cube = (sgr_256 - 16) // 36 + green_6cube = (sgr_256 - (16 + red_6cube * 36)) // 6 + blue_6cube = (sgr_256 - 16) % 6 red = str(51 * red_6cube) green = str(51 * green_6cube) blue = str(51 * blue_6cube) return sgr_24bit_color_to_html([sgr_params[0], "2", red, green, blue]) elif 231 < sgr_256 < 256: - gray = str(8 + (sgr_256-232) * 10) + gray = str(8 + (sgr_256 - 232) * 10) return sgr_24bit_color_to_html([sgr_params[0], "2", gray, gray, gray]) elif sgr_params[0] == "38": if sgr_256 < 8: @@ -914,7 +872,7 @@ def sgr_24bit_color_to_html(sgr_params: List[str]) -> str: elif len(sgr_params) > 1 and sgr_params[:2] == ["48", "2"]: return f'' else: - return '' + return "" def html_header() -> str: diff --git a/shell_logger/shell.py b/shell_logger/shell.py index d2ce3e6..04bb8c4 100644 --- a/shell_logger/shell.py +++ b/shell_logger/shell.py @@ -42,9 +42,7 @@ class Shell: """ def __init__( - self, - pwd: Path = Path.cwd(), - login_shell: bool = False + self, pwd: Path = Path.cwd(), login_shell: bool = False ) -> None: """ Initialize a :class:`Shell` object. @@ -65,24 +63,22 @@ def __init__( # Get the current flags of the file descriptors. aux_stdout_write_flags = fcntl.fcntl( - self.aux_stdout_wfd, - fcntl.F_GETFL + self.aux_stdout_wfd, fcntl.F_GETFL ) aux_stderr_write_flags = fcntl.fcntl( - self.aux_stderr_wfd, - fcntl.F_GETFL + self.aux_stderr_wfd, fcntl.F_GETFL ) # Make writes non-blocking. fcntl.fcntl( self.aux_stdout_wfd, fcntl.F_SETFL, - aux_stdout_write_flags | os.O_NONBLOCK + aux_stdout_write_flags | os.O_NONBLOCK, ) fcntl.fcntl( self.aux_stderr_wfd, fcntl.F_SETFL, - aux_stderr_write_flags | os.O_NONBLOCK + aux_stderr_write_flags | os.O_NONBLOCK, ) # Ensure the file descriptors are inheritable by the shell @@ -97,7 +93,7 @@ def __init__( stdin=self.aux_stdin_rfd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - close_fds=False + close_fds=False, ) os.set_inheritable(self.aux_stdout_wfd, False) os.set_inheritable(self.aux_stderr_wfd, False) @@ -115,7 +111,7 @@ def __del__(self) -> None: self.aux_stdout_rfd, self.aux_stdout_wfd, self.aux_stderr_rfd, - self.aux_stderr_wfd + self.aux_stderr_wfd, ]: try: os.close(fd) @@ -177,8 +173,7 @@ def run(self, command: str, **kwargs) -> SimpleNamespace: # Then run the command. if kwargs.get("devnull_stdin"): os.write( - self.aux_stdin_wfd, - f"{{\n{command}\n}} SimpleNamespace: output = self.tee( self.shell_subprocess.stdout, self.shell_subprocess.stderr, - **kwargs + **kwargs, ) # Note: If something goes wrong in `tee()`, the only way to reliably @@ -228,14 +223,12 @@ def run(self, command: str, **kwargs) -> SimpleNamespace: stderr=output.stderr_str, start=start, finish=finish, - wall=finish - start + wall=finish - start, ) @staticmethod def tee( - stdout: Optional[IO[bytes]], - stderr: Optional[IO[bytes]], - **kwargs + stdout: Optional[IO[bytes]], stderr: Optional[IO[bytes]], **kwargs ) -> SimpleNamespace: """ Split ``stdout`` and ``stderr`` file objects to write to @@ -308,7 +301,7 @@ def write(input_file: TextIO, output_files: List[TextIO]) -> None: # Close any open file descriptors and return the `stdout` and # `stderr`. - for file in (stdout_tee + stderr_tee): + for file in stdout_tee + stderr_tee: if ( file not in [None, sys.stdout, sys.stderr, sys.stdin] and not file.closed @@ -317,8 +310,7 @@ def write(input_file: TextIO, output_files: List[TextIO]) -> None: return SimpleNamespace(stdout_str=stdout_str, stderr_str=stderr_str) def auxiliary_command( - self, - **kwargs + self, **kwargs ) -> Tuple[Optional[str], Optional[str]]: # yapf: disable """ Run auxiliary commands like `umask`, `pwd`, `env`, etc. @@ -354,8 +346,7 @@ def auxiliary_command( while aux[-1] != 4: stdout += aux.decode() aux = os.read( - self.aux_stdout_rfd, - max_anonymous_pipe_buffer_size + self.aux_stdout_rfd, max_anonymous_pipe_buffer_size ) aux = aux[:-1] stdout += aux.decode() @@ -363,8 +354,7 @@ def auxiliary_command( while aux[-1] != 4: stderr += aux.decode() aux = os.read( - self.aux_stderr_rfd, - max_anonymous_pipe_buffer_size + self.aux_stderr_rfd, max_anonymous_pipe_buffer_size ) aux = aux[:-1] stderr += aux.decode() diff --git a/shell_logger/shell_logger.py b/shell_logger/shell_logger.py index 153d358..21a3387 100644 --- a/shell_logger/shell_logger.py +++ b/shell_logger/shell_logger.py @@ -22,7 +22,7 @@ message_card, command_card, child_logger_card, - parent_logger_card_html + parent_logger_card_html, ) from collections.abc import Iterable, Mapping from datetime import datetime, timedelta @@ -139,7 +139,7 @@ def __init__( log: Optional[List[object]] = None, init_time: Optional[datetime] = None, done_time: Optional[datetime] = None, - duration: Optional[str] = None + duration: Optional[str] = None, ) -> None: """ Initialize a :class:`ShellLogger` object. @@ -197,7 +197,7 @@ def __init__( self.stream_dir = Path( tempfile.mkdtemp( dir=self.log_dir, - prefix=self.init_time.strftime("%Y-%m-%d_%H.%M.%S.%f_") + prefix=self.init_time.strftime("%Y-%m-%d_%H.%M.%S.%f_"), ) ).resolve() else: @@ -206,13 +206,13 @@ def __init__( # Create (or append to) the HTML log file. if html_file is None: self.html_file = self.stream_dir / ( - self.name.replace(' ', '_') + '.html' + self.name.replace(" ", "_") + ".html" ) # yapf: disable else: self.html_file = html_file.resolve() if self.is_parent(): if self.html_file.exists(): - with open(self.html_file, 'a') as f: + with open(self.html_file, "a") as f: f.write( f"" ) @@ -246,8 +246,7 @@ def __update_duration(self) -> None: """ self.update_done_time() self.duration = self.strfdelta( - self.done_time - self.init_time, - "{hrs}h {min}m {sec}s" + self.done_time - self.init_time, "{hrs}h {min}m {sec}s" ) def check_duration(self) -> str: @@ -259,8 +258,7 @@ def check_duration(self) -> str: A string representation of the total duration. """ return self.strfdelta( - datetime.now() - self.init_time, - "{hrs}h {min}m {sec}s" + datetime.now() - self.init_time, "{hrs}h {min}m {sec}s" ) def change_log_dir(self, new_log_dir: Path) -> None: @@ -289,8 +287,8 @@ def change_log_dir(self, new_log_dir: Path) -> None: # Change the `stream_dir`, `html_file`, and `log_dir` for every # child `ShellLogger` recursively. - self.stream_dir = ( - new_log_dir / self.stream_dir.relative_to(self.log_dir) + self.stream_dir = new_log_dir / self.stream_dir.relative_to( + self.log_dir ) self.html_file = new_log_dir / self.html_file.relative_to(self.log_dir) self.log_dir = new_log_dir.resolve() @@ -321,7 +319,7 @@ def add_child(self, child_name: str) -> ShellLogger: self.stream_dir, self.html_file, self.indent + 1, - self.login_shell + self.login_shell, ) self.log_book.append(child) return child @@ -341,27 +339,27 @@ def strfdelta(delta: timedelta, fmt: str) -> str: """ # Dictionary to hold time delta info. - d = {'days': delta.days} + d = {"days": delta.days} microseconds_per_second = 10**6 seconds_per_minute = 60 minutes_per_hour = 60 - total_ms = ( - delta.microseconds + delta.seconds * microseconds_per_second + total_ms = delta.microseconds + delta.seconds * microseconds_per_second + d["hrs"], rem = divmod( + total_ms, + (minutes_per_hour * seconds_per_minute * microseconds_per_second), ) - d['hrs'], rem = divmod(total_ms, (minutes_per_hour - * seconds_per_minute - * microseconds_per_second)) - d['min'], rem = divmod(rem, (seconds_per_minute - * microseconds_per_second)) - d['sec'] = rem / microseconds_per_second + d["min"], rem = divmod( + rem, (seconds_per_minute * microseconds_per_second) + ) + d["sec"] = rem / microseconds_per_second # Round to 2 decimals - d['sec'] = round(d['sec'], 2) + d["sec"] = round(d["sec"], 2) # String template to help with recognizing the format. return fmt.format(**d) - def print(self, msg: str, end: str = '\n') -> None: + def print(self, msg: str, end: str = "\n") -> None: """ Print a message and save it to the log. @@ -370,7 +368,7 @@ def print(self, msg: str, end: str = '\n') -> None: end: The string appended after the message: """ print(msg, end=end) - log = {'msg': msg, 'timestamp': str(datetime.now()), 'cmd': None} + log = {"msg": msg, "timestamp": str(datetime.now()), "cmd": None} self.log_book.append(log) def html_print(self, msg: str, msg_title: str = "HTML Message") -> None: @@ -382,10 +380,10 @@ def html_print(self, msg: str, msg_title: str = "HTML Message") -> None: msg_title: Title of the message to save to the log. """ log = { - 'msg': msg, - 'msg_title': msg_title, - 'timestamp': str(datetime.now()), - 'cmd': None + "msg": msg, + "msg_title": msg_title, + "timestamp": str(datetime.now()), + "cmd": None, } self.log_book.append(log) @@ -436,16 +434,16 @@ def finalize(self) -> None: """ if self.is_parent(): html_text = opening_html_text() + "\n" - with open(self.html_file, 'w') as f: + with open(self.html_file, "w") as f: f.write(html_text) for element in self.to_html(): append_html(element, output=self.html_file) if self.is_parent(): - with open(self.html_file, 'a') as html: + with open(self.html_file, "a") as html: html.write(closing_html_text()) - html.write('\n') + html.write("\n") # Create a symlink in `log_dir` to the HTML file in # `stream_dir`. @@ -458,15 +456,11 @@ def finalize(self) -> None: # Save everything to a JSON file in the timestamped # `stream_dir`. json_file = self.stream_dir / ( - self.name.replace(' ', '_') + '.json' + self.name.replace(" ", "_") + ".json" ) # yapf: disable - with open(json_file, 'w') as jf: + with open(json_file, "w") as jf: json.dump( - self, - jf, - cls=ShellLoggerEncoder, - sort_keys=True, - indent=4 + self, jf, cls=ShellLoggerEncoder, sort_keys=True, indent=4 ) def log( @@ -479,7 +473,7 @@ def log( return_info: bool = False, verbose: bool = False, stdin_redirect: bool = True, - **kwargs + **kwargs, ) -> dict: """ Execute a command, and log the corresponding information. @@ -527,7 +521,7 @@ def log( # Create a unique command ID that will be used to find the # location of the `stdout`/`stderr` files in the temporary # directory during finalization. - cmd_id = 'cmd_' + ''.join( + cmd_id = "cmd_" + "".join( random.choice(string.ascii_lowercase) for _ in range(9) ) @@ -537,23 +531,24 @@ def log( stderr_path = self.stream_dir / f"{time_str}_{cmd_id}_stderr" trace_path = ( self.stream_dir / f"{time_str}_{cmd_id}_trace" - if kwargs.get("trace") else None + if kwargs.get("trace") + else None ) # yapf: disable # Print the command to be executed. - with open(stdout_path, 'a'), open(stderr_path, 'a'): + with open(stdout_path, "a"), open(stderr_path, "a"): if verbose: print(cmd) # Initialize the log information. log = { - 'msg': msg, - 'duration': None, - 'timestamp': start_time.strftime("%Y-%m-%d_%H%M%S"), - 'cmd': cmd, - 'cmd_id': cmd_id, - 'cwd': cwd, - 'return_code': 0 + "msg": msg, + "duration": None, + "timestamp": start_time.strftime("%Y-%m-%d_%H%M%S"), + "cmd": cmd, + "cmd_id": cmd_id, + "cwd": cwd, + "return_code": 0, } # Execute the command. @@ -569,7 +564,7 @@ def log( trace_path=trace_path, devnull_stdin=stdin_redirect, pwd=cwd, - **kwargs + **kwargs, ) # Update the log information and save it to the `log_book`. @@ -581,9 +576,9 @@ def log( log = {**log, **nested_simplenamespace_to_dict(result)} self.log_book.append(log) return { - 'return_code': log['return_code'], - 'stdout': result.stdout, - 'stderr': result.stderr + "return_code": log["return_code"], + "stdout": result.stdout, + "stderr": result.stderr, } def _run(self, command: str, **kwargs) -> SimpleNamespace: @@ -656,8 +651,7 @@ def _run(self, command: str, **kwargs) -> SimpleNamespace: if kwargs.get("pwd"): self.shell.cd(old_pwd) return SimpleNamespace( - **completed_process.__dict__, - **aux_info.__dict__ + **completed_process.__dict__, **aux_info.__dict__ ) def auxiliary_information(self) -> SimpleNamespace: @@ -672,12 +666,14 @@ def auxiliary_information(self) -> SimpleNamespace: pwd, _ = self.shell.auxiliary_command(posix="pwd", strip=True) environment, _ = self.shell.auxiliary_command(posix="env") umask, _ = self.shell.auxiliary_command(posix="umask", strip=True) - hostname, _ = self.shell.auxiliary_command(posix="hostname", - strip=True) + hostname, _ = self.shell.auxiliary_command( + posix="hostname", strip=True + ) user, _ = self.shell.auxiliary_command(posix="whoami", strip=True) group, _ = self.shell.auxiliary_command(posix="id -gn", strip=True) - shell, _ = self.shell.auxiliary_command(posix="printenv SHELL", - strip=True) + shell, _ = self.shell.auxiliary_command( + posix="printenv SHELL", strip=True + ) ulimit, _ = self.shell.auxiliary_command(posix="ulimit -a") return SimpleNamespace( pwd=pwd, @@ -687,7 +683,7 @@ def auxiliary_information(self) -> SimpleNamespace: user=user, group=group, shell=shell, - ulimit=ulimit + ulimit=ulimit, ) @@ -716,32 +712,32 @@ def default(self, obj: object) -> object: """ if isinstance(obj, ShellLogger): return { - **{'__type__': 'ShellLogger'}, - **{k: self.default(v) for k, v in obj.__dict__.items()} + **{"__type__": "ShellLogger"}, + **{k: self.default(v) for k, v in obj.__dict__.items()}, } # yapf: disable elif isinstance(obj, (int, float, str, bytes)): return obj elif isinstance(obj, Mapping): return {k: self.default(v) for k, v in obj.items()} elif isinstance(obj, tuple): - return {'__type__': 'tuple', 'items': obj} + return {"__type__": "tuple", "items": obj} elif isinstance(obj, Iterable): return [self.default(x) for x in obj] elif isinstance(obj, datetime): return { - '__type__': 'datetime', - 'value': obj.strftime('%Y-%m-%d_%H:%M:%S:%f'), - 'format': '%Y-%m-%d_%H:%M:%S:%f' + "__type__": "datetime", + "value": obj.strftime("%Y-%m-%d_%H:%M:%S:%f"), + "format": "%Y-%m-%d_%H:%M:%S:%f", } elif isinstance(obj, Path): - return {'__type__': 'Path', 'value': str(obj)} + return {"__type__": "Path", "value": str(obj)} elif obj is None: return None elif isinstance(obj, Shell): return { "__type__": "Shell", "pwd": obj.pwd(), - "login_shell": obj.login_shell + "login_shell": obj.login_shell, } else: return json.JSONEncoder.default(self, obj) @@ -778,9 +774,9 @@ def dict_to_object(obj: dict) -> object: Returns: The object represented by the JSON serialization. """ - if '__type__' not in obj: + if "__type__" not in obj: return obj - elif obj['__type__'] == 'ShellLogger': + elif obj["__type__"] == "ShellLogger": logger = ShellLogger( obj["name"], obj["log_dir"], @@ -791,14 +787,14 @@ def dict_to_object(obj: dict) -> object: obj["log_book"], obj["init_time"], obj["done_time"], - obj["duration"] + obj["duration"], ) return logger - elif obj['__type__'] == 'datetime': - return datetime.strptime(obj['value'], obj['format']) - elif obj['__type__'] == 'Path': - return Path(obj['value']) - elif obj['__type__'] == 'tuple': - return tuple(obj['items']) - elif obj['__type__'] == 'Shell': - return Shell(Path(obj['pwd']), obj["login_shell"]) + elif obj["__type__"] == "datetime": + return datetime.strptime(obj["value"], obj["format"]) + elif obj["__type__"] == "Path": + return Path(obj["value"]) + elif obj["__type__"] == "tuple": + return tuple(obj["items"]) + elif obj["__type__"] == "Shell": + return Shell(Path(obj["pwd"]), obj["login_shell"]) diff --git a/shell_logger/stats_collector.py b/shell_logger/stats_collector.py index 3edadbd..2d08217 100644 --- a/shell_logger/stats_collector.py +++ b/shell_logger/stats_collector.py @@ -19,6 +19,7 @@ from pathlib import Path from time import sleep, time from typing import List, Tuple + try: import psutil except ModuleNotFoundError: @@ -53,6 +54,7 @@ class StatsCollector: Provides an interface for the :class:`ShellLogger` to run commands while collecting various system statistics. """ + stat_name = "undefined" # Should be defined by subclasses. subclasses = [] @@ -138,6 +140,7 @@ class DiskStatsCollector(StatsCollector): A means of running commands while collecting disk usage statistics. """ + stat_name = "disk" def __init__(self, interval: float, manager: SyncManager) -> None: @@ -157,7 +160,7 @@ def __init__(self, interval: float, manager: SyncManager) -> None: for location in [ "/tmp", "/dev/shm", - f"/var/run/user/{os.getuid()}" + f"/var/run/user/{os.getuid()}", ]: if ( location not in self.mount_points @@ -193,6 +196,7 @@ class CPUStatsCollector(StatsCollector): A means of running commands while collecting CPU usage statistics. """ + stat_name = "cpu" def __init__(self, interval: float, manager: SyncManager) -> None: @@ -231,6 +235,7 @@ class MemoryStatsCollector(StatsCollector): A means of running commands while collecting memory usage statistics. """ + stat_name = "memory" def __init__(self, interval: float, manager: SyncManager) -> None: @@ -273,6 +278,7 @@ class DiskStatsCollector(StatsCollector): A phony :class:`DiskStatsCollector` used when ``psutil`` is unavailable. This collects no disk statistics. """ + stat_name = "disk" def __init__(self, interval: float, manager: SyncManager) -> None: @@ -307,6 +313,7 @@ class CPUStatsCollector(StatsCollector): A phony :class:`CPUStatsCollector` used when ``psutil`` is unavailable. This collects no CPU statistics. """ + stat_name = "cpu" def __init__(self, interval: float, manager: SyncManager) -> None: @@ -341,6 +348,7 @@ class MemoryStatsCollector(StatsCollector): A phony :class:`MemoryStatsCollector` used when ``psutil`` is unavailable. This collects no memory statistics. """ + stat_name = "memory" def __init__(self, interval: float, manager: SyncManager) -> None: diff --git a/shell_logger/trace.py b/shell_logger/trace.py index a0c461b..23364f2 100644 --- a/shell_logger/trace.py +++ b/shell_logger/trace.py @@ -43,6 +43,7 @@ class Trace: Provides an interface for the :class:`ShellLogger` to run commands with a certain trace (e.g., ``strace`` or ``ltrace``). """ + trace_name = "undefined" # Should be defined by subclasses. subclasses = [] @@ -97,6 +98,7 @@ class STrace(Trace): An interface between :class:`ShellLogger` and the ``strace`` command. """ + trace_name = "strace" def __init__(self, **kwargs) -> None: @@ -126,6 +128,7 @@ class LTrace(Trace): An interface between :class:`ShellLogger` and the ``ltrace`` command. """ + trace_name = "ltrace" def __init__(self, **kwargs): diff --git a/test/test_shell_logger.py b/test/test_shell_logger.py index 9a383cf..50bee35 100644 --- a/test/test_shell_logger.py +++ b/test/test_shell_logger.py @@ -17,6 +17,7 @@ import pytest import re from shell_logger import ShellLogger, ShellLoggerDecoder + try: import psutil except ModuleNotFoundError: @@ -55,7 +56,7 @@ def shell_logger() -> ShellLogger: """ # Initialize a parent `ShellLogger`. - parent = ShellLogger('Parent', Path.cwd()) + parent = ShellLogger("Parent", Path.cwd()) # Run the command. # `stdout` ; `stderr` @@ -66,11 +67,9 @@ def shell_logger() -> ShellLogger: measure = ["cpu", "memory", "disk"] kwargs = {"measure": measure, "return_info": True, "interval": 0.1} if os.uname().sysname == "Linux": - kwargs.update({ - "trace": "ltrace", - "expression": "setlocale", - "summary": True - }) + kwargs.update( + {"trace": "ltrace", "expression": "setlocale", "summary": True} + ) else: print( f"Warning: uname is not 'Linux': {os.uname()}; ltrace not tested." @@ -109,11 +108,11 @@ def test_initialization_creates_html_file() -> None: logger = ShellLogger(stack()[0][3], Path.cwd()) timestamp = logger.init_time.strftime("%Y-%m-%d_%H.%M.%S.%f") streamm_dir = next(Path.cwd().glob(f"{timestamp}_*")) - assert (streamm_dir / f'{stack()[0][3]}.html').exists() + assert (streamm_dir / f"{stack()[0][3]}.html").exists() def test_log_method_creates_tmp_stdout_stderr_files( - shell_logger: ShellLogger + shell_logger: ShellLogger, ) -> None: """ Verify that logging a command will create files in the @@ -125,8 +124,8 @@ def test_log_method_creates_tmp_stdout_stderr_files( """ # Get the paths for the `stdout`/`stderr` files. - cmd_id = shell_logger.log_book[0]['cmd_id'] - cmd_ts = shell_logger.log_book[0]['timestamp'] + cmd_id = shell_logger.log_book[0]["cmd_id"] + cmd_ts = shell_logger.log_book[0]["timestamp"] stdout_file = shell_logger.stream_dir / f"{cmd_ts}_{cmd_id}_stdout" stderr_file = shell_logger.stream_dir / f"{cmd_ts}_{cmd_id}_stderr" assert stdout_file.exists() @@ -135,14 +134,14 @@ def test_log_method_creates_tmp_stdout_stderr_files( print(f"{stderr_file}") # Make sure the information written to these files is correct. - with open(stdout_file, 'r') as out, open(stderr_file, 'r') as err: + with open(stdout_file, "r") as out, open(stderr_file, "r") as err: out_txt = out.readline() err_txt = err.readline() - assert 'Hello world out' in out_txt - assert 'Hello world error' in err_txt + assert "Hello world out" in out_txt + assert "Hello world error" in err_txt -@pytest.mark.parametrize('return_info', [True, False]) +@pytest.mark.parametrize("return_info", [True, False]) def test_log_method_return_info_works_correctly(return_info: bool) -> None: """ **@pytest.mark.parametrize('return_info', [True, False])** @@ -162,21 +161,19 @@ def test_log_method_return_info_works_correctly(return_info: bool) -> None: cmd = "echo 'Hello world out'; echo 'Hello world error' 1>&2" result = logger.log("test cmd", cmd, Path.cwd(), return_info=return_info) if return_info: - assert "Hello world out" in result['stdout'] - assert "Hello world error" in result['stderr'] - assert result['return_code'] == 0 + assert "Hello world out" in result["stdout"] + assert "Hello world error" in result["stderr"] + assert result["return_code"] == 0 else: - assert result['stdout'] is None - assert result['stderr'] is None - assert result['return_code'] == 0 + assert result["stdout"] is None + assert result["stderr"] is None + assert result["return_code"] == 0 -@pytest.mark.parametrize('live_stdout', [True, False]) -@pytest.mark.parametrize('live_stderr', [True, False]) +@pytest.mark.parametrize("live_stdout", [True, False]) +@pytest.mark.parametrize("live_stderr", [True, False]) def test_log_method_live_stdout_stderr_works_correctly( - capsys: CaptureFixture, - live_stdout: bool, - live_stderr: bool + capsys: CaptureFixture, live_stdout: bool, live_stderr: bool ) -> None: """ Verify that the ``live_stdout`` and ``live_stdout`` flags work as @@ -204,7 +201,7 @@ def test_log_method_live_stdout_stderr_works_correctly( def test_child_logger_duration_displayed_correctly_in_html( - shell_logger: ShellLogger + shell_logger: ShellLogger, ) -> None: """ Verify that the overview of child loggers in the HTML file displays @@ -219,7 +216,7 @@ def test_child_logger_duration_displayed_correctly_in_html( child3 = shell_logger.add_child("Child 3") child3.log("Wait 0.006s", "sleep 0.006") shell_logger.finalize() - with open(shell_logger.html_file, 'r') as hf: + with open(shell_logger.html_file, "r") as hf: html_text = hf.read() assert child2.duration is not None assert f"Duration: {child2.duration}" in html_text @@ -228,7 +225,7 @@ def test_child_logger_duration_displayed_correctly_in_html( def test_finalize_creates_json_with_correct_information( - shell_logger: ShellLogger + shell_logger: ShellLogger, ) -> None: """ Verify that the :func:`finalize` method creates a JSON file with the @@ -242,7 +239,7 @@ def test_finalize_creates_json_with_correct_information( # Load from JSON. json_file = shell_logger.stream_dir / "Parent.json" assert json_file.exists() - with open(json_file, 'r') as jf: + with open(json_file, "r") as jf: loaded_logger = json.load(jf, cls=ShellLoggerDecoder) # Parent `ShellLogger`. @@ -269,7 +266,7 @@ def test_finalize_creates_json_with_correct_information( def test_finalize_creates_html_with_correct_information( - shell_logger: ShellLogger + shell_logger: ShellLogger, ) -> None: """ Verify that the :func:`finalize` method creates an HTML file with @@ -283,7 +280,7 @@ def test_finalize_creates_html_with_correct_information( # Load the HTML file. html_file = shell_logger.stream_dir / "Parent.html" assert html_file.exists() - with open(html_file, 'r') as hf: + with open(html_file, "r") as hf: html_text = hf.read() # Ensure the command information is present. @@ -326,7 +323,7 @@ def test_finalize_creates_html_with_correct_information( def test_log_dir_html_symlinks_to_stream_dir_html( - shell_logger: ShellLogger + shell_logger: ShellLogger, ) -> None: """ Verify that the :func:`finalize` method symlinks @@ -346,7 +343,7 @@ def test_log_dir_html_symlinks_to_stream_dir_html( def test_json_file_can_reproduce_html_file( - shell_logger: ShellLogger + shell_logger: ShellLogger, ) -> None: # yapf: disable """ Verify that a JSON file can properly recreate the original HTML file @@ -360,7 +357,7 @@ def test_json_file_can_reproduce_html_file( # Load the original HTML file's contents. html_file = shell_logger.log_dir / "Parent.html" assert html_file.exists() - with open(html_file, 'r') as hf: + with open(html_file, "r") as hf: original_html = hf.read() # Delete the HTML file. @@ -369,7 +366,7 @@ def test_json_file_can_reproduce_html_file( # Load the JSON data. json_file = shell_logger.stream_dir / "Parent.json" assert json_file.exists() - with open(json_file, 'r') as jf: + with open(json_file, "r") as jf: loaded_logger = json.load(jf, cls=ShellLoggerDecoder) # Finalize the loaded `ShellLogger` object. @@ -377,7 +374,7 @@ def test_json_file_can_reproduce_html_file( # Load the new HTML file's contents and compare. assert html_file.exists() - with open(html_file, 'r') as hf: + with open(html_file, "r") as hf: new_html = hf.read() print(f"New Read: {html_file.resolve()}") assert original_html == new_html @@ -453,8 +450,7 @@ def test_logger_does_not_store_stdout_string_by_default() -> None: @pytest.mark.skipif( - os.uname().sysname == "Darwin", - reason="`ltrace` doesn't exist for Darwin" + os.uname().sysname == "Darwin", reason="`ltrace` doesn't exist for Darwin" ) def test_logger_does_not_store_trace_string_by_default() -> None: """ @@ -468,7 +464,7 @@ def test_logger_does_not_store_trace_string_by_default() -> None: "echo hello", Path.cwd(), return_info=True, - trace="ltrace" + trace="ltrace", ) assert logger.log_book[1]["trace"] is not None @@ -587,9 +583,9 @@ def test_trace_expression() -> None: """ logger = ShellLogger(stack()[0][3], Path.cwd()) if os.uname().sysname == "Linux": - result = logger._run("echo hello", trace="ltrace", expression='getenv') + result = logger._run("echo hello", trace="ltrace", expression="getenv") assert 'getenv("POSIXLY_CORRECT")' in result.trace - assert result.trace.count('\n') == 2 + assert result.trace.count("\n") == 2 else: print( f"Warning: uname is not 'Linux': {os.uname()}; ltrace expression " @@ -626,19 +622,13 @@ def test_trace_expression_and_summary() -> None: if os.uname().sysname == "Linux": echo_location = logger._run("which echo").stdout.strip() result = logger._run( - "echo hello", - trace="strace", - expression="execve", - summary=True + "echo hello", trace="strace", expression="execve", summary=True ) assert f'execve("{echo_location}' not in result.trace assert "execve" in result.trace assert "getenv" not in result.trace result = logger._run( - "echo hello", - trace="ltrace", - expression="getenv", - summary=True + "echo hello", trace="ltrace", expression="getenv", summary=True ) assert 'getenv("POSIXLY_CORRECT")' not in result.trace assert "getenv" in result.trace @@ -685,7 +675,7 @@ def test_trace_and_stats() -> None: interval=0.1, trace="ltrace", expression="setlocale", - summary=True + summary=True, ) assert "setlocale" in result.trace assert "sleep" not in result.trace @@ -714,7 +704,7 @@ def test_trace_and_stat() -> None: interval=0.1, trace="ltrace", expression="setlocale", - summary=True + summary=True, ) assert "setlocale" in result.trace assert "sleep" not in result.trace @@ -729,7 +719,7 @@ def test_trace_and_stat() -> None: @pytest.mark.skipif( os.uname().sysname == "Darwin", - reason="`ltrace`/`strace` don't exist for Darwin" + reason="`ltrace`/`strace` don't exist for Darwin", ) @pytest.mark.skip(reason="Not sure it's worth it to fix this or not") def test_set_env_trace() -> None: @@ -759,7 +749,7 @@ def test_log_book_trace_and_stats() -> None: interval=0.1, trace="ltrace", expression="setlocale", - summary=True + summary=True, ) assert "setlocale" in logger.log_book[0]["trace"] assert "sleep" not in logger.log_book[0]["trace"] @@ -825,7 +815,7 @@ def test_sgr_gets_converted_to_html() -> None: # Load the HTML file and make sure it checks out. html_file = logger.stream_dir / f"{logger.name}.html" assert html_file.exists() - with open(html_file, 'r') as hf: + with open(html_file, "r") as hf: html_text = hf.read() assert "\x1B" not in html_text assert ">Hello" in html_text @@ -849,8 +839,7 @@ def test_html_print(capsys: CaptureFixture) -> None: """ logger = ShellLogger(stack()[0][3], Path.cwd()) logger.html_print( - "The quick brown fox jumps over the lazy dog.", - msg_title="Brown Fox" + "The quick brown fox jumps over the lazy dog.", msg_title="Brown Fox" ) logger.print("The quick orange zebra jumps over the lazy dog.") out, err = capsys.readouterr() @@ -859,7 +848,7 @@ def test_html_print(capsys: CaptureFixture) -> None: # Load the HTML file and make sure it checks out. html_file = logger.stream_dir / f"{logger.name}.html" assert html_file.exists() - with open(html_file, 'r') as hf: + with open(html_file, "r") as hf: html_text = hf.read() assert "brown fox" not in out assert "brown fox" not in err @@ -909,7 +898,7 @@ def test_append_mode() -> None: # Load the HTML file and ensure it checks out. html_file = logger1.stream_dir / f"{logger1.name}.html" assert html_file.exists() - with open(html_file, 'r') as hf: + with open(html_file, "r") as hf: html_text = hf.read() assert "once" in html_text assert "ONCE" in html_text @@ -936,6 +925,6 @@ def test_invalid_decodings() -> None: result = logger.log( "Print invalid start byte for bytes decode()", "printf '\\xFDHello\\n'", - return_info=True + return_info=True, ) assert result["stdout"] == "Hello\n"