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
|
# Copyright (c) 2016, NVIDIA CORPORATION. All rights reserved.
#
# SPDX-License-Identifier: GPL-2.0
# Utility code shared across multiple tests.
import hashlib
import os
import os.path
import sys
import time
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 xrange(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 xrange(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[].
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:
Nothing.
'''
runner = u_boot_console.log.get_runner(cmd[0], sys.stdout)
runner.run(cmd, ignore_errors=ignore_errors)
runner.close()
|
