arm/OEFVPTarget: Add support for model state transitions

To better support firmware testing alongside Linux runtime testing,
introduce model state support to OEFVPTarget. The following states are
supported using self.target.transition(state):
 * off
 * on
 * linux

Instead of assuming a specific state in OEFVPTarget.start,
responsibility is delegated to test cases to lazily put the model in
the required state. But to support OE-core test cases, OEFVPTarget.run
automatically puts the model in the "linux" state for running the
command. Firmware and Linux tests can subsequently run alongside each
other without introducing complex test dependencies.

The concept is inspired by Labgrid strategies [1], albeit simplified.

Tweak log file handling so that output is collected across (possibly)
multiple model processes.

[1] https://labgrid.readthedocs.io/en/latest/overview.html#strategies

Signed-off-by: Peter Hoyes <Peter.Hoyes@arm.com>
Signed-off-by: Jon Mason <jon.mason@arm.com>

meta-arm/lib/oeqa/controllers/fvp.py

@@ -1,3 +1,5 @@
+import contextlib
+import enum
 import pathlib
 import pexpect
 import os
@@ -5,6 +7,11 @@ import os
 from oeqa.core.target.ssh import OESSHTarget
 from fvp import runner
 
+class OEFVPTargetState(str, enum.Enum):
+    OFF = "off"
+    ON = "on"
+    LINUX = "linux"
+
 
 class OEFVPTarget(OESSHTarget):
     """
@@ -27,37 +34,50 @@ class OEFVPTarget(OESSHTarget):
 
         self.bootlog = bootlog
         self.terminals = {}
-        self.booted = False
+        self.stack = None
+        self.state = OEFVPTargetState.OFF
 
-    def start(self, **kwargs):
+    def transition(self, state, timeout=10*60):
-        self.fvp_log = self._create_logfile("fvp")
+        if state == self.state:
-        self.fvp = runner.FVPRunner(self.logger)
-        self.fvp.start(self.fvpconf, stdout=self.fvp_log)
-        self.logger.debug(f"Started FVP PID {self.fvp.pid()}")
-        self._setup_consoles()
-
-    def await_boot(self):
-        if self.booted:
             return
 
-        # FVPs boot slowly, so allow ten minutes
+        if state == OEFVPTargetState.OFF:
-        boot_timeout = 10*60
+            returncode = self.fvp.stop()
-        try:
+            self.logger.debug(f"Stopped FVP with return code {returncode}")
-            self.expect(OEFVPTarget.DEFAULT_CONSOLE, "login\\:", timeout=boot_timeout)
+            self.stack.close()
-            self.logger.debug("Found login prompt")
+        elif state == OEFVPTargetState.ON:
-            self.booted = True
+            self.transition(OEFVPTargetState.OFF, timeout)
-        except pexpect.TIMEOUT:
+            self.stack = contextlib.ExitStack()
-            self.logger.info("Timed out waiting for login prompt.")
+            self.fvp = runner.FVPRunner(self.logger)
-            self.logger.info("Boot log follows:")
+            self.fvp_log = self._create_logfile("fvp", "wb")
-            self.logger.info(b"\n".join(self.before(OEFVPTarget.DEFAULT_CONSOLE).splitlines()[-200:]).decode("utf-8", errors="replace"))
+            self.fvp.start(self.fvpconf, stdout=self.fvp_log)
-            raise RuntimeError("Failed to start FVP.")
+            self.logger.debug(f"Started FVP PID {self.fvp.pid()}")
+            self._setup_consoles()
+        elif state == OEFVPTargetState.LINUX:
+            self.transition(OEFVPTargetState.ON, timeout)
+            try:
+                self.expect(OEFVPTarget.DEFAULT_CONSOLE, "login\\:", timeout=timeout)
+                self.logger.debug("Found login prompt")
+                self.state = OEFVPTargetState.LINUX
+            except pexpect.TIMEOUT:
+                self.logger.info("Timed out waiting for login prompt.")
+                self.logger.info("Boot log follows:")
+                self.logger.info(b"\n".join(self.before(OEFVPTarget.DEFAULT_CONSOLE).splitlines()[-200:]).decode("utf-8", errors="replace"))
+                raise RuntimeError("Failed to start FVP.")
+
+        self.logger.info(f"Transitioned to {state}")
+        self.state = state
+
+    def start(self, **kwargs):
+        # No-op - put the FVP in the required state lazily
+        pass
 
     def stop(self, **kwargs):
-        returncode = self.fvp.stop()
+        self.transition(OEFVPTargetState.OFF)
-        self.logger.debug(f"Stopped FVP with return code {returncode}")
 
     def run(self, cmd, timeout=None):
-        self.await_boot()
+        # Running a command implies the LINUX state
+        self.transition(OEFVPTargetState.LINUX)
         return super().run(cmd, timeout)
 
     def _setup_consoles(self):
@@ -73,11 +93,12 @@ class OEFVPTarget(OESSHTarget):
 
                 # testimage.bbclass expects to see a log file at `bootlog`,
                 # so make a symlink to the 'default' log file
-                if name == 'default':
+                test_log_suffix = pathlib.Path(self.bootlog).suffix
-                    default_test_file = f"{name}_log{self.test_log_suffix}"
+                default_test_file = f"{name}_log{test_log_suffix}"
+                if name == 'default' and not os.path.exists(self.bootlog):
                     os.symlink(default_test_file, self.bootlog)
 
-    def _create_logfile(self, name):
+    def _create_logfile(self, name, mode='ab'):
         if not self.bootlog:
             return None
 
@@ -91,7 +112,7 @@ class OEFVPTarget(OESSHTarget):
         except:
             pass
         os.symlink(fvp_log_file, fvp_log_symlink)
-        return open(fvp_log_path, 'wb')
+        return self.stack.enter_context(open(fvp_log_path, mode))
 
     def _get_terminal(self, name):
         return self.terminals[name]