~funderscore blog cgit wiki get in touch
aboutsummaryrefslogtreecommitdiff
blob: 9e161fbc238803ca7fe2ceb427eda95d98d91346 (plain)
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
# 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 pathlib
import signal
import sys
import time
import re
import pytest

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:
    """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 _ 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, stdin=None, env=None):
    """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.
        stdin: Input string to pass to the command as stdin (or None)
            env: Environment to use, or None to use the current one

    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, stdin=stdin, env=env)
    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:
        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). Besides, on some RISC-V targets the low memory
    # is protected that prevents S-mode U-Boot from access.
    # Let's add 2MiB then (size of an ARM LPAE/v8 section).

    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)
                pathlib.Path(self.filename).unlink()

    def __exit__(self, extype, value, traceback):
        if extype:
            try:
                pathlib.Path(self.filename).unlink()
            except Exception:
                pass
            return
        logged = False
        for _ 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)

def waitpid(pid, timeout=60, kill=False):
    """Wait a process to terminate by its PID

    This is an alternative to a os.waitpid(pid, 0) call that works on
    processes that aren't children of the python process.

    Args:
        pid: PID of a running process.
        timeout: Time in seconds to wait.
        kill: Whether to forcibly kill the process after timeout.

    Returns:
        True, if the process ended on its own.
        False, if the process was killed by this function.

    Raises:
        TimeoutError, if the process is still running after timeout.
    """
    try:
        for _ in range(timeout):
            os.kill(pid, 0)
            time.sleep(1)

        if kill:
            os.kill(pid, signal.SIGKILL)
            return False

    except ProcessLookupError:
        return True

    raise TimeoutError(
        "Process with PID {} did not terminate after {} seconds."
        .format(pid, timeout)
    )