diff --git a/src/dvsim/job/deploy.py b/src/dvsim/job/deploy.py index c620ddbd..8a09abdb 100644 --- a/src/dvsim/job/deploy.py +++ b/src/dvsim/job/deploy.py @@ -17,7 +17,7 @@ from dvsim.job.time import JobTime from dvsim.launcher.base import Launcher from dvsim.logging import log -from dvsim.sim_utils import get_cov_summary_table +from dvsim.tool.utils import get_sim_tool_plugin from dvsim.utils import ( clean_odirs, find_and_substitute_wildcards, @@ -813,7 +813,11 @@ def callback(status: str) -> None: if self.dry_run or status != "P": return - results, self.cov_total = get_cov_summary_table(self.cov_report_txt, self.sim_cfg.tool) + plugin = get_sim_tool_plugin(tool=self.sim_cfg.tool) + + results, self.cov_total = plugin.get_cov_summary_table( + cov_report_path=self.cov_report_txt, + ) colalign = ("center",) * len(results[0]) self.cov_results = tabulate( diff --git a/src/dvsim/launcher/base.py b/src/dvsim/launcher/base.py index 69850bcb..f0e475d3 100644 --- a/src/dvsim/launcher/base.py +++ b/src/dvsim/launcher/base.py @@ -17,7 +17,7 @@ from dvsim.job.time import JobTime from dvsim.logging import log -from dvsim.sim_utils import get_job_runtime, get_simulated_time +from dvsim.tool.utils import get_sim_tool_plugin from dvsim.utils import clean_odirs, mk_symlink, rm_path if TYPE_CHECKING: @@ -339,11 +339,10 @@ def _find_patterns(patterns: Sequence[str], line: str) -> Sequence[str] | None: # since it is devoid of the delays incurred due to infrastructure and # setup overhead. + plugin = get_sim_tool_plugin(tool=self.job_spec.tool) + try: - time, unit = get_job_runtime( - log_text=lines, - tool=self.job_spec.tool, - ) + time, unit = plugin.get_job_runtime(log_text=lines) self.job_runtime.set(time, unit) except RuntimeError as e: @@ -354,10 +353,7 @@ def _find_patterns(patterns: Sequence[str], line: str) -> Sequence[str] | None: if self.job_spec.job_type == "RunTest": try: - time, unit = get_simulated_time( - log_text=lines, - tool=self.job_spec.tool, - ) + time, unit = plugin.get_simulated_time(log_text=lines) self.simulated_time.set(time, unit) except RuntimeError as e: diff --git a/src/dvsim/sim_utils.py b/src/dvsim/sim_utils.py deleted file mode 100644 index 48ee6c87..00000000 --- a/src/dvsim/sim_utils.py +++ /dev/null @@ -1,282 +0,0 @@ -# Copyright lowRISC contributors (OpenTitan project). -# Licensed under the Apache License, Version 2.0, see LICENSE for details. -# SPDX-License-Identifier: Apache-2.0 - -"""Common DV simulation specific utilities.""" - -import re -from collections import OrderedDict -from collections.abc import Sequence -from io import TextIOWrapper -from pathlib import Path - - -def get_cov_summary_table( - txt_cov_report: Path, - tool: str, -) -> tuple[Sequence[Sequence[str]], str]: - """Capture the summary results as a list of lists. - - The text coverage report is passed as input to the function, in addition to - the tool used. - - Returns: - tuple of, List of metrics and values, and final coverage total - - Raises: - the appropriate exception if the coverage summary extraction fails. - - """ - with Path(txt_cov_report).open() as f: - if tool == "xcelium": - return xcelium_cov_summary_table(f) - - if tool == "vcs": - return vcs_cov_summary_table(f) - - msg = f"{tool} is unsupported for cov extraction." - raise NotImplementedError(msg) - - -# Same desc as above, but specific to Xcelium and takes an opened input stream. -def xcelium_cov_summary_table(buf: TextIOWrapper) -> tuple[Sequence[Sequence[str]], str]: - """Capture the summary results as a list of lists from Xcelium.""" - for line in buf: - if "name" in line: - # Strip the line and remove the unwanted "* Covered" string. - metrics = line.strip().replace("* Covered", "").split() - # Change first item to 'Score'. - metrics[0] = "Score" - - # Gather the list of metrics. - items = OrderedDict() - for metric in metrics: - items[metric] = {} - items[metric]["covered"] = 0 - items[metric]["total"] = 0 - - # Next line is a separator. - line = buf.readline() - - # Subsequent lines are coverage items to be aggregated. - for line in buf: - line = re.sub(r"%\s+\(", "%(", line) - values = line.strip().split() - for i, value in enumerate(values): - value = value.strip() - m = re.search(r"\((\d+)/(\d+).*\)", value) - if m: - items[metrics[i]]["covered"] += int(m.group(1)) - items[metrics[i]]["total"] += int(m.group(2)) - items["Score"]["covered"] += int(m.group(1)) - items["Score"]["total"] += int(m.group(2)) - - # Capture the percentages and the aggregate. - values = [] - cov_total = None - for metric in items: - if items[metric]["total"] == 0: - values.append("-- %") - else: - value = items[metric]["covered"] / items[metric]["total"] * 100 - value = f"{round(value, 2):.2f} %" - values.append(value) - if metric == "Score": - cov_total = value - - return [items.keys(), values], cov_total - - # If we reached here, then we were unable to extract the coverage. - msg = f"Coverage data not found in {buf.name}!" - raise SyntaxError(msg) - - -# Same desc as above, but specific to VCS and takes an opened input stream. -def vcs_cov_summary_table(buf: TextIOWrapper) -> tuple[Sequence[Sequence[str]], str]: - """Capture the summary results as a list of lists from VCS.""" - for line in buf: - match = re.match("total coverage summary", line, re.IGNORECASE) - if match: - # Metrics on the next line. - line = buf.readline().strip() - metrics = line.split() - # Values on the next. - line = buf.readline().strip() - # Pretty up the values - add % sign for ease of post - # processing. - values = [] - for val in line.split(): - val += " %" - values.append(val) - # first row is coverage total - cov_total = values[0] - return [metrics, values], cov_total - - # If we reached here, then we were unable to extract the coverage. - msg = f"Coverage data not found in {buf.name}!" - raise SyntaxError(msg) - - -def get_job_runtime(log_text: list, tool: str) -> tuple[float, str]: - """Return the job runtime (wall clock time) along with its units. - - EDA tools indicate how long the job ran in terms of CPU time in the log - file. This method invokes the tool specific method which parses the log - text and returns the runtime as a floating point value followed by its - units as a tuple. - - Args: - log_text: is the job's log file contents as a list of lines. - tool: is the EDA tool used to run the job. - - Returns: - a tuple of (runtime, units). - - Raises: - NotImplementedError: exception if the EDA tool is not supported. - - """ - if tool == "xcelium": - return xcelium_job_runtime(log_text) - if tool == "vcs": - return vcs_job_runtime(log_text) - msg = f"{tool} is unsupported for job runtime extraction." - raise NotImplementedError(msg) - - -def vcs_job_runtime(log_text: list) -> tuple[float, str]: - """Return the VCS job runtime (wall clock time) along with its units. - - Search pattern example: - CPU time: 22.170 seconds to compile + .518 seconds to elab + 1.901 \ - seconds to link - CPU Time: 0.610 seconds; Data structure size: 1.6Mb - - Args: - log_text: is the job's log file contents as a list of lines. - - Returns: - the runtime, units as a tuple. - - Raises: - RuntimeError exception if the search pattern is not found. - - """ - pattern = r"^CPU [tT]ime:\s*(\d+\.?\d*?)\s*(seconds|minutes|hours).*$" - for line in reversed(log_text): - m = re.search(pattern, line) - if m: - return float(m.group(1)), m.group(2)[0] - msg = "Job runtime not found in the log." - raise RuntimeError(msg) - - -def xcelium_job_runtime(log_text: list) -> tuple[float, str]: - """Return the Xcelium job runtime (wall clock time) along with its units. - - Search pattern example: - TOOL: xrun(64) 21.09-s006: Exiting on Aug 01, 2022 at 00:21:18 PDT \ - (total: 00:00:05) - - Args: - log_text: is the job's log file contents as a list of lines. - - Returns: - the runtime, units as a tuple. - - Raises: - RuntimeError: exception if the search pattern is not found. - - """ - pattern = r"^TOOL:\s*xrun.*: Exiting on .*\(total:\s*(\d+):(\d+):(\d+)\)\s*$" - for line in reversed(log_text): - if m := re.search(pattern, line): - t = int(m.group(1)) * 3600 + int(m.group(2)) * 60 + int(m.group(3)) - return t, "s" - msg = "Job runtime not found in the log." - raise RuntimeError(msg) - - -def get_simulated_time(log_text: list, tool: str) -> tuple[float, str]: - """Return the simulated time along with its units. - - EDA tools indicate how long the design was simulated for in the log file. - This method invokes the tool specific method which parses the log text and - returns the simulated time as a floating point value followed by its - units (typically, pico|nano|micro|milliseconds) as a tuple. - - Args: - log_text: is the job's log file contents as a list of lines. - tool: is the EDA tool used to run the job. - - Returns: - the simulated, units as a tuple. - - Raises: - NotImplementedError: exception if the EDA tool is not supported. - - """ - if tool == "xcelium": - return xcelium_simulated_time(log_text) - if tool == "vcs": - return vcs_simulated_time(log_text) - msg = f"{tool} is unsupported for simulated time extraction." - raise NotImplementedError(msg) - - -def xcelium_simulated_time(log_text: list) -> tuple[float, str]: - """Return the Xcelium simulated time along with its units. - - Search pattern example: - Simulation complete via $finish(2) at time 11724965 PS + 13 - - Args: - log_text: is the job's log file contents as a list of lines. - - Returns: - the simulated time, units as a tuple. - - Raises: - RuntimeError: exception if the search pattern is not found. - - """ - pattern = r"^Simulation complete .* at time (\d+\.?\d*?)\s*(.?[sS]).*$" - for line in reversed(log_text): - m = re.search(pattern, line) - if m: - return float(m.group(1)), m.group(2).lower() - msg = "Simulated time not found in the log." - raise RuntimeError(msg) - - -def vcs_simulated_time(log_text: list) -> tuple[float, str]: - """Return the VCS simulated time along with its units. - - Search pattern example: - V C S S i m u l a t i o n R e p o r t - Time: 12241752 ps - - Args: - log_text: is the job's log file contents as a list of lines. - - Returns: - the simulated time, units as a tuple. - - Raises: - RuntimeError: exception if the search pattern is not found. - - """ - pattern = r"^Time:\s*(\d+\.?\d*?)\s*(.?[sS])\s*$" - next_line = "" - - for line in reversed(log_text): - if "V C S S i m u l a t i o n R e p o r t" not in line: - continue - - if m := re.search(pattern, next_line): - return float(m.group(1)), m.group(2).lower() - - next_line = line - - msg = "Simulated time not found in the log." - raise RuntimeError(msg) diff --git a/src/dvsim/tool/sim.py b/src/dvsim/tool/sim.py new file mode 100644 index 00000000..49249e55 --- /dev/null +++ b/src/dvsim/tool/sim.py @@ -0,0 +1,69 @@ +# Copyright lowRISC contributors (OpenTitan project). +# Licensed under the Apache License, Version 2.0, see LICENSE for details. +# SPDX-License-Identifier: Apache-2.0 + +"""EDA simulation tool interface.""" + +from collections.abc import Sequence +from pathlib import Path +from typing import Protocol, runtime_checkable + +__all__ = ("SimTool",) + + +@runtime_checkable +class SimTool(Protocol): + """Simulation tool interface required by the Sim workflow.""" + + @staticmethod + def get_cov_summary_table(cov_report_path: Path) -> tuple[Sequence[Sequence[str]], str]: + """Get a coverage summary. + + Args: + cov_report_path: path to the raw coverage report + + Returns: + tuple of, List of metrics and values, and final coverage total + + """ + ... + + @staticmethod + def get_job_runtime(log_text: Sequence[str]) -> tuple[float, str]: + """Return the job runtime (wall clock time) along with its units. + + EDA tools indicate how long the job ran in terms of CPU time in the log + file. This method invokes the tool specific method which parses the log + text and returns the runtime as a floating point value followed by its + units as a tuple. + + Args: + log_text: is the job's log file contents as a list of lines. + tool: is the EDA tool used to run the job. + + Returns: + a tuple of (runtime, units). + + """ + ... + + @staticmethod + def get_simulated_time(log_text: Sequence[str]) -> tuple[float, str]: + """Return the simulated time along with its units. + + EDA tools indicate how long the design was simulated for in the log file. + This method invokes the tool specific method which parses the log text and + returns the simulated time as a floating point value followed by its + units (typically, pico|nano|micro|milliseconds) as a tuple. + + Args: + log_text: is the job's log file contents as a list of lines. + + Returns: + the simulated, units as a tuple. + + Raises: + RuntimeError: exception if the search pattern is not found. + + """ + ... diff --git a/src/dvsim/tool/utils.py b/src/dvsim/tool/utils.py new file mode 100644 index 00000000..0997f853 --- /dev/null +++ b/src/dvsim/tool/utils.py @@ -0,0 +1,31 @@ +# Copyright lowRISC contributors (OpenTitan project). +# Licensed under the Apache License, Version 2.0, see LICENSE for details. +# SPDX-License-Identifier: Apache-2.0 + +"""EDA Tool base.""" + +from dvsim.logging import log +from dvsim.tool.sim import SimTool +from dvsim.tool.vcs import VCS +from dvsim.tool.xcelium import Xcelium + +__all__ = ("get_sim_tool_plugin",) + +_SUPPORTED_SIM_TOOLS = { + "vcs": VCS, + "xcelium": Xcelium, +} + + +def get_sim_tool_plugin(tool: str) -> SimTool: + """Get a simulation tool plugin.""" + if tool not in _SUPPORTED_SIM_TOOLS: + log.error( + "Unsupported tool '%s', please use one of [%s]", + tool, + ",".join(_SUPPORTED_SIM_TOOLS.keys()), + ) + msg = f"{tool} not supported" + raise NotImplementedError(msg) + + return _SUPPORTED_SIM_TOOLS[tool] diff --git a/src/dvsim/tool/vcs.py b/src/dvsim/tool/vcs.py new file mode 100644 index 00000000..6f84ede6 --- /dev/null +++ b/src/dvsim/tool/vcs.py @@ -0,0 +1,105 @@ +# Copyright lowRISC contributors (OpenTitan project). +# Licensed under the Apache License, Version 2.0, see LICENSE for details. +# SPDX-License-Identifier: Apache-2.0 + +"""EDA tool plugin providing VCS support to DVSim.""" + +import re +from collections.abc import Sequence +from pathlib import Path + +__all__ = ("VCS",) + + +class VCS: + """Implement VCS tool support.""" + + @staticmethod + def get_cov_summary_table(cov_report_path: Path) -> tuple[Sequence[Sequence[str]], str]: + """Get a coverage summary. + + Args: + cov_report_path: path to the raw coverage report + + Returns: + tuple of, List of metrics and values, and final coverage total + + """ + with Path(cov_report_path).open() as buf: + for line in buf: + match = re.match("total coverage summary", line, re.IGNORECASE) + if match: + # Metrics on the next line. + line = buf.readline().strip() + metrics = line.split() + # Values on the next. + line = buf.readline().strip() + # Pretty up the values - add % sign for ease of post + # processing. + values = [] + for val in line.split(): + val += " %" + values.append(val) + # first row is coverage total + cov_total = values[0] + return [metrics, values], cov_total + + # If we reached here, then we were unable to extract the coverage. + msg = f"Coverage data not found in {cov_report_path}!" + raise SyntaxError(msg) + + @staticmethod + def get_job_runtime(log_text: Sequence[str]) -> tuple[float, str]: + """Return the job runtime (wall clock time) along with its units. + + EDA tools indicate how long the job ran in terms of CPU time in the log + file. This method invokes the tool specific method which parses the log + text and returns the runtime as a floating point value followed by its + units as a tuple. + + Args: + log_text: is the job's log file contents as a list of lines. + tool: is the EDA tool used to run the job. + + Returns: + a tuple of (runtime, units). + + """ + pattern = r"^CPU [tT]ime:\s*(\d+\.?\d*?)\s*(seconds|minutes|hours).*$" + for line in reversed(log_text): + m = re.search(pattern, line) + if m: + return float(m.group(1)), m.group(2)[0] + msg = "Job runtime not found in the log." + raise RuntimeError(msg) + + @staticmethod + def get_simulated_time(log_text: Sequence[str]) -> tuple[float, str]: + """Return the simulated time along with its units. + + EDA tools indicate how long the design was simulated for in the log file. + This method invokes the tool specific method which parses the log text and + returns the simulated time as a floating point value followed by its + units (typically, pico|nano|micro|milliseconds) as a tuple. + + Args: + log_text: is the job's log file contents as a list of lines. + + Returns: + the simulated, units as a tuple. + + Raises: + RuntimeError: exception if the search pattern is not found. + + """ + pattern = re.compile(r"^Time:\s*(\d+\.?\d*?)\s*(.?[sS])\s*$") + + for line in reversed(log_text): + if "V C S S i m u l a t i o n R e p o r t" in line: + raise RuntimeError("Header found before sim time value") + + if m := pattern.search(line): + return float(m.group(1)), m.group(2).lower() + + msg = "Simulated time not found in the log." + raise RuntimeError(msg) diff --git a/src/dvsim/tool/xcelium.py b/src/dvsim/tool/xcelium.py new file mode 100644 index 00000000..2df28b6d --- /dev/null +++ b/src/dvsim/tool/xcelium.py @@ -0,0 +1,130 @@ +# Copyright lowRISC contributors (OpenTitan project). +# Licensed under the Apache License, Version 2.0, see LICENSE for details. +# SPDX-License-Identifier: Apache-2.0 + +"""EDA tool plugin providing Xcelium support to DVSim.""" + +import re +from collections import OrderedDict +from collections.abc import Sequence +from pathlib import Path + +__all__ = ("Xcelium",) + + +class Xcelium: + """Implement Xcelium tool support.""" + + @staticmethod + def get_cov_summary_table(cov_report_path: Path) -> tuple[Sequence[Sequence[str]], str]: + """Get a coverage summary. + + Args: + cov_report_path: path to the raw coverage report + + Returns: + tuple of, List of metrics and values, and final coverage total + + """ + with Path(cov_report_path).open() as buf: + for line in buf: + if "name" in line: + # Strip the line and remove the unwanted "* Covered" string. + metrics = line.strip().replace("* Covered", "").split() + # Change first item to 'Score'. + metrics[0] = "Score" + + # Gather the list of metrics. + items = OrderedDict() + for metric in metrics: + items[metric] = {} + items[metric]["covered"] = 0 + items[metric]["total"] = 0 + + # Next line is a separator. + line = buf.readline() + + # Subsequent lines are coverage items to be aggregated. + for line in buf: + line = re.sub(r"%\s+\(", "%(", line) + values = line.strip().split() + for i, value in enumerate(values): + value = value.strip() + m = re.search(r"\((\d+)/(\d+).*\)", value) + if m: + items[metrics[i]]["covered"] += int(m.group(1)) + items[metrics[i]]["total"] += int(m.group(2)) + items["Score"]["covered"] += int(m.group(1)) + items["Score"]["total"] += int(m.group(2)) + + # Capture the percentages and the aggregate. + values = [] + cov_total = None + for metric in items: + if items[metric]["total"] == 0: + values.append("-- %") + else: + value = items[metric]["covered"] / items[metric]["total"] * 100 + value = f"{round(value, 2):.2f} %" + values.append(value) + if metric == "Score": + cov_total = value + + return [items.keys(), values], cov_total + + # If we reached here, then we were unable to extract the coverage. + msg = f"Coverage data not found in {buf.name}!" + raise SyntaxError(msg) + + @staticmethod + def get_job_runtime(log_text: Sequence[str]) -> tuple[float, str]: + """Return the job runtime (wall clock time) along with its units. + + EDA tools indicate how long the job ran in terms of CPU time in the log + file. This method invokes the tool specific method which parses the log + text and returns the runtime as a floating point value followed by its + units as a tuple. + + Args: + log_text: is the job's log file contents as a list of lines. + tool: is the EDA tool used to run the job. + + Returns: + a tuple of (runtime, units). + + """ + pattern = r"^TOOL:\s*xrun.*: Exiting on .*\(total:\s*(\d+):(\d+):(\d+)\)\s*$" + for line in reversed(log_text): + if m := re.search(pattern, line): + t = int(m.group(1)) * 3600 + int(m.group(2)) * 60 + int(m.group(3)) + return t, "s" + + msg = "Job runtime not found in the log." + raise RuntimeError(msg) + + @staticmethod + def get_simulated_time(log_text: Sequence[str]) -> tuple[float, str]: + """Return the simulated time along with its units. + + EDA tools indicate how long the design was simulated for in the log file. + This method invokes the tool specific method which parses the log text and + returns the simulated time as a floating point value followed by its + units (typically, pico|nano|micro|milliseconds) as a tuple. + + Args: + log_text: is the job's log file contents as a list of lines. + + Returns: + the simulated, units as a tuple. + + Raises: + RuntimeError: exception if the search pattern is not found. + + """ + pattern = r"^Simulation complete .* at time (\d+\.?\d*?)\s*(.?[sS]).*$" + for line in reversed(log_text): + if m := re.search(pattern, line): + return float(m.group(1)), m.group(2).lower() + + msg = "Simulated time not found in the log." + raise RuntimeError(msg) diff --git a/tests/tool/__init__.py b/tests/tool/__init__.py new file mode 100644 index 00000000..658eb905 --- /dev/null +++ b/tests/tool/__init__.py @@ -0,0 +1,5 @@ +# Copyright lowRISC contributors (OpenTitan project). +# Licensed under the Apache License, Version 2.0, see LICENSE for details. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for EDA tool plugins and the plug-in system.""" diff --git a/tests/tool/test_utils.py b/tests/tool/test_utils.py new file mode 100644 index 00000000..8389eb55 --- /dev/null +++ b/tests/tool/test_utils.py @@ -0,0 +1,34 @@ +# Copyright lowRISC contributors (OpenTitan project). +# Licensed under the Apache License, Version 2.0, see LICENSE for details. +# SPDX-License-Identifier: Apache-2.0 + +"""Test the EDA tool utilities.""" + +import pytest +from hamcrest import assert_that, equal_to, instance_of + +from dvsim.tool.sim import SimTool +from dvsim.tool.utils import _SUPPORTED_SIM_TOOLS, get_sim_tool_plugin + +__all__ = ("TestEDAToolPlugins",) + + +class TestEDAToolPlugins: + """Test the EDA tool plug-ins.""" + + @staticmethod + @pytest.mark.parametrize("tool", _SUPPORTED_SIM_TOOLS.keys()) + def test_get_sim_tool_plugin(tool: str) -> None: + """Test that sim plugins can be retrieved correctly.""" + assert_that( + get_sim_tool_plugin(tool), + equal_to(_SUPPORTED_SIM_TOOLS[tool]), + ) + + @staticmethod + @pytest.mark.parametrize("tool", _SUPPORTED_SIM_TOOLS.keys()) + def test_plugins_implement_simtool_protocol(tool: str) -> None: + """Test that all sim plugins implement the SimTool interface.""" + plugin = get_sim_tool_plugin(tool) + + assert_that(plugin, instance_of(SimTool)) diff --git a/tests/tool/test_vcs.py b/tests/tool/test_vcs.py new file mode 100644 index 00000000..bce1ef56 --- /dev/null +++ b/tests/tool/test_vcs.py @@ -0,0 +1,55 @@ +# Copyright lowRISC contributors (OpenTitan project). +# Licensed under the Apache License, Version 2.0, see LICENSE for details. +# SPDX-License-Identifier: Apache-2.0 + +"""Test the VCS tool plugin.""" + +from collections.abc import Sequence + +import pytest +from hamcrest import assert_that, equal_to + +from dvsim.tool.utils import get_sim_tool_plugin + +__all__ = ("TestVCSToolPlugin",) + + +def fake_log(sim_time: float = 1, sim_time_units: str = "s") -> Sequence[str]: + """Fabricate a log.""" + return [ + "Other", + "log", + "content", + "", + " V C S S i m u l a t i o n R e p o r t ", + f"Time: {sim_time} {sim_time_units}", + ] + + +class TestVCSToolPlugin: + """Test the VCS tool plug-in.""" + + @staticmethod + @pytest.mark.parametrize( + ("time", "units"), + [ + (1.2, "s"), + (2.12, "ps"), + (3.73, "S"), + (4.235, "pS"), + (5.5134, "PS"), + ], + ) + def test_get_simulated_time(time: int, units: str) -> None: + """Test that sim plugins can be retrieved correctly.""" + plugin = get_sim_tool_plugin("vcs") + + assert_that( + plugin.get_simulated_time( + log_text=fake_log( + sim_time=time, + sim_time_units=units, + ) + ), + equal_to((time, units.lower())), + )