1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
import pathlib
import pexpect
import os
from oeqa.core.target.ssh import OESSHTarget
from fvp import conffile, runner
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, dir_image=None, rootfs=None, bootlog=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)
self.bootlog = bootlog
if not self.fvpconf.exists():
raise FileNotFoundError(f"Cannot find {self.fvpconf}")
def _after_start(self):
pass
def start(self, **kwargs):
self.fvp_log = self._create_logfile("fvp")
self.fvp = runner.FVPRunner(self.logger)
self.fvp.start(self.config, stdout=self.fvp_log)
self.logger.debug(f"Started FVP PID {self.fvp.pid()}")
self._after_start()
def stop(self, **kwargs):
returncode = self.fvp.stop()
self.logger.debug(f"Stopped FVP with return code {returncode}")
def _create_logfile(self, name):
if not self.bootlog:
return None
test_log_path = pathlib.Path(self.bootlog).parent
test_log_suffix = pathlib.Path(self.bootlog).suffix
fvp_log_file = f"{name}_log{test_log_suffix}"
fvp_log_path = pathlib.Path(test_log_path, fvp_log_file)
fvp_log_symlink = pathlib.Path(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')
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, **kwargs):
super().__init__(logger, target_ip, server_ip, **kwargs)
self.logfile = self.bootlog and open(self.bootlog, "wb") or None
# FVPs boot slowly, so allow ten minutes
self.boot_timeout = 10 * 60
def _after_start(self):
with open(self.fvp_log.name, 'rb') as logfile:
parser = runner.ConsolePortParser(logfile)
self.logger.debug(f"Awaiting console on terminal {self.config['consoles']['default']}")
port = parser.parse_port(self.config['consoles']['default'])
console = self.fvp.create_pexpect(port)
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, e.g. by depending on the
"linuxboot" test case in meta-arm.
"""
DEFAULT_CONSOLE = "default"
def __init__(self, logger, target_ip, server_ip, **kwargs):
super().__init__(logger, target_ip, server_ip, **kwargs)
self.terminals = {}
def _after_start(self):
with open(self.fvp_log.name, 'rb') as logfile:
parser = runner.ConsolePortParser(logfile)
for name, console in self.config["consoles"].items():
logfile = self._create_logfile(name)
self.logger.info(f'Creating terminal {name} on {console}')
port = parser.parse_port(console)
self.terminals[name] = \
self.fvp.create_pexpect(port, logfile=logfile)
# 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)
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
|
