Version 2.8.5: Prompt hinter Steuerzeichen wird erkannt

apps/tesm/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "tesm"
-version = "2.8.4"
+version = "2.8.5"
 description = "TESM -- Verwaltung und Ueberwachung von PoE-Endgeraeten und Aruba-Switchen"
 requires-python = ">=3.11"
 dependencies = [

apps/tesm/src/tesm/cli.py

@@ -318,6 +318,15 @@ def cmd_switch_test(args: argparse.Namespace) -> int:
                             return 1
             except SshError as fehler:
                 sagen("FEHLGESCHLAGEN", str(fehler))
+                # Gerade dann sind die Bytes interessant: "kein Prompt" kann
+                # heissen, dass nichts kam -- oder dass etwas kam, das nicht
+                # erkannt wurde. Das sieht man nur im Rohtext, und
+                # Steuerzeichen sieht man nur als repr().
+                roh = "".join(sitzung.transcript)
+                print()
+                print("--- rohe Antwort des Switches (mit Steuerzeichen) ---")
+                print(repr(roh) if roh else "(nichts)")
+                print("--- Ende ---")
                 return 1
     print()
     print("Fertig.")

apps/tesm/src/tesm/services/switchcli.py

@@ -30,9 +30,26 @@ from .ssh import SshError, open_transport
 
 logger = logging.getLogger(__name__)
 
+#: Steuerzeichen eines Terminals -- ANSI-Sequenzen, Glocke, Einzelbefehle.
+#:
+#: Ein ArubaOS-Switch an einem Pseudoterminal zeichnet seine Zeile neu und
+#: haengt dabei Sequenzen wie ``ESC [ K`` ("bis Zeilenende loeschen") an. Im
+#: Terminal sieht man davon **nichts** -- und genau das war der Fehler, der
+#: Konsole und PoE-Neustart lahmgelegt hat: der Prompt stand da, aber hinter
+#: ihm standen drei unsichtbare Bytes, und das verankerte Muster traf nicht
+#: mehr. Im Protokoll las man "kein Prompt", waehrend der Prompt im Text
+#: stand. Deshalb kommen die Steuerzeichen weg, bevor irgendetwas gesucht
+#: wird -- das macht zugleich die Ausgabe in der Konsole lesbar.
+ANSI_RE = re.compile(rb"\x1b\[[0-9;?]*[ -/]*[@-~]|\x1b[@-Z\\-_]|[\x07\x00]")
+
 #: Erkennt Operator- (``>``) und privilegierten Modus (``#``) sowie den
 #: Konfigurationsmodus (``switch(config)#``).
-PROMPT_RE = re.compile(rb"[\r\n][^\r\n]{0,80}?[>#]\s?$")
+#:
+#: Zwei Lockerungen gegenueber der ersten Fassung, beide aus dem Betrieb:
+#: der Prompt darf am Anfang des Puffers stehen (nicht nur nach einem
+#: Zeilenumbruch -- so kommt er nach einem geleerten Puffer), und hinter ihm
+#: darf beliebiger Leerraum stehen (nicht genau eines).
+PROMPT_RE = re.compile(rb"(?:^|[\r\n])[^\r\n]{0,120}?[>#]\s*$")
 PAGER_RE = re.compile(rb"(--\s?MORE\s?--|Press any key to continue)", re.IGNORECASE)
 ERROR_RE = re.compile(
     r"(invalid input|unknown command|incomplete|ambiguous|not supported|% *error)", re.IGNORECASE
@@ -175,27 +192,30 @@ class SwitchCli:
         """
         assert self._channel is not None
         deadline = time.monotonic() + timeout
-        collected = b""
+        # Zwei Puffer: das Rohe kommt herein, gesucht wird im Sichtbaren.
+        # Ohne diese Trennung entscheiden unsichtbare Steuerzeichen darueber,
+        # ob ein Prompt erkannt wird.
+        sichtbar = b""
         prompt_gesehen = False
         while time.monotonic() < deadline:
             if self._channel.recv_ready():
                 chunk = self._channel.recv(READ_CHUNK)
                 if not chunk:
                     break
-                collected += chunk
-                if PAGER_RE.search(collected[-200:]):
+                sichtbar = ANSI_RE.sub(b"", sichtbar + chunk)
+                if PAGER_RE.search(sichtbar[-300:]):
                     self._channel.send(" ")
-                    collected = PAGER_RE.sub(b"", collected)
+                    sichtbar = PAGER_RE.sub(b"", sichtbar)
                     continue
-                if PROMPT_RE.search(collected[-200:]):
+                if PROMPT_RE.search(sichtbar[-300:]):
                     prompt_gesehen = True
                     break
                 # Der Login-Banner mancher Geraete verlangt eine Bestaetigung.
-                if initial and collected.rstrip().endswith(b"continue"):
+                if initial and sichtbar.rstrip().endswith(b"continue"):
                     self._channel.send("\r\n")
             else:
                 time.sleep(0.05)
-        text = collected.decode("utf-8", errors="replace")
+        text = sichtbar.decode("utf-8", errors="replace")
         self.transcript.append(text)
         if not prompt_gesehen:
             letzte = " ".join(text.split())[-200:]