1# Copyright 2025 The Hafnium Authors.
2#
3# Use of this source code is governed by a BSD-style
4# license that can be found in the LICENSE file or at
5# https://opensource.org/licenses/BSD-3-Clause.
6
7import collections
8import subprocess
9
10from common import (
11    append_file,
12    read_file,
13    HFTEST_LOG_FAILURE_PREFIX,
14    HFTEST_LOG_PREFIX,
15)
16
17# Tuple holding the arguments common to all driver constructors.
18# This is to avoid having to pass arguments from subclasses to superclasses.
19DriverArgs = collections.namedtuple("DriverArgs", [
20        "artifacts",
21        "hypervisor",
22        "spmc",
23        "initrd",
24        "vm_args",
25        "cpu",
26        "partitions",
27        "global_run_name",
28        "coverage_plugin",
29        "disable_visualisation"
30    ])
31
32# State shared between the common Driver class and its subclasses during
33# a single invocation of the target platform.
34class DriverRunState:
35    def __init__(self, log_path):
36        self.log_path = log_path
37        self.ret_code = 0
38
39    def set_ret_code(self, ret_code):
40        self.ret_code = ret_code
41
42class DriverRunException(Exception):
43    """Exception thrown if subprocess invoked by a driver returned non-zero
44    status code. Used to fast-exit from a driver command sequence."""
45    pass
46
47
48class Driver:
49    """Parent class of drivers for all testable platforms."""
50
51    def __init__(self, args):
52        self.args = args
53
54    def get_run_log(self, run_name):
55        """Return path to the main log of a given test run."""
56        return self.args.artifacts.get_file(run_name, ".log")
57
58    def start_run(self, run_name):
59        """Hook called by Driver subclasses before they invoke the target
60        platform."""
61        return DriverRunState(self.args.artifacts.create_file(run_name, ".log"))
62
63    def exec_logged(self, run_state, exec_args, cwd=None, env=None):
64        """Run a subprocess on behalf of a Driver subclass and append its
65        stdout and stderr to the main log."""
66        assert(run_state.ret_code == 0)
67        with open(run_state.log_path, "a") as f:
68            f.write("$ {}\r\n".format(" ".join(exec_args)))
69            f.flush()
70            ret_code = subprocess.call(exec_args, stdout=f, stderr=f, cwd=cwd, env=env)
71            if ret_code != 0:
72                run_state.set_ret_code(ret_code)
73                raise DriverRunException()
74
75    def finish_run(self, run_state):
76        """Hook called by Driver subclasses after they finished running the
77        target platform. `ret_code` argument is the return code of the main
78        command run by the driver. A corresponding log message is printed."""
79        # Decode return code and add a message to the log.
80        with open(run_state.log_path, "a") as f:
81            if run_state.ret_code == 124:
82                f.write("\r\n{}{} timed out\r\n".format(
83                    HFTEST_LOG_PREFIX, HFTEST_LOG_FAILURE_PREFIX))
84            elif run_state.ret_code != 0:
85                f.write("\r\n{}{} process return code {}\r\n".format(
86                    HFTEST_LOG_PREFIX, HFTEST_LOG_FAILURE_PREFIX,
87                    run_state.ret_code))
88
89        # Append log of this run to full test log.
90        log_content = read_file(run_state.log_path)
91        append_file(
92            self.args.artifacts.sponge_log_path,
93            log_content + "\r\n\r\n")
94        return log_content
95