mirror of
https://github.com/ARM-software/workload-automation.git
synced 2025-01-19 12:24:32 +00:00
234 lines
9.3 KiB
Python
234 lines
9.3 KiB
Python
# Copyright 2014-2015 ARM Limited
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
#
|
|
|
|
|
|
import re
|
|
import time
|
|
import logging
|
|
from copy import copy
|
|
|
|
from wlauto.exceptions import ConfigError
|
|
from wlauto.utils.serial_port import TIMEOUT
|
|
from wlauto.utils.types import boolean
|
|
|
|
|
|
logger = logging.getLogger('UEFI')
|
|
|
|
|
|
class UefiConfig(object):
|
|
|
|
def __init__(self, config_dict):
|
|
if isinstance(config_dict, UefiConfig):
|
|
self.__dict__ = copy(config_dict.__dict__)
|
|
else:
|
|
try:
|
|
self.image_name = config_dict['image_name']
|
|
self.image_args = config_dict['image_args']
|
|
self.fdt_support = boolean(config_dict['fdt_support'])
|
|
except KeyError as e:
|
|
raise ConfigError('Missing mandatory parameter for UEFI entry config: "{}"'.format(e))
|
|
self.initrd = config_dict.get('initrd')
|
|
self.fdt_path = config_dict.get('fdt_path')
|
|
if self.fdt_path and not self.fdt_support:
|
|
raise ConfigError('FDT path has been specfied for UEFI entry, when FDT support is "False"')
|
|
|
|
|
|
class UefiMenu(object):
|
|
"""
|
|
Allows navigating UEFI menu over serial (it relies on a pexpect connection).
|
|
|
|
"""
|
|
|
|
option_regex = re.compile(r'^\[(\d+)\]\s+([^\r]+)\r\n', re.M)
|
|
prompt_regex = re.compile(r'^(\S[^\r\n]+):\s*', re.M)
|
|
invalid_regex = re.compile(r'Invalid input \(max (\d+)\)', re.M)
|
|
|
|
load_delay = 1 # seconds
|
|
default_timeout = 60 # seconds
|
|
|
|
def __init__(self, conn, prompt='The default boot selection will start in'):
|
|
"""
|
|
:param conn: A serial connection as returned by ``pexect.spawn()``.
|
|
:param prompt: The starting prompt to wait for during ``open()``.
|
|
|
|
"""
|
|
self.conn = conn
|
|
self.start_prompt = prompt
|
|
self.options = {}
|
|
self.prompt = None
|
|
|
|
def open(self, timeout=default_timeout):
|
|
"""
|
|
"Open" the UEFI menu by sending an interrupt on STDIN after seeing the
|
|
starting prompt (configurable upon creation of the ``UefiMenu`` object.
|
|
|
|
"""
|
|
self.conn.expect(self.start_prompt, timeout)
|
|
self.conn.sendline('')
|
|
time.sleep(self.load_delay)
|
|
|
|
def create_entry(self, name, config):
|
|
"""Create a new UEFI entry using the parameters. The menu is assumed
|
|
to be at the top level. Upon return, the menu will be at the top level."""
|
|
logger.debug('Creating UEFI entry {}'.format(name))
|
|
self.nudge()
|
|
self.select('Boot Manager')
|
|
self.select('Add Boot Device Entry')
|
|
self.select('NOR Flash')
|
|
self.enter(config.image_name)
|
|
self.enter('y' if config.fdt_support else 'n')
|
|
if config.initrd:
|
|
self.enter('y')
|
|
self.enter(config.initrd)
|
|
else:
|
|
self.enter('n')
|
|
self.enter(config.image_args)
|
|
self.enter(name)
|
|
|
|
if config.fdt_path:
|
|
self.select('Update FDT path')
|
|
self.enter(config.fdt_path)
|
|
|
|
self.select('Return to main menu')
|
|
|
|
def delete_entry(self, name):
|
|
"""Delete the specified UEFI entry. The menu is assumed
|
|
to be at the top level. Upon return, the menu will be at the top level."""
|
|
logger.debug('Removing UEFI entry {}'.format(name))
|
|
self.nudge()
|
|
self.select('Boot Manager')
|
|
self.select('Remove Boot Device Entry')
|
|
self.select(name)
|
|
self.select('Return to main menu')
|
|
|
|
def select(self, option, timeout=default_timeout):
|
|
"""
|
|
Select the specified option from the current menu.
|
|
|
|
:param option: Could be an ``int`` index of the option, or a string/regex to
|
|
match option text against.
|
|
:param timeout: If a non-``int`` option is specified, the option list may need
|
|
need to be parsed (if it hasn't been already), this may block
|
|
and the timeout is used to cap that , resulting in a ``TIMEOUT``
|
|
exception.
|
|
:param delay: A fixed delay to wait after sending the input to the serial connection.
|
|
This should be set if input this action is known to result in a
|
|
long-running operation.
|
|
|
|
"""
|
|
if isinstance(option, basestring):
|
|
option = self.get_option_index(option, timeout)
|
|
self.enter(option)
|
|
|
|
def enter(self, value, delay=load_delay):
|
|
"""Like ``select()`` except no resolution is performed -- the value is sent directly
|
|
to the serial connection."""
|
|
# Empty the buffer first, so that only response to the input about to
|
|
# be sent will be processed by subsequent commands.
|
|
value = str(value)
|
|
self._reset()
|
|
self.write_characters(value)
|
|
# TODO: in case the value is long an complicated, things may get
|
|
# screwed up (e.g. there may be line breaks injected), additionally,
|
|
# special chars might cause regex to fail. To avoid these issues i'm
|
|
# only matching against the first 5 chars of the value. This is
|
|
# entirely arbitrary and I'll probably have to find a better way of
|
|
# doing this at some point.
|
|
self.conn.expect(value[:5], timeout=delay)
|
|
time.sleep(self.load_delay)
|
|
|
|
def read_menu(self, timeout=default_timeout):
|
|
"""Parse serial output to get the menu options and the following prompt."""
|
|
attempting_timeout_retry = False
|
|
attempting_invalid_retry = False
|
|
while True:
|
|
index = self.conn.expect([self.option_regex, self.prompt_regex, self.invalid_regex, TIMEOUT],
|
|
timeout=timeout)
|
|
match = self.conn.match
|
|
if index == 0: # matched menu option
|
|
self.options[match.group(1)] = match.group(2)
|
|
elif index == 1: # matched prompt
|
|
self.prompt = match.group(1)
|
|
break
|
|
elif index == 2: # matched invalid selection
|
|
# We've sent an invalid input (which includes an empty line) at
|
|
# the top-level menu. To get back the menu options, it seems we
|
|
# need to enter what the error reports as the max + 1, so...
|
|
if not attempting_invalid_retry:
|
|
attempting_invalid_retry = True
|
|
val = int(match.group(1)) + 1
|
|
self.empty_buffer()
|
|
self.enter(val)
|
|
else: # OK, that didn't work; panic!
|
|
raise RuntimeError('Could not read menu entries stuck on "{}" prompt'.format(self.prompt))
|
|
elif index == 3: # timed out
|
|
if not attempting_timeout_retry:
|
|
attempting_timeout_retry = True
|
|
self.nudge()
|
|
else: # Didn't help. Run away!
|
|
raise RuntimeError('Did not see a valid UEFI menu.')
|
|
else:
|
|
raise AssertionError('Unexpected response waiting for UEFI menu') # should never get here
|
|
|
|
def get_option_index(self, text, timeout=default_timeout):
|
|
"""Returns the menu index of the specified option text (uses regex matching). If the option
|
|
is not in the current menu, ``LookupError`` will be raised."""
|
|
if not self.prompt:
|
|
self.read_menu(timeout)
|
|
for k, v in self.options.iteritems():
|
|
if re.search(text, v):
|
|
return k
|
|
raise LookupError(text)
|
|
|
|
def has_option(self, text, timeout=default_timeout):
|
|
"""Returns ``True`` if at least one of the options in the current menu has
|
|
matched (using regex) the specified text."""
|
|
try:
|
|
self.get_option_index(text, timeout)
|
|
return True
|
|
except LookupError:
|
|
return False
|
|
|
|
def nudge(self):
|
|
"""Send a little nudge to ensure there is something to read. This is useful when you're not
|
|
sure if all out put from the serial has been read already."""
|
|
self.enter('')
|
|
|
|
def empty_buffer(self):
|
|
"""Read everything from the serial and clear the internal pexpect buffer. This ensures
|
|
that the next ``expect()`` call will time out (unless further input will be sent to the
|
|
serial beforehand. This is used to create a "known" state and avoid unexpected matches."""
|
|
try:
|
|
while True:
|
|
time.sleep(0.1)
|
|
self.conn.read_nonblocking(size=1024, timeout=0.1)
|
|
except TIMEOUT:
|
|
pass
|
|
self.conn.buffer = ''
|
|
|
|
def write_characters(self, line):
|
|
"""Write a single line out to serial charcter-by-character. This will ensure that nothing will
|
|
be dropped for longer lines."""
|
|
line = line.rstrip('\r\n')
|
|
for c in line:
|
|
self.conn.send(c)
|
|
time.sleep(0.05)
|
|
self.conn.sendline('')
|
|
|
|
def _reset(self):
|
|
self.options = {}
|
|
self.prompt = None
|
|
self.empty_buffer()
|