341 lines
11 KiB
Python
341 lines
11 KiB
Python
# SPDX-License-Identifier: GPL-2.0
|
|
# Copyright (c) 2016, NVIDIA CORPORATION. All rights reserved.
|
|
|
|
# Utility code shared across multiple tests.
|
|
|
|
import hashlib
|
|
import inspect
|
|
import os
|
|
import os.path
|
|
import pytest
|
|
import sys
|
|
import time
|
|
import re
|
|
|
|
def md5sum_data(data):
|
|
"""Calculate the MD5 hash of some data.
|
|
|
|
Args:
|
|
data: The data to hash.
|
|
|
|
Returns:
|
|
The hash of the data, as a binary string.
|
|
"""
|
|
|
|
h = hashlib.md5()
|
|
h.update(data)
|
|
return h.digest()
|
|
|
|
def md5sum_file(fn, max_length=None):
|
|
"""Calculate the MD5 hash of the contents of a file.
|
|
|
|
Args:
|
|
fn: The filename of the file to hash.
|
|
max_length: The number of bytes to hash. If the file has more
|
|
bytes than this, they will be ignored. If None or omitted, the
|
|
entire file will be hashed.
|
|
|
|
Returns:
|
|
The hash of the file content, as a binary string.
|
|
"""
|
|
|
|
with open(fn, 'rb') as fh:
|
|
if max_length:
|
|
params = [max_length]
|
|
else:
|
|
params = []
|
|
data = fh.read(*params)
|
|
return md5sum_data(data)
|
|
|
|
class PersistentRandomFile(object):
|
|
"""Generate and store information about a persistent file containing
|
|
random data."""
|
|
|
|
def __init__(self, u_boot_console, fn, size):
|
|
"""Create or process the persistent file.
|
|
|
|
If the file does not exist, it is generated.
|
|
|
|
If the file does exist, its content is hashed for later comparison.
|
|
|
|
These files are always located in the "persistent data directory" of
|
|
the current test run.
|
|
|
|
Args:
|
|
u_boot_console: A console connection to U-Boot.
|
|
fn: The filename (without path) to create.
|
|
size: The desired size of the file in bytes.
|
|
|
|
Returns:
|
|
Nothing.
|
|
"""
|
|
|
|
self.fn = fn
|
|
|
|
self.abs_fn = u_boot_console.config.persistent_data_dir + '/' + fn
|
|
|
|
if os.path.exists(self.abs_fn):
|
|
u_boot_console.log.action('Persistent data file ' + self.abs_fn +
|
|
' already exists')
|
|
self.content_hash = md5sum_file(self.abs_fn)
|
|
else:
|
|
u_boot_console.log.action('Generating ' + self.abs_fn +
|
|
' (random, persistent, %d bytes)' % size)
|
|
data = os.urandom(size)
|
|
with open(self.abs_fn, 'wb') as fh:
|
|
fh.write(data)
|
|
self.content_hash = md5sum_data(data)
|
|
|
|
def attempt_to_open_file(fn):
|
|
"""Attempt to open a file, without throwing exceptions.
|
|
|
|
Any errors (exceptions) that occur during the attempt to open the file
|
|
are ignored. This is useful in order to test whether a file (in
|
|
particular, a device node) exists and can be successfully opened, in order
|
|
to poll for e.g. USB enumeration completion.
|
|
|
|
Args:
|
|
fn: The filename to attempt to open.
|
|
|
|
Returns:
|
|
An open file handle to the file, or None if the file could not be
|
|
opened.
|
|
"""
|
|
|
|
try:
|
|
return open(fn, 'rb')
|
|
except:
|
|
return None
|
|
|
|
def wait_until_open_succeeds(fn):
|
|
"""Poll until a file can be opened, or a timeout occurs.
|
|
|
|
Continually attempt to open a file, and return when this succeeds, or
|
|
raise an exception after a timeout.
|
|
|
|
Args:
|
|
fn: The filename to attempt to open.
|
|
|
|
Returns:
|
|
An open file handle to the file.
|
|
"""
|
|
|
|
for i in range(100):
|
|
fh = attempt_to_open_file(fn)
|
|
if fh:
|
|
return fh
|
|
time.sleep(0.1)
|
|
raise Exception('File could not be opened')
|
|
|
|
def wait_until_file_open_fails(fn, ignore_errors):
|
|
"""Poll until a file cannot be opened, or a timeout occurs.
|
|
|
|
Continually attempt to open a file, and return when this fails, or
|
|
raise an exception after a timeout.
|
|
|
|
Args:
|
|
fn: The filename to attempt to open.
|
|
ignore_errors: Indicate whether to ignore timeout errors. If True, the
|
|
function will simply return if a timeout occurs, otherwise an
|
|
exception will be raised.
|
|
|
|
Returns:
|
|
Nothing.
|
|
"""
|
|
|
|
for i in range(100):
|
|
fh = attempt_to_open_file(fn)
|
|
if not fh:
|
|
return
|
|
fh.close()
|
|
time.sleep(0.1)
|
|
if ignore_errors:
|
|
return
|
|
raise Exception('File can still be opened')
|
|
|
|
def run_and_log(u_boot_console, cmd, ignore_errors=False):
|
|
"""Run a command and log its output.
|
|
|
|
Args:
|
|
u_boot_console: A console connection to U-Boot.
|
|
cmd: The command to run, as an array of argv[], or a string.
|
|
If a string, note that it is split up so that quoted spaces
|
|
will not be preserved. E.g. "fred and" becomes ['"fred', 'and"']
|
|
ignore_errors: Indicate whether to ignore errors. If True, the function
|
|
will simply return if the command cannot be executed or exits with
|
|
an error code, otherwise an exception will be raised if such
|
|
problems occur.
|
|
|
|
Returns:
|
|
The output as a string.
|
|
"""
|
|
if isinstance(cmd, str):
|
|
cmd = cmd.split()
|
|
runner = u_boot_console.log.get_runner(cmd[0], sys.stdout)
|
|
output = runner.run(cmd, ignore_errors=ignore_errors)
|
|
runner.close()
|
|
return output
|
|
|
|
def run_and_log_expect_exception(u_boot_console, cmd, retcode, msg):
|
|
"""Run a command that is expected to fail.
|
|
|
|
This runs a command and checks that it fails with the expected return code
|
|
and exception method. If not, an exception is raised.
|
|
|
|
Args:
|
|
u_boot_console: A console connection to U-Boot.
|
|
cmd: The command to run, as an array of argv[].
|
|
retcode: Expected non-zero return code from the command.
|
|
msg: String that should be contained within the command's output.
|
|
"""
|
|
try:
|
|
runner = u_boot_console.log.get_runner(cmd[0], sys.stdout)
|
|
runner.run(cmd)
|
|
except Exception as e:
|
|
assert(retcode == runner.exit_status)
|
|
assert(msg in runner.output)
|
|
else:
|
|
raise Exception("Expected an exception with retcode %d message '%s',"
|
|
"but it was not raised" % (retcode, msg))
|
|
finally:
|
|
runner.close()
|
|
|
|
ram_base = None
|
|
def find_ram_base(u_boot_console):
|
|
"""Find the running U-Boot's RAM location.
|
|
|
|
Probe the running U-Boot to determine the address of the first bank
|
|
of RAM. This is useful for tests that test reading/writing RAM, or
|
|
load/save files that aren't associated with some standard address
|
|
typically represented in an environment variable such as
|
|
${kernel_addr_r}. The value is cached so that it only needs to be
|
|
actively read once.
|
|
|
|
Args:
|
|
u_boot_console: A console connection to U-Boot.
|
|
|
|
Returns:
|
|
The address of U-Boot's first RAM bank, as an integer.
|
|
"""
|
|
|
|
global ram_base
|
|
if u_boot_console.config.buildconfig.get('config_cmd_bdi', 'n') != 'y':
|
|
pytest.skip('bdinfo command not supported')
|
|
if ram_base == -1:
|
|
pytest.skip('Previously failed to find RAM bank start')
|
|
if ram_base is not None:
|
|
return ram_base
|
|
|
|
with u_boot_console.log.section('find_ram_base'):
|
|
response = u_boot_console.run_command('bdinfo')
|
|
for l in response.split('\n'):
|
|
if '-> start' in l or 'memstart =' in l:
|
|
ram_base = int(l.split('=')[1].strip(), 16)
|
|
break
|
|
if ram_base is None:
|
|
ram_base = -1
|
|
raise Exception('Failed to find RAM bank start in `bdinfo`')
|
|
|
|
# We don't want ram_base to be zero as some functions test if the given
|
|
# address is NULL (0). Let's add 2MiB then (size of an ARM LPAE/v8 section).
|
|
|
|
if ram_base == 0:
|
|
ram_base += 1024 * 1024 * 2
|
|
|
|
return ram_base
|
|
|
|
class PersistentFileHelperCtxMgr(object):
|
|
"""A context manager for Python's "with" statement, which ensures that any
|
|
generated file is deleted (and hence regenerated) if its mtime is older
|
|
than the mtime of the Python module which generated it, and gets an mtime
|
|
newer than the mtime of the Python module which generated after it is
|
|
generated. Objects of this type should be created by factory function
|
|
persistent_file_helper rather than directly."""
|
|
|
|
def __init__(self, log, filename):
|
|
"""Initialize a new object.
|
|
|
|
Args:
|
|
log: The Logfile object to log to.
|
|
filename: The filename of the generated file.
|
|
|
|
Returns:
|
|
Nothing.
|
|
"""
|
|
|
|
self.log = log
|
|
self.filename = filename
|
|
|
|
def __enter__(self):
|
|
frame = inspect.stack()[1]
|
|
module = inspect.getmodule(frame[0])
|
|
self.module_filename = module.__file__
|
|
self.module_timestamp = os.path.getmtime(self.module_filename)
|
|
|
|
if os.path.exists(self.filename):
|
|
filename_timestamp = os.path.getmtime(self.filename)
|
|
if filename_timestamp < self.module_timestamp:
|
|
self.log.action('Removing stale generated file ' +
|
|
self.filename)
|
|
os.unlink(self.filename)
|
|
|
|
def __exit__(self, extype, value, traceback):
|
|
if extype:
|
|
try:
|
|
os.path.unlink(self.filename)
|
|
except:
|
|
pass
|
|
return
|
|
logged = False
|
|
for i in range(20):
|
|
filename_timestamp = os.path.getmtime(self.filename)
|
|
if filename_timestamp > self.module_timestamp:
|
|
break
|
|
if not logged:
|
|
self.log.action(
|
|
'Waiting for generated file timestamp to increase')
|
|
logged = True
|
|
os.utime(self.filename)
|
|
time.sleep(0.1)
|
|
|
|
def persistent_file_helper(u_boot_log, filename):
|
|
"""Manage the timestamps and regeneration of a persistent generated
|
|
file. This function creates a context manager for Python's "with"
|
|
statement
|
|
|
|
Usage:
|
|
with persistent_file_helper(u_boot_console.log, filename):
|
|
code to generate the file, if it's missing.
|
|
|
|
Args:
|
|
u_boot_log: u_boot_console.log.
|
|
filename: The filename of the generated file.
|
|
|
|
Returns:
|
|
A context manager object.
|
|
"""
|
|
|
|
return PersistentFileHelperCtxMgr(u_boot_log, filename)
|
|
|
|
def crc32(u_boot_console, address, count):
|
|
"""Helper function used to compute the CRC32 value of a section of RAM.
|
|
|
|
Args:
|
|
u_boot_console: A U-Boot console connection.
|
|
address: Address where data starts.
|
|
count: Amount of data to use for calculation.
|
|
|
|
Returns:
|
|
CRC32 value
|
|
"""
|
|
|
|
bcfg = u_boot_console.config.buildconfig
|
|
has_cmd_crc32 = bcfg.get('config_cmd_crc32', 'n') == 'y'
|
|
assert has_cmd_crc32, 'Cannot compute crc32 without CONFIG_CMD_CRC32.'
|
|
output = u_boot_console.run_command('crc32 %08x %x' % (address, count))
|
|
|
|
m = re.search('==> ([0-9a-fA-F]{8})$', output)
|
|
assert m, 'CRC32 operation failed.'
|
|
|
|
return m.group(1)
|