[launcher] Fake launcher to produce random results

src/dvsim/cli.py

@@ -34,6 +34,7 @@
 from dvsim.job.deploy import RunTest
 from dvsim.launcher.base import Launcher
 from dvsim.launcher.factory import set_launcher_type
+from dvsim.launcher.fake import FakeLauncher
 from dvsim.launcher.local import LocalLauncher
 from dvsim.launcher.lsf import LsfLauncher
 from dvsim.launcher.nc import NcLauncher
@@ -785,6 +786,12 @@ def parse_args():
         help=("Print dvsim tool messages but don't actually run any command"),
     )
 
+    dvg.add_argument(
+        "--fake",
+        action="store_true",
+        help=("Use a fake launcher that generates random results"),
+    )
+
     args = parser.parse_args()
 
     if args.version:
@@ -843,7 +850,7 @@ def main() -> None:
         args.cfg = os.path.join(proj_root, cfg_path)
 
     # Add timestamp to args that all downstream objects can use.
-    curr_ts = datetime.datetime.utcnow()
+    curr_ts = datetime.datetime.now(datetime.UTC)
     args.timestamp_long = curr_ts.strftime(TS_FORMAT_LONG)
     args.timestamp = curr_ts.strftime(TS_FORMAT)
 
@@ -863,11 +870,11 @@ def main() -> None:
     LsfLauncher.max_parallel = args.max_parallel
     NcLauncher.max_parallel = args.max_parallel
     Launcher.max_odirs = args.max_odirs
-    set_launcher_type(args.local)
+    FakeLauncher.max_parallel = args.max_parallel
+    set_launcher_type(is_local=args.local, fake=args.fake)
 
     # Build infrastructure from hjson file and create the list of items to
     # be deployed.
-    global cfg
     cfg = make_cfg(args.cfg, args, proj_root)
 
     # List items available for run if --list switch is passed, and exit.

src/dvsim/flow/sim.py

@@ -807,7 +807,7 @@ def create_bucket_report(buckets):
             return fail_msgs
 
         deployed_items = self.deploy
-        results = SimResults(deployed_items, results)
+        sim_results = SimResults(deployed_items, results)
 
         # Generate results table for runs.
         results_str = "## " + self.results_title + "\n"
@@ -857,13 +857,13 @@ def create_bucket_report(buckets):
         if self.build_seed and not self.run_only:
             results_str += f"### Build randomization enabled with --build-seed {self.build_seed}\n"
 
-        if not results.table:
+        if not sim_results.table:
             results_str += "No results to display.\n"
 
         else:
             # Map regr results to the testplan entries.
             if not self.testplan.test_results_mapped:
-                self.testplan.map_test_results(test_results=results.table)
+                self.testplan.map_test_results(test_results=sim_results.table)
 
             results_str += self.testplan.get_test_results_table(
                 map_full_testplan=self.map_full_testplan,
@@ -889,9 +889,9 @@ def create_bucket_report(buckets):
                 else:
                     self.results_summary["Coverage"] = "--"
 
-        if results.buckets:
+        if sim_results.buckets:
             self.errors_seen = True
-            results_str += "\n".join(create_bucket_report(results.buckets))
+            results_str += "\n".join(create_bucket_report(sim_results.buckets))
 
         self.results_md = results_str
         return results_str

src/dvsim/launcher/factory.py

@@ -5,6 +5,7 @@
 import os
 
 from dvsim.launcher.base import Launcher
+from dvsim.launcher.fake import FakeLauncher
 from dvsim.launcher.local import LocalLauncher
 from dvsim.launcher.lsf import LsfLauncher
 from dvsim.launcher.nc import NcLauncher
@@ -23,7 +24,7 @@
 _LAUNCHER_CLS: type[Launcher] | None = None
 
 
-def set_launcher_type(is_local: bool = False) -> None:
+def set_launcher_type(is_local: bool = False, fake: bool = False) -> None:
     """Set the launcher type that will be used to launch the jobs.
 
     The env variable `DVSIM_LAUNCHER` is used to identify what launcher system
@@ -35,6 +36,8 @@ def set_launcher_type(is_local: bool = False) -> None:
     launcher = os.environ.get("DVSIM_LAUNCHER", "local")
     if is_local:
         launcher = "local"
+    if fake:
+        launcher = "fake"
     Launcher.variant = launcher
 
     global _LAUNCHER_CLS
@@ -53,6 +56,9 @@ def set_launcher_type(is_local: bool = False) -> None:
     elif launcher == "slurm":
         _LAUNCHER_CLS = SlurmLauncher
 
+    elif launcher == "fake":
+        _LAUNCHER_CLS = FakeLauncher
+
     # These custom launchers are site specific. They may not be committed to
     # the open source repo.
     elif launcher == "edacloud" and EDACLOUD_LAUNCHER_EXISTS:

src/dvsim/launcher/fake.py

@@ -0,0 +1,72 @@
+# Copyright lowRISC contributors (OpenTitan project).
+# Licensed under the Apache License, Version 2.0, see LICENSE for details.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Fake Launcher that returns random results."""
+
+from random import choice, random
+from typing import TYPE_CHECKING
+
+from dvsim.launcher.base import ErrorMessage, Launcher
+
+if TYPE_CHECKING:
+    from dvsim.job.deploy import CovReport, Deploy, RunTest
+
+
+def _run_test_handler(deploy: "RunTest") -> str:
+    """Handle a RunTest deploy job."""
+    return choice(("P", "F"))
+
+
+def _cov_report_handler(deploy: "CovReport") -> str:
+    """Handle a CompileSim deploy job."""
+    keys = [
+        "score",
+        "line",
+        "cond",
+        "toggle",
+        "fsm",
+        "branch",
+        "assert",
+        "group",
+    ]
+
+    deploy.cov_results_dict = {k: f"{random() * 100:.2f} %" for k in keys}
+
+    return "P"
+
+
+_DEPLOY_HANDLER = {
+    "RunTest": _run_test_handler,
+    "CovReport": _cov_report_handler,
+}
+
+
+class FakeLauncher(Launcher):
+    """Launch jobs and return fake results."""
+
+    # Poll job's completion status every this many seconds
+    poll_freq = 0
+
+    def __init__(self, deploy: "Deploy") -> None:
+        """Initialize common class members."""
+        super().__init__(deploy)
+
+    def _do_launch(self) -> None:
+        """Do the launch."""
+
+    def poll(self) -> str | None:
+        """Check status of the running process."""
+        deploy_cls = self.deploy.__class__.__name__
+        if deploy_cls in _DEPLOY_HANDLER:
+            return _DEPLOY_HANDLER[deploy_cls](deploy=self.deploy)
+
+        # Default result is Pass
+        return "P"
+
+    def kill(self) -> None:
+        """Kill the running process."""
+        self._post_finish(
+            "K",
+            ErrorMessage(line_number=None, message="Job killed!", context=[]),
+        )