refactor: make Job Status an Enum

Currently the job status is passed around DVSim as a string, where the
expectation is that it is always a single upper case character in ('Q',
'D', 'P', 'F', 'K') - some limited parts of the code also handle an 'E'
error case.

Turn these into a small Enum type so that we have much stricter typing
on these values and improved ergonomics when referring to different job
statuses in code.

This status Enum lives in its own module instead of with the other job
data to avoid circular imports (as the job status' registered callbacks
reference the launcher definition).

Signed-off-by: Alex Jones <alex.jones@lowrisc.org>

Author: AlexJones0

src/dvsim/flow/formal.py

@@ -10,6 +10,7 @@
 
 from dvsim.flow.one_shot import OneShotCfg
 from dvsim.job.data import CompletedJobStatus
+from dvsim.job.status import JobStatus
 from dvsim.logging import log
 from dvsim.utils import subst_wildcards
 
@@ -230,7 +231,7 @@ def _gen_results(self, results: Sequence[CompletedJobStatus]) -> None:
         assert len(self.deploy) == 1
         mode = self.deploy[0]
 
-        if complete_job.status == "P":
+        if complete_job.status == JobStatus.PASSED:
             result_data = Path(
                 subst_wildcards(self.build_dir, {"build_mode": mode.name}),
                 "results.hjson",
@@ -262,7 +263,7 @@ def _gen_results(self, results: Sequence[CompletedJobStatus]) -> None:
         else:
             summary += ["N/A", "N/A", "N/A"]
 
-        if complete_job.status != "P":
+        if complete_job.status != JobStatus.PASSED:
             results_str += "\n## List of Failures\n" + "".join(complete_job.fail_msg.message)
 
         messages = self.result.get("messages")

src/dvsim/flow/one_shot.py

@@ -11,6 +11,7 @@
 
 from dvsim.flow.base import FlowCfg
 from dvsim.job.data import CompletedJobStatus
+from dvsim.job.status import JobStatus
 from dvsim.job.deploy import CompileOneShot
 from dvsim.logging import log
 from dvsim.modes import BuildMode, Mode
@@ -92,10 +93,10 @@ def _expand(self) -> None:
 
             # Set directories with links for ease of debug / triage.
             self.links = {
-                "D": self.scratch_path + "/" + "dispatched",
-                "P": self.scratch_path + "/" + "passed",
-                "F": self.scratch_path + "/" + "failed",
-                "K": self.scratch_path + "/" + "killed",
+                JobStatus.DISPATCHED: self.scratch_path + "/" + "dispatched",
+                JobStatus.PASSED: self.scratch_path + "/" + "passed",
+                JobStatus.FAILED: self.scratch_path + "/" + "failed",
+                JobStatus.KILLED: self.scratch_path + "/" + "killed",
             }
 
             # Use the default build mode for tests that do not specify it

src/dvsim/job/data.py

@@ -14,6 +14,7 @@
 
 from pydantic import BaseModel, ConfigDict
 
+from dvsim.job.status import JobStatus
 from dvsim.launcher.base import ErrorMessage, Launcher
 from dvsim.report.data import IPMeta, ToolMeta
 
@@ -97,13 +98,13 @@ class JobSpec(BaseModel):
     """Output directory for the job results files."""
     log_path: Path
     """Path for the job log file."""
-    links: Mapping[str, Path]
+    links: Mapping[JobStatus, Path]
     """Path for links directories."""
 
     # TODO: remove the need for these callables here
     pre_launch: Callable[[Launcher], None]
     """Callback function for pre-launch actions."""
-    post_finish: Callable[[str], None]
+    post_finish: Callable[[JobStatus], None]
     """Callback function for tidy up actions once the job is finished."""
 
     pass_patterns: Sequence[str]
@@ -153,7 +154,7 @@ class CompletedJobStatus(BaseModel):
     simulated_time: float
     """Simulation time."""
 
-    status: str
-    """Job status string [P,F,K,...]"""
+    status: JobStatus
+    """Status of the job."""
     fail_msg: ErrorMessage
     """Error message."""

src/dvsim/job/deploy.py

@@ -14,6 +14,7 @@
 from tabulate import tabulate
 
 from dvsim.job.data import JobSpec, WorkspaceConfig
+from dvsim.job.status import JobStatus
 from dvsim.job.time import JobTime
 from dvsim.launcher.base import Launcher
 from dvsim.logging import log
@@ -347,10 +348,10 @@ def callback(launcher: Launcher) -> None:
 
         return callback
 
-    def post_finish(self) -> Callable[[str], None]:
+    def post_finish(self) -> Callable[[JobStatus], None]:
         """Get post finish callback."""
 
-        def callback(status: str) -> None:
+        def callback(status: JobStatus) -> None:
             """Perform additional post-finish activities (callback).
 
             This is invoked by launcher::_post_finish().
@@ -641,12 +642,12 @@ def callback(launcher: Launcher) -> None:
 
         return callback
 
-    def post_finish(self) -> Callable[[str], None]:
+    def post_finish(self) -> Callable[[JobStatus], None]:
         """Get post finish callback."""
 
-        def callback(status: str) -> None:
+        def callback(status: JobStatus) -> None:
             """Perform tidy up tasks."""
-            if status != "P":
+            if status != JobStatus.PASSED:
                 # Delete the coverage data if available.
                 rm_path(self.cov_db_test_dir)
 
@@ -812,16 +813,16 @@ def _set_attrs(self) -> None:
         self.cov_results = ""
         self.cov_results_dict = {}
 
-    def post_finish(self) -> Callable[[str], None]:
+    def post_finish(self) -> Callable[[JobStatus], None]:
         """Get post finish callback."""
 
-        def callback(status: str) -> None:
+        def callback(status: JobStatus) -> None:
             """Extract the coverage results summary for the dashboard.
 
             If the extraction fails, an appropriate exception is raised, which must
             be caught by the caller to mark the job as a failure.
             """
-            if self.dry_run or status != "P":
+            if self.dry_run or status != JobStatus.PASSED:
                 return
 
             plugin = get_sim_tool_plugin(tool=self.sim_cfg.tool)

src/dvsim/job/status.py

@@ -0,0 +1,30 @@
+# Copyright lowRISC contributors (OpenTitan project).
+# Licensed under the Apache License, Version 2.0, see LICENSE for details.
+# SPDX-License-Identifier: Apache-2.0
+
+"""An enum definition for the various job statuses."""
+
+from enum import Enum
+
+__all__ = ("JobStatus",)
+
+
+class JobStatus(Enum):
+    """Status of a Job."""
+
+    ERROR = 0  # Internal error / failure to build
+    QUEUED = 1
+    DISPATCHED = 2
+    PASSED = 3
+    FAILED = 4  # A 'failed' result from a valid run.
+    KILLED = 5
+
+    @property
+    def shorthand(self) -> str:
+        """Shorthand for the job status, e.g. 'D' for 'Dispatched'."""
+        return self.name[0]
+
+    @property
+    def completed(self) -> bool:
+        """Whether this status corresponds to some (non-error) completed job."""
+        return self in (JobStatus.PASSED, JobStatus.FAILED, JobStatus.KILLED)

src/dvsim/launcher/base.py

@@ -15,6 +15,7 @@
 
 from pydantic import BaseModel, ConfigDict
 
+from dvsim.job.status import JobStatus
 from dvsim.job.time import JobTime
 from dvsim.logging import log
 from dvsim.tool.utils import get_sim_tool_plugin
@@ -210,7 +211,7 @@ def _make_odir(self) -> None:
 
         Path(self.job_spec.odir).mkdir(exist_ok=True, parents=True)
 
-    def _link_odir(self, status: str) -> None:
+    def _link_odir(self, status: JobStatus) -> None:
         """Soft-links the job's directory based on job's status.
 
         The dispatched, passed and failed directories in the scratch area
@@ -220,8 +221,8 @@ def _link_odir(self, status: str) -> None:
         mk_symlink(path=self.job_spec.odir, link=dest)
 
         # Delete the symlink from dispatched directory if it exists.
-        if status != "D":
-            old = Path(self.job_spec.links["D"], self.job_spec.qual_name)
+        if status != JobStatus.DISPATCHED:
+            old = Path(self.job_spec.links[JobStatus.DISPATCHED], self.job_spec.qual_name)
             rm_path(old)
 
     def _dump_env_vars(self, exports: Mapping[str, str]) -> None:
@@ -258,7 +259,7 @@ def launch(self) -> None:
         self._do_launch()
 
     @abstractmethod
-    def poll(self) -> str | None:
+    def poll(self) -> JobStatus | None:
         """Poll the launched job for completion.
 
         Invokes _check_status() and _post_finish() when the job completes.
@@ -272,14 +273,14 @@ def poll(self) -> str | None:
     def kill(self) -> None:
         """Terminate the job."""
 
-    def _check_status(self) -> tuple[str, ErrorMessage | None]:
-        """Determine the outcome of the job (P/F if it ran to completion).
+    def _check_status(self) -> tuple[JobStatus, ErrorMessage | None]:
+        """Determine the outcome of the job (Passed/Failed if it ran to completion).
 
         Returns:
             (status, err_msg) extracted from the log, where the status is
-            "P" if the it passed, "F" otherwise. This is invoked by poll() just
-            after the job finishes. err_msg is an instance of the named tuple
-            ErrorMessage.
+            Passed if the job passed, Failed otherwise. This is invoked by
+            poll() just after the job finishes. err_msg is an instance of the
+            named tuple ErrorMessage.
 
         """
 
@@ -307,7 +308,7 @@ def _find_patterns(patterns: Sequence[str], line: str) -> Sequence[str] | None:
             return None
 
         if self.job_spec.dry_run:
-            return "P", None
+            return JobStatus.PASSED, None
 
         # Only one fail pattern needs to be seen.
         chk_failed = bool(self.job_spec.fail_patterns)
@@ -324,7 +325,7 @@ def _find_patterns(patterns: Sequence[str], line: str) -> Sequence[str] | None:
             ) as f:
                 lines = f.readlines()
         except OSError as e:
-            return "F", ErrorMessage(
+            return JobStatus.FAILED, ErrorMessage(
                 line_number=None,
                 message=f"Error opening file {self.job_spec.log_path}:\n{e}",
                 context=[],
@@ -368,7 +369,7 @@ def _find_patterns(patterns: Sequence[str], line: str) -> Sequence[str] | None:
                     # If failed, then nothing else to do. Just return.
                     # Provide some extra lines for context.
                     end = cnt + 5
-                    return "F", ErrorMessage(
+                    return JobStatus.FAILED, ErrorMessage(
                         line_number=cnt + 1,
                         message=line.strip(),
                         context=lines[cnt:end],
@@ -384,32 +385,32 @@ def _find_patterns(patterns: Sequence[str], line: str) -> Sequence[str] | None:
         # exit code for whatever reason, then show the last 10 lines of the log
         # as the failure message, which might help with the debug.
         if self.exit_code != 0:
-            return "F", ErrorMessage(
+            return JobStatus.FAILED, ErrorMessage(
                 line_number=None,
                 message="Job returned non-zero exit code",
                 context=lines[-10:],
             )
         if chk_passed:
-            return "F", ErrorMessage(
+            return JobStatus.FAILED, ErrorMessage(
                 line_number=None,
                 message=f"Some pass patterns missing: {pass_patterns}",
                 context=lines[-10:],
             )
-        return "P", None
+        return JobStatus.PASSED, None
 
-    def _post_finish(self, status: str, err_msg: ErrorMessage) -> None:
+    def _post_finish(self, status: JobStatus, err_msg: ErrorMessage) -> None:
         """Do post-completion activities, such as preparing the results.
 
         Must be invoked by poll(), after the job outcome is determined.
 
         Args:
-            status: status of the job, either 'P', 'F' or 'K'.
+            status: status of the completed job (must be either Passed, Failed or Killed).
             err_msg: an instance of the named tuple ErrorMessage.
 
         """
-        assert status in ["P", "F", "K"]
+        assert status.completed
         self._link_odir(status)
-        log.debug("Item %s has completed execution: %s", self, status)
+        log.debug("Item %s has completed execution: %s", self, status.shorthand)
 
         try:
             # Run the target-specific cleanup tasks regardless of the job's
@@ -419,16 +420,16 @@ def _post_finish(self, status: str, err_msg: ErrorMessage) -> None:
         except Exception as e:
             # If the job had already failed, then don't do anything. If it's
             # cleanup task failed, then mark the job as failed.
-            if status == "P":
-                status = "F"
+            if status == JobStatus.PASSED:
+                status = JobStatus.FAILED
                 err_msg = ErrorMessage(
                     line_number=None,
                     message=f"{e}",
                     context=[f"{e}"],
                 )
 
         self.status = status
-        if self.status != "P":
+        if self.status != JobStatus.PASSED:
             assert err_msg
             assert isinstance(err_msg, ErrorMessage)
             self.fail_msg = err_msg

src/dvsim/launcher/fake.py

@@ -7,6 +7,7 @@
 from random import choice
 from typing import TYPE_CHECKING
 
+from dvsim.job.status import JobStatus
 from dvsim.launcher.base import ErrorMessage, Launcher
 
 if TYPE_CHECKING:
@@ -16,12 +17,12 @@
 __all__ = ("FakeLauncher",)
 
 
-def _run_test_handler(job_spec: "JobSpec") -> str:
+def _run_test_handler(job_spec: "JobSpec") -> JobStatus:
     """Handle a RunTest deploy job."""
-    return choice(("P", "F"))
+    return choice((JobStatus.PASSED, JobStatus.FAILED))
 
 
-def _cov_report_handler(job_spec: "JobSpec") -> str:
+def _cov_report_handler(job_spec: "JobSpec") -> JobStatus:
     """Handle a CompileSim deploy job."""
     # TODO: this hack doesn't work any more and needs implementing by writing
     # a file that can be parsed as if it's been generated by the tool.
@@ -38,7 +39,7 @@ def _cov_report_handler(job_spec: "JobSpec") -> str:
     # ]
     # job_spec.cov_results_dict = {k: f"{random() * 100:.2f} %" for k in keys}
 
-    return "P"
+    return JobStatus.PASSED
 
 
 _DEPLOY_HANDLER = {
@@ -56,19 +57,19 @@ class FakeLauncher(Launcher):
     def _do_launch(self) -> None:
         """Do the launch."""
 
-    def poll(self) -> str | None:
+    def poll(self) -> JobStatus | None:
         """Check status of the running process."""
         deploy_cls = self.job_spec.job_type
         if deploy_cls in _DEPLOY_HANDLER:
             return _DEPLOY_HANDLER[deploy_cls](job_spec=self.job_spec)
 
         # Default result is Pass
-        return "P"
+        return JobStatus.PASSED
 
     def kill(self) -> None:
         """Kill the running process."""
         self._post_finish(
-            "K",
+            JobStatus.KILLED,
             ErrorMessage(line_number=None, message="Job killed!", context=[]),
         )