|
| 1 | +import re |
1 | 2 | from typing import List |
2 | 3 |
|
3 | 4 | from lisa.feature import Feature |
@@ -44,17 +45,58 @@ def _execute(self, commands: List[str]) -> List[str]: |
44 | 45 | # write a newline and read to make sure serial console has the prompt |
45 | 46 | serial_console.write("\n") |
46 | 47 | response = serial_console.read() |
47 | | - if not response or "$" not in response and "#" not in response: |
48 | | - raise LisaException("Serial console prompt not found in output") |
| 48 | + |
| 49 | + # Check for full prompt pattern instead of individual characters |
| 50 | + if not self._is_valid_prompt(response): |
| 51 | + raise LisaException( |
| 52 | + f"Valid shell prompt not found in output. " |
| 53 | + f"Expected a shell prompt ending with $, #, or >, " |
| 54 | + f"but got: {response.strip()}" |
| 55 | + ) |
| 56 | + |
49 | 57 | for command in commands: |
50 | 58 | serial_console.write(self._add_newline(command)) |
51 | 59 | out.append(serial_console.read()) |
| 60 | + collected_info = "\n\n".join(out) |
| 61 | + self._log.info( |
| 62 | + f"Collected information using NonSshExecutor:\n{collected_info}" |
| 63 | + ) |
52 | 64 | return out |
53 | 65 | except Exception as e: |
54 | 66 | raise LisaException(f"Failed to execute commands: {e}") from e |
55 | 67 | finally: |
56 | 68 | serial_console.close() |
57 | 69 |
|
| 70 | + def _is_valid_prompt(self, response: str) -> bool: |
| 71 | + """ |
| 72 | + Check if the response contains a valid shell prompt pattern. |
| 73 | +
|
| 74 | + :param response: The response from the serial console |
| 75 | + :return: True if a valid prompt is found, False otherwise |
| 76 | + """ |
| 77 | + if not response: |
| 78 | + return False |
| 79 | + |
| 80 | + # Generic pattern that matches any prompt format: |
| 81 | + # - Username and hostname part: word chars, @, hyphens, dots |
| 82 | + # - Colon separator |
| 83 | + # - Path part: ~, /, word chars, dots, hyphens, slashes |
| 84 | + # - Optional whitespace |
| 85 | + # - Ending with $, #, or > |
| 86 | + # - Optional trailing whitespace |
| 87 | + prompt_pattern = r"[[email protected]]+:[~/a-zA-Z0-9_./-]*\s*[\$#>]\s*$" |
| 88 | + |
| 89 | + # Check each line in the response for the prompt pattern |
| 90 | + lines = response.split("\n") |
| 91 | + for line in lines: |
| 92 | + line = line.strip() |
| 93 | + if re.search(prompt_pattern, line): |
| 94 | + self._log.debug(f"Valid prompt found: '{line}'") |
| 95 | + return True |
| 96 | + |
| 97 | + self._log.debug(f"No valid prompt found in response: '{response.strip()}'") |
| 98 | + return False |
| 99 | + |
58 | 100 | def _add_newline(self, command: str) -> str: |
59 | 101 | """ |
60 | 102 | Adds a newline character to the command if it does not already end with one. |
|
0 commit comments