@@ -1,45 +1,45 @@
import asyncio
import pathlib
import pexpect
+import os
-import oeqa.core.target.ssh
+from oeqa.core.target.ssh import OESSHTarget
from fvp import conffile, runner
-class OEFVPTarget(oeqa.core.target.ssh.OESSHTarget):
+
+class OEFVPSSHTarget(OESSHTarget):
+ """
+ Base class for meta-arm FVP targets.
+ Contains common logic to start and stop an FVP.
+ """
def __init__(self, logger, target_ip, server_ip, timeout=300, user='root',
- port=None, server_port=0, dir_image=None, rootfs=None, bootlog=None,
- **kwargs):
+ port=None, dir_image=None, rootfs=None, **kwargs):
super().__init__(logger, target_ip, server_ip, timeout, user, port)
image_dir = pathlib.Path(dir_image)
# rootfs may have multiple extensions so we need to strip *all* suffixes
basename = pathlib.Path(rootfs)
basename = basename.name.replace("".join(basename.suffixes), "")
self.fvpconf = image_dir / (basename + ".fvpconf")
+ self.config = conffile.load(self.fvpconf)
if not self.fvpconf.exists():
raise FileNotFoundError(f"Cannot find {self.fvpconf}")
- # FVPs boot slowly, so allow ten minutes
- self.boot_timeout = 10 * 60
-
- self.logfile = bootlog and open(bootlog, "wb") or None
async def boot_fvp(self):
- config = conffile.load(self.fvpconf)
self.fvp = runner.FVPRunner(self.logger)
- await self.fvp.start(config)
+ await self.fvp.start(self.config)
self.logger.debug(f"Started FVP PID {self.fvp.pid()}")
- console = await self.fvp.create_pexpect(config["console"])
- try:
- console.expect("login\:", timeout=self.boot_timeout)
- self.logger.debug("Found login prompt")
- except pexpect.TIMEOUT:
- self.logger.info("Timed out waiting for login prompt.")
- self.logger.info("Boot log follows:")
- self.logger.info(b"\n".join(console.before.splitlines()[-200:]).decode("utf-8", errors="replace"))
- raise RuntimeError("Failed to start FVP.")
+ await self._after_start()
+
+ async def _after_start(self):
+ pass
+
+ async def _after_stop(self):
+ pass
async def stop_fvp(self):
returncode = await self.fvp.stop()
+ await self._after_stop()
self.logger.debug(f"Stopped FVP with return code {returncode}")
@@ -51,3 +51,102 @@ class OEFVPTarget(oeqa.core.target.ssh.OESSHTarget):
def stop(self, **kwargs):
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(self.stop_fvp()))
+
+
+class OEFVPTarget(OEFVPSSHTarget):
+ """
+ For compatibility with OE-core test cases, this target's start() method
+ waits for a Linux shell before returning to ensure that SSH commands work
+ with the default test dependencies.
+ """
+ def __init__(self, logger, target_ip, server_ip, bootlog=None, **kwargs):
+ super().__init__(logger, target_ip, server_ip, **kwargs)
+ self.logfile = bootlog and open(bootlog, "wb") or None
+
+ # FVPs boot slowly, so allow ten minutes
+ self.boot_timeout = 10 * 60
+
+ async def _after_start(self):
+ self.logger.debug(f"Awaiting console on terminal {self.config['consoles']['default']}")
+ console = await self.fvp.create_pexpect(self.config['consoles']['default'])
+ try:
+ console.expect("login\\:", timeout=self.boot_timeout)
+ self.logger.debug("Found login prompt")
+ except pexpect.TIMEOUT:
+ self.logger.info("Timed out waiting for login prompt.")
+ self.logger.info("Boot log follows:")
+ self.logger.info(b"\n".join(console.before.splitlines()[-200:]).decode("utf-8", errors="replace"))
+ raise RuntimeError("Failed to start FVP.")
+
+
+class OEFVPSerialTarget(OEFVPSSHTarget):
+ """
+ This target is intended for interaction with the target over one or more
+ telnet consoles using pexpect.
+
+ This still depends on OEFVPSSHTarget so SSH commands can still be run on
+ the target, but note that this class does not inherently guarantee that
+ the SSH server is running prior to running test cases. Test cases that use
+ SSH should first validate that SSH is available.
+ """
+ DEFAULT_CONSOLE = "default"
+
+ def __init__(self, logger, target_ip, server_ip, bootlog=None, **kwargs):
+ super().__init__(logger, target_ip, server_ip, **kwargs)
+ self.terminals = {}
+
+ self.test_log_path = pathlib.Path(bootlog).parent
+ self.test_log_suffix = pathlib.Path(bootlog).suffix
+ self.bootlog = bootlog
+
+ async def _add_terminal(self, name, fvp_name):
+ logfile = self._create_logfile(name)
+ self.logger.info(f'Creating terminal {name} on {fvp_name}')
+ self.terminals[name] = \
+ await self.fvp.create_pexpect(fvp_name, logfile=logfile)
+
+ def _create_logfile(self, name):
+ fvp_log_file = f"{name}_log{self.test_log_suffix}"
+ fvp_log_path = pathlib.Path(self.test_log_path, fvp_log_file)
+ fvp_log_symlink = pathlib.Path(self.test_log_path, f"{name}_log")
+ try:
+ os.remove(fvp_log_symlink)
+ except:
+ pass
+ os.symlink(fvp_log_file, fvp_log_symlink)
+ return open(fvp_log_path, 'wb')
+
+ async def _after_start(self):
+ for name, console in self.config["consoles"].items():
+ await self._add_terminal(name, console)
+
+ # testimage.bbclass expects to see a log file at `bootlog`,
+ # so make a symlink to the 'default' log file
+ if name == 'default':
+ default_test_file = f"{name}_log{self.test_log_suffix}"
+ os.symlink(default_test_file, self.bootlog)
+
+ async def _after_stop(self):
+ # Ensure pexpect logs all remaining output to the logfile
+ for terminal in self.terminals.values():
+ terminal.expect(pexpect.EOF, timeout=5)
+ terminal.close()
+
+ def _get_terminal(self, name):
+ return self.terminals[name]
+
+ def __getattr__(self, name):
+ """
+ Magic method which automatically exposes the whole pexpect API on the
+ target, with the first argument being the terminal name.
+
+ e.g. self.target.expect(self.target.DEFAULT_CONSOLE, "login\\:")
+ """
+ def call_pexpect(terminal, *args, **kwargs):
+ attr = getattr(self.terminals[terminal], name)
+ if callable(attr):
+ return attr(*args, **kwargs)
+ else:
+ return attr
+
+ return call_pexpect