[RFC XEN PATCH 05/13] pytest: Introduce reading the console, and add a test

[Anthony PERARD <anthony@xenproject.org>]

From: Anthony PERARD <anthony.perard@vates.tech>

The test will do more than just checking dom0 booted in a later patch.

Signed-off-by: Anthony PERARD <anthony.perard@vates.tech>
---
 automation/pytest/lib/ipmi.py    | 86 ++++++++++++++++++++++++++++++++
 automation/pytest/pyproject.toml |  1 +
 automation/pytest/test_boot.py   | 24 +++++++++
 3 files changed, 111 insertions(+)

diff --git a/automation/pytest/lib/ipmi.py b/automation/pytest/lib/ipmi.py
index 0d3b085f704c..bda5d8077577 100644
--- a/automation/pytest/lib/ipmi.py
+++ b/automation/pytest/lib/ipmi.py
@@ -1,6 +1,8 @@
 import logging
 import paramiko
 import re
+import time
+from typing import Generator
 
 
 class IPMI:
@@ -67,3 +69,87 @@ class IPMI:
 
     def power_off(self):
         self._power('off force')
+
+    def get_console(self):
+        logging.info(f"Start console acquisition on {self.blade_node}")
+        ssh = self._new_client()
+        channel = ssh.invoke_shell()
+
+        output = ""
+        logging.debug("wait for prompt")
+        while True:
+            if channel.recv_ready():
+                output += channel.recv(9999).decode('utf-8')
+                # wait for prompt
+                logging.debug(output)
+                if output.strip().endswith('hpiLO->'):
+                    logging.debug("got prompt")
+                    break
+            elif channel.exit_status_ready():
+                raise Exception("Shell closed unexpectedly.")
+            else:
+                time.sleep(0.1)  # Wait a bit for more output
+
+        logging.debug("run connect node vsp acquire")
+        cmd = f"connect node vsp acquire {self.blade_node}\n"
+        channel.send(cmd)
+
+        self.channel = channel
+
+    def console_output_generator(self) -> Generator[str]:
+        if not self.channel:
+            raise Exception("Must call get_console")
+        channel = self.channel
+        previous_output = ''
+        # `count` when this reach 0, we had a line from the console that isn't
+        # finished, like a prompt, for sometime already, so display it in the
+        # debug.
+        count = 0
+        ansi_escape = re.compile(r'''
+            \x1B  # ESC
+            (?:   # 7-bit C1 Fe (except CSI)
+                [@-Z\\-_]
+            |     # or [ for CSI, followed by a control sequence
+                \[
+                [0-?]*  # Parameter bytes
+                [ -/]*  # Intermediate bytes
+                [@-~]   # Final byte
+            )
+        ''', re.VERBOSE)
+        while True:
+            if channel.recv_ready():
+                output = previous_output
+                output += channel.recv(9999).decode('utf-8')
+                output_lines = output.splitlines(keepends=True)
+                for line in output_lines[:-1]:
+                    line = ansi_escape.sub('', line)
+                    line = line.rstrip()
+                    if line in ['', '\r', '\r\n']:
+                        continue
+                    logging.debug(line)
+                    yield line
+                previous_output = output_lines[-1]
+                if previous_output.endswith('\n'):
+                    line = previous_output
+                    line = ansi_escape.sub('', line)
+                    line = line.rstrip()
+                    if line not in ['', '\r', '\r\n']:
+                        logging.debug(line)
+                    previous_output = ''
+                    count = 0
+                else:
+                    count = 20
+                # previous_output = output[-len(expect):]
+            # Check if the shell closed unexpectedly
+            elif channel.exit_status_ready():
+                raise Exception("Shell closed unexpectedly.")
+            else:
+                if count > 1:
+                    count -= 1
+                elif count == 1:
+                    line = previous_output
+                    line = ansi_escape.sub('', line)
+                    if line not in ['\r']:
+                        logging.debug("XXXXX output buffer: %s XXXXXX" % repr(line))
+                    count -= 1
+                time.sleep(0.5)  # Wait a bit for more output
diff --git a/automation/pytest/pyproject.toml b/automation/pytest/pyproject.toml
index 78f60c4f8e77..4ab7094e0d84 100644
--- a/automation/pytest/pyproject.toml
+++ b/automation/pytest/pyproject.toml
@@ -6,6 +6,7 @@ dependencies = [
     "paramiko>=4.0.0",
     "pip>=26.0.1",
     "pytest>=9.0.2",
+    "pytest-timeout>=2.4.0",
     "types-paramiko>=4.0.0.20260322",
 ]
 [dependency-groups]
diff --git a/automation/pytest/test_boot.py b/automation/pytest/test_boot.py
index b7610736fa25..158f1a30dcb6 100644
--- a/automation/pytest/test_boot.py
+++ b/automation/pytest/test_boot.py
@@ -1,3 +1,4 @@
+import logging
 import os
 from pathlib import Path
 import pytest
@@ -53,8 +54,31 @@ def machine_on_ipmi(prepare_netboot: None) -> Generator[IPMI]:
             password=os.environ['MOON_CONSOLE_PASS'],
             blade_node=os.environ['IPMI_BLADE_NODE'],
             )
+    # start console
+    ipmi.get_console()
     ipmi.power_off()
     ipmi.power_status()
     ipmi.power_on()
     yield ipmi
     ipmi.power_off()
+
+
+@pytest.mark.timeout(600)
+def test_xen_tools_tests(machine_on_ipmi: IPMI,
+                         ) -> None:
+    expect_string = "Dom0 ready"
+    for line in machine_on_ipmi.console_output_generator():
+        #
+        # Wait for `expect_string` for the test to succeed
+        #
+        if expect_string in line:
+            logging.info(f"got expect_string '{line}'.")
+            break
+        #
+        # Some common expected or not console outputs.
+        #
+        if "Latest ChangeSet:" in line:
+            logging.info(f"Test xen build from '{line}'.")
+        if "Manual reset required ('noreboot' specified)" in line:
+            # Test failed, could wait a bit before failing the test.
+            logging.error("Xen panic")
-- 
Anthony PERARD