From 3612dd6e4549e7edecbb4e58a25a4334075bbcfe Mon Sep 17 00:00:00 2001 From: oliver Date: Thu, 3 Sep 2026 14:19:51 +0800 Subject: [PATCH] Fix log audit record bug --- netx_api/webcrt_channel.py | 32 ++++++++++- tests/test_webcrt_audit.py | 105 +++++++++++++++++++++++++++++++++++++ 2 files changed, 136 insertions(+), 1 deletion(-) diff --git a/netx_api/webcrt_channel.py b/netx_api/webcrt_channel.py index 3e977bd..3194464 100644 --- a/netx_api/webcrt_channel.py +++ b/netx_api/webcrt_channel.py @@ -670,6 +670,11 @@ def _stdout_has_inplace_edit(text: str) -> bool: Only inspects the last fragment (live input line). Older history-edit CSI still sitting in ``stdout_tail`` must not make bare Enter look like an in-place edit. Excludes the common ``---- More ----`` wipe (``ESC[16D``). + + Huawei rewrites the input row with CSI cursor moves (``ESC[D/C/P/@``). ZTE + instead erases-to-EOL (``ESC[K``) and reprints, so the ``[DCP@]`` scan alone + misses ZTE history recalls — also treat a live prompt+command fragment that + carries ``ESC[K`` as an in-place redraw. """ s = str(text or "")[-4000:] if not s: @@ -694,6 +699,11 @@ def _stdout_has_inplace_edit(text: str) -> bool: return True if cmd == "D" and n != 16: return True + # ZTE redraws the live input row with ESC[K (erase-to-EOL) + reprint / backspace + # rather than CSI cursor moves. A prompt+command fragment carrying ESC[K is an + # in-place edit (a trailing prompt cleanup alone renders as prompt-only above). + if "\x1b[K" in frag and _command_tail(rendered).strip(): + return True return False @@ -868,7 +878,18 @@ def render_pty_line(text: str) -> str: cursor = 0 i += 1 continue - if ch in ("\b", "\x7f"): + if ch == "\b": + # Backspace (0x08) in PTY output is cursor-left WITHOUT erasing — + # ZTE line redraws move the cursor via long backspace runs then + # overwrite/ESC[K to repaint. A destructive delete here mangles + # the recalled command text, so render_pty_line ends up with a + # fragment ("p") instead of the full "show ip interface brief". + if cursor > 0: + cursor -= 1 + i += 1 + continue + if ch == "\x7f": + # DEL (0x7f) deletes the cell to the left of the cursor. if cursor > 0: cursor -= 1 if cursor < len(cells): @@ -1066,6 +1087,15 @@ def pick_audit_command( return echo_n if len(et) == len(bt) and et != bt: return echo_n + # Echo longer with the xterm hint as a token-prefix of the fully + # redrawn device line: the xterm visible row lagged a partial prefix + # at Enter (ZTE reprints via ESC[K, so the snapshot can still read + # ``show ip`` while the device already shows ``show ip interface + # brief``). Trust the complete device echo. + if len(et) > len(bt) and all( + et[i].lower() == bt[i].lower() for i in range(len(bt)) + ): + return echo_n return base_n return base_n diff --git a/tests/test_webcrt_audit.py b/tests/test_webcrt_audit.py index bee4b7e..a1a548e 100644 --- a/tests/test_webcrt_audit.py +++ b/tests/test_webcrt_audit.py @@ -286,6 +286,111 @@ class ResolveAuditCommandsTests(unittest.TestCase): ) self.assertEqual(cmd, "[~r1]display ip interface brief") + def test_zte_history_insert_partial_hint_prefers_echo(self) -> None: + """ZTE redraws via ESC[K + backspace (no CSI cursor moves). + + Up-arrow recalls ``show interface brief``, the operator inserts ``ip`` so + the device redraws ``show ip interface brief``. The xterm visible row can + lag a partial ``show ip`` snapshot at Enter — the full device echo must win. + """ + from netx_api.webcrt_channel import _stdout_has_inplace_edit + + prompt = "AL5458-ACC-6120HS#" + # ZTE recall + ESC[K reprint of the fully-edited line. + stdout = ( + "\r\r\n" + prompt + "\x1b[Kshow interface brief" + + "\x08\x08\x08\x1b[Kshow interface brief" + + "\r" + prompt + "\x1b[Kshow ip interface brief" + ) + # ZTE redraw carries no [DCP@] CSI — without the ESC[K path this is False. + self.assertTrue(_stdout_has_inplace_edit(stdout)) + # Lagging partial xterm hint; device echo is the full line. + cmd = pick_audit_command( + "ip", + prompt + "show ip", + prompt_hint=prompt + "show interface brief", + stdout_tail=stdout, + source="stdin", + ) + self.assertEqual(cmd, prompt + "show ip interface brief") + # Same even when the frontend managed to send the full hint. + cmd2 = pick_audit_command( + "ip", + prompt + "show ip interface brief", + prompt_hint=prompt + "show interface brief", + stdout_tail=stdout, + source="stdin", + ) + self.assertEqual(cmd2, prompt + "show ip interface brief") + # Resolve path (Enter flush) must also pick the full line. + out = resolve_audit_commands( + ["ip"], + audit_line=prompt + "show ip", + prompt_hint=prompt + "show interface brief", + stdout_tail=stdout, + source="stdin", + ) + self.assertEqual(out, [prompt + "show ip interface brief"]) + + def test_zte_real_backspace_redraw_reconstructs_full_command(self) -> None: + """Real ZTE device redraws via backspace runs (not CSI cursor moves). + + Up-arrow recalls ``show interface brief``, left-arrows move the cursor + after ``show``, then ``ip`` is inserted. The device redraws using + long ``\\x08`` backspace sequences + ``ESC[K`` reprints — it never + echoes the full ``show ip interface brief`` as one contiguous string. + + ``render_pty_line`` must treat ``\\x08`` as non-destructive cursor-left + (standard terminal behaviour) so the recalled text survives the redraw + and the full edited line is reconstructed. + """ + import os + + from netx_api.webcrt_channel import ( + _live_input_line, + _stdout_has_inplace_edit, + extract_last_prompt_command, + render_pty_line, + ) + + fixture = os.path.join(os.path.dirname(__file__), "fixtures", + "zte_history_edit_stdout.txt") + with open(fixture, encoding="utf-8") as f: + stdout = f.read() + + prompt = "AL5458-ACC-6120HS#" + expected = prompt + "show ip interface brief" + + # render_pty_line must reconstruct the full edited command. + self.assertEqual(render_pty_line(stdout.split("\n")[-1]), expected) + # extract_last_prompt_command must find the full line. + self.assertEqual(extract_last_prompt_command(stdout), expected) + # Live input line must show the full edited command. + self.assertEqual(_live_input_line(stdout), expected) + # In-place edit detected (ESC[K redraw). + self.assertTrue(_stdout_has_inplace_edit(stdout)) + + # Even with a lagging partial xterm hint ("show ip"), the full device + # echo must win. + cmd = pick_audit_command( + "ip", + prompt + "show ip", + prompt_hint=prompt + "show interface brief", + stdout_tail=stdout, + source="stdin", + ) + self.assertEqual(cmd, expected) + + # Resolve path (Enter flush) must also pick the full line. + out = resolve_audit_commands( + ["ip"], + audit_line=prompt + "show ip", + prompt_hint=prompt + "show interface brief", + stdout_tail=stdout, + source="stdin", + ) + self.assertEqual(out, [expected]) + def test_history_midline_delete_prefers_device_echo(self) -> None: """Up-arrow then delete middle ``ip`` — must not keep stale longer audit_line.""" frag = (