1# Copyright (c) 2015 Stephen Warren
2# Copyright (c) 2015-2016, NVIDIA CORPORATION. All rights reserved.
3#
4# SPDX-License-Identifier: GPL-2.0
5
6# Common logic to interact with U-Boot via the console. This class provides
7# the interface that tests use to execute U-Boot shell commands and wait for
8# their results. Sub-classes exist to perform board-type-specific setup
9# operations, such as spawning a sub-process for Sandbox, or attaching to the
10# serial console of real hardware.
11
12import multiplexed_log
13import os
14import pytest
15import re
16import sys
17import u_boot_spawn
18
19# Regexes for text we expect U-Boot to send to the console.
20pattern_u_boot_spl_signon = re.compile('(U-Boot SPL \\d{4}\\.\\d{2}[^\r\n]*\\))')
21pattern_u_boot_main_signon = re.compile('(U-Boot \\d{4}\\.\\d{2}[^\r\n]*\\))')
22pattern_stop_autoboot_prompt = re.compile('Hit any key to stop autoboot: ')
23pattern_unknown_command = re.compile('Unknown command \'.*\' - try \'help\'')
24pattern_error_notification = re.compile('## Error: ')
25pattern_error_please_reset = re.compile('### ERROR ### Please RESET the board ###')
26
27PAT_ID = 0
28PAT_RE = 1
29
30bad_pattern_defs = (
31    ('spl_signon', pattern_u_boot_spl_signon),
32    ('main_signon', pattern_u_boot_main_signon),
33    ('stop_autoboot_prompt', pattern_stop_autoboot_prompt),
34    ('unknown_command', pattern_unknown_command),
35    ('error_notification', pattern_error_notification),
36    ('error_please_reset', pattern_error_please_reset),
37)
38
39class ConsoleDisableCheck(object):
40    """Context manager (for Python's with statement) that temporarily disables
41    the specified console output error check. This is useful when deliberately
42    executing a command that is known to trigger one of the error checks, in
43    order to test that the error condition is actually raised. This class is
44    used internally by ConsoleBase::disable_check(); it is not intended for
45    direct usage."""
46
47    def __init__(self, console, check_type):
48        self.console = console
49        self.check_type = check_type
50
51    def __enter__(self):
52        self.console.disable_check_count[self.check_type] += 1
53        self.console.eval_bad_patterns()
54
55    def __exit__(self, extype, value, traceback):
56        self.console.disable_check_count[self.check_type] -= 1
57        self.console.eval_bad_patterns()
58
59class ConsoleSetupTimeout(object):
60    """Context manager (for Python's with statement) that temporarily sets up
61    timeout for specific command. This is useful when execution time is greater
62    then default 30s."""
63
64    def __init__(self, console, timeout):
65        self.p = console.p
66        self.orig_timeout = self.p.timeout
67        self.p.timeout = timeout
68
69    def __enter__(self):
70        return self
71
72    def __exit__(self, extype, value, traceback):
73        self.p.timeout = self.orig_timeout
74
75class ConsoleBase(object):
76    """The interface through which test functions interact with the U-Boot
77    console. This primarily involves executing shell commands, capturing their
78    results, and checking for common error conditions. Some common utilities
79    are also provided too."""
80
81    def __init__(self, log, config, max_fifo_fill):
82        """Initialize a U-Boot console connection.
83
84        Can only usefully be called by sub-classes.
85
86        Args:
87            log: A mulptiplex_log.Logfile object, to which the U-Boot output
88                will be logged.
89            config: A configuration data structure, as built by conftest.py.
90            max_fifo_fill: The maximum number of characters to send to U-Boot
91                command-line before waiting for U-Boot to echo the characters
92                back. For UART-based HW without HW flow control, this value
93                should be set less than the UART RX FIFO size to avoid
94                overflow, assuming that U-Boot can't keep up with full-rate
95                traffic at the baud rate.
96
97        Returns:
98            Nothing.
99        """
100
101        self.log = log
102        self.config = config
103        self.max_fifo_fill = max_fifo_fill
104
105        self.logstream = self.log.get_stream('console', sys.stdout)
106
107        # Array slice removes leading/trailing quotes
108        self.prompt = self.config.buildconfig['config_sys_prompt'][1:-1]
109        self.prompt_escaped = re.escape(self.prompt)
110        self.p = None
111        self.disable_check_count = {pat[PAT_ID]: 0 for pat in bad_pattern_defs}
112        self.eval_bad_patterns()
113
114        self.at_prompt = False
115        self.at_prompt_logevt = None
116
117    def eval_bad_patterns(self):
118        self.bad_patterns = [pat[PAT_RE] for pat in bad_pattern_defs \
119            if self.disable_check_count[pat[PAT_ID]] == 0]
120        self.bad_pattern_ids = [pat[PAT_ID] for pat in bad_pattern_defs \
121            if self.disable_check_count[pat[PAT_ID]] == 0]
122
123    def close(self):
124        """Terminate the connection to the U-Boot console.
125
126        This function is only useful once all interaction with U-Boot is
127        complete. Once this function is called, data cannot be sent to or
128        received from U-Boot.
129
130        Args:
131            None.
132
133        Returns:
134            Nothing.
135        """
136
137        if self.p:
138            self.p.close()
139        self.logstream.close()
140
141    def run_command(self, cmd, wait_for_echo=True, send_nl=True,
142            wait_for_prompt=True):
143        """Execute a command via the U-Boot console.
144
145        The command is always sent to U-Boot.
146
147        U-Boot echoes any command back to its output, and this function
148        typically waits for that to occur. The wait can be disabled by setting
149        wait_for_echo=False, which is useful e.g. when sending CTRL-C to
150        interrupt a long-running command such as "ums".
151
152        Command execution is typically triggered by sending a newline
153        character. This can be disabled by setting send_nl=False, which is
154        also useful when sending CTRL-C.
155
156        This function typically waits for the command to finish executing, and
157        returns the console output that it generated. This can be disabled by
158        setting wait_for_prompt=False, which is useful when invoking a long-
159        running command such as "ums".
160
161        Args:
162            cmd: The command to send.
163            wait_for_each: Boolean indicating whether to wait for U-Boot to
164                echo the command text back to its output.
165            send_nl: Boolean indicating whether to send a newline character
166                after the command string.
167            wait_for_prompt: Boolean indicating whether to wait for the
168                command prompt to be sent by U-Boot. This typically occurs
169                immediately after the command has been executed.
170
171        Returns:
172            If wait_for_prompt == False:
173                Nothing.
174            Else:
175                The output from U-Boot during command execution. In other
176                words, the text U-Boot emitted between the point it echod the
177                command string and emitted the subsequent command prompts.
178        """
179
180        if self.at_prompt and \
181                self.at_prompt_logevt != self.logstream.logfile.cur_evt:
182            self.logstream.write(self.prompt, implicit=True)
183
184        try:
185            self.at_prompt = False
186            if send_nl:
187                cmd += '\n'
188            while cmd:
189                # Limit max outstanding data, so UART FIFOs don't overflow
190                chunk = cmd[:self.max_fifo_fill]
191                cmd = cmd[self.max_fifo_fill:]
192                self.p.send(chunk)
193                if not wait_for_echo:
194                    continue
195                chunk = re.escape(chunk)
196                chunk = chunk.replace('\\\n', '[\r\n]')
197                m = self.p.expect([chunk] + self.bad_patterns)
198                if m != 0:
199                    self.at_prompt = False
200                    raise Exception('Bad pattern found on console: ' +
201                                    self.bad_pattern_ids[m - 1])
202            if not wait_for_prompt:
203                return
204            m = self.p.expect([self.prompt_escaped] + self.bad_patterns)
205            if m != 0:
206                self.at_prompt = False
207                raise Exception('Bad pattern found on console: ' +
208                                self.bad_pattern_ids[m - 1])
209            self.at_prompt = True
210            self.at_prompt_logevt = self.logstream.logfile.cur_evt
211            # Only strip \r\n; space/TAB might be significant if testing
212            # indentation.
213            return self.p.before.strip('\r\n')
214        except Exception as ex:
215            self.log.error(str(ex))
216            self.cleanup_spawn()
217            raise
218
219    def run_command_list(self, cmds):
220        """Run a list of commands.
221
222        This is a helper function to call run_command() with default arguments
223        for each command in a list.
224
225        Args:
226            cmd: List of commands (each a string)
227        Returns:
228            Combined output of all commands, as a string
229        """
230        output = ''
231        for cmd in cmds:
232            output += self.run_command(cmd)
233        return output
234
235    def ctrlc(self):
236        """Send a CTRL-C character to U-Boot.
237
238        This is useful in order to stop execution of long-running synchronous
239        commands such as "ums".
240
241        Args:
242            None.
243
244        Returns:
245            Nothing.
246        """
247
248        self.log.action('Sending Ctrl-C')
249        self.run_command(chr(3), wait_for_echo=False, send_nl=False)
250
251    def wait_for(self, text):
252        """Wait for a pattern to be emitted by U-Boot.
253
254        This is useful when a long-running command such as "dfu" is executing,
255        and it periodically emits some text that should show up at a specific
256        location in the log file.
257
258        Args:
259            text: The text to wait for; either a string (containing raw text,
260                not a regular expression) or an re object.
261
262        Returns:
263            Nothing.
264        """
265
266        if type(text) == type(''):
267            text = re.escape(text)
268        m = self.p.expect([text] + self.bad_patterns)
269        if m != 0:
270            raise Exception('Bad pattern found on console: ' +
271                            self.bad_pattern_ids[m - 1])
272
273    def drain_console(self):
274        """Read from and log the U-Boot console for a short time.
275
276        U-Boot's console output is only logged when the test code actively
277        waits for U-Boot to emit specific data. There are cases where tests
278        can fail without doing this. For example, if a test asks U-Boot to
279        enable USB device mode, then polls until a host-side device node
280        exists. In such a case, it is useful to log U-Boot's console output
281        in case U-Boot printed clues as to why the host-side even did not
282        occur. This function will do that.
283
284        Args:
285            None.
286
287        Returns:
288            Nothing.
289        """
290
291        # If we are already not connected to U-Boot, there's nothing to drain.
292        # This should only happen when a previous call to run_command() or
293        # wait_for() failed (and hence the output has already been logged), or
294        # the system is shutting down.
295        if not self.p:
296            return
297
298        orig_timeout = self.p.timeout
299        try:
300            # Drain the log for a relatively short time.
301            self.p.timeout = 1000
302            # Wait for something U-Boot will likely never send. This will
303            # cause the console output to be read and logged.
304            self.p.expect(['This should never match U-Boot output'])
305        except u_boot_spawn.Timeout:
306            pass
307        finally:
308            self.p.timeout = orig_timeout
309
310    def ensure_spawned(self):
311        """Ensure a connection to a correctly running U-Boot instance.
312
313        This may require spawning a new Sandbox process or resetting target
314        hardware, as defined by the implementation sub-class.
315
316        This is an internal function and should not be called directly.
317
318        Args:
319            None.
320
321        Returns:
322            Nothing.
323        """
324
325        if self.p:
326            return
327        try:
328            self.log.start_section('Starting U-Boot')
329            self.at_prompt = False
330            self.p = self.get_spawn()
331            # Real targets can take a long time to scroll large amounts of
332            # text if LCD is enabled. This value may need tweaking in the
333            # future, possibly per-test to be optimal. This works for 'help'
334            # on board 'seaboard'.
335            if not self.config.gdbserver:
336                self.p.timeout = 30000
337            self.p.logfile_read = self.logstream
338            bcfg = self.config.buildconfig
339            config_spl = bcfg.get('config_spl', 'n') == 'y'
340            config_spl_serial_support = bcfg.get('config_spl_serial_support',
341                                                 'n') == 'y'
342            env_spl_skipped = self.config.env.get('env__spl_skipped',
343                                                  False)
344            if config_spl and config_spl_serial_support and not env_spl_skipped:
345                m = self.p.expect([pattern_u_boot_spl_signon] +
346                                  self.bad_patterns)
347                if m != 0:
348                    raise Exception('Bad pattern found on SPL console: ' +
349                                    self.bad_pattern_ids[m - 1])
350            m = self.p.expect([pattern_u_boot_main_signon] + self.bad_patterns)
351            if m != 0:
352                raise Exception('Bad pattern found on console: ' +
353                                self.bad_pattern_ids[m - 1])
354            self.u_boot_version_string = self.p.after
355            while True:
356                m = self.p.expect([self.prompt_escaped,
357                    pattern_stop_autoboot_prompt] + self.bad_patterns)
358                if m == 0:
359                    break
360                if m == 1:
361                    self.p.send(' ')
362                    continue
363                raise Exception('Bad pattern found on console: ' +
364                                self.bad_pattern_ids[m - 2])
365            self.at_prompt = True
366            self.at_prompt_logevt = self.logstream.logfile.cur_evt
367        except Exception as ex:
368            self.log.error(str(ex))
369            self.cleanup_spawn()
370            raise
371        finally:
372            self.log.end_section('Starting U-Boot')
373
374    def cleanup_spawn(self):
375        """Shut down all interaction with the U-Boot instance.
376
377        This is used when an error is detected prior to re-establishing a
378        connection with a fresh U-Boot instance.
379
380        This is an internal function and should not be called directly.
381
382        Args:
383            None.
384
385        Returns:
386            Nothing.
387        """
388
389        try:
390            if self.p:
391                self.p.close()
392        except:
393            pass
394        self.p = None
395
396    def get_spawn_output(self):
397        """Return the start-up output from U-Boot
398
399        Returns:
400            The output produced by ensure_spawed(), as a string.
401        """
402        if self.p:
403            return self.p.get_expect_output()
404        return None
405
406    def validate_version_string_in_text(self, text):
407        """Assert that a command's output includes the U-Boot signon message.
408
409        This is primarily useful for validating the "version" command without
410        duplicating the signon text regex in a test function.
411
412        Args:
413            text: The command output text to check.
414
415        Returns:
416            Nothing. An exception is raised if the validation fails.
417        """
418
419        assert(self.u_boot_version_string in text)
420
421    def disable_check(self, check_type):
422        """Temporarily disable an error check of U-Boot's output.
423
424        Create a new context manager (for use with the "with" statement) which
425        temporarily disables a particular console output error check.
426
427        Args:
428            check_type: The type of error-check to disable. Valid values may
429            be found in self.disable_check_count above.
430
431        Returns:
432            A context manager object.
433        """
434
435        return ConsoleDisableCheck(self, check_type)
436
437    def temporary_timeout(self, timeout):
438        """Temporarily set up different timeout for commands.
439
440        Create a new context manager (for use with the "with" statement) which
441        temporarily change timeout.
442
443        Args:
444            timeout: Time in milliseconds.
445
446        Returns:
447            A context manager object.
448        """
449
450        return ConsoleSetupTimeout(self, timeout)
451