From a36de061cd5ff145755e5e451fc069b138902378 Mon Sep 17 00:00:00 2001 From: oliver Date: Fri, 16 Oct 2026 19:21:42 +0800 Subject: [PATCH] Cisco IOS collection: disable paging after prompt drain. Send terminal length 0 ourselves and wait for a full prompt line, so the echo line cannot end the read and long show output no longer stalls on --More--. Co-Authored-By: Claude Opus 5.5 --- netx_api/ne_session_connect.py | 14 ++++++++------ tests/test_ne_session_interactive.py | 19 ++++++++++++------- 2 files changed, 20 insertions(+), 13 deletions(-) diff --git a/netx_api/ne_session_connect.py b/netx_api/ne_session_connect.py index 601e669..ea34998 100644 --- a/netx_api/ne_session_connect.py +++ b/netx_api/ne_session_connect.py @@ -309,18 +309,20 @@ def _cisco_ios_collection_driver_class(base_cls: type) -> type: class _CiscoIosCollectionSession(base_cls): # type: ignore[misc,valid-type] def session_preparation(self) -> None: # Stock Netmiko width/paging on IOSv+CR races leftover prompts - # (``terexit`` / ``terminal lengexit``) or blocks forever waiting for - # setup echo once VTY pressure builds. The interactive WebCRT path - # skips these commands and still returns complete show output; MCP - # show/display bodies are short enough that --More-- is not required. - # Keep hostname discovery so send_show_command can use a full-line - # prompt expect without auto_find_prompt. + # (``terexit`` / ``terminal lengexit``). Drain first, then disable + # paging ourselves and wait for a full prompt line: the echo line + # ``R2#terminal length 0`` cannot match, so long show output never + # stalls on --More--. from .ne_netmiko import drain_read_channel initial_pattern = r"(?m:^[^\r\n#>]+[>#][ \t]*$)" self._test_channel_read(pattern=initial_pattern) self.set_base_prompt(pattern=initial_pattern) drain_read_channel(self, idle_reads=8, pause_sec=0.1) + self.write_channel("terminal length 0" + self.RETURN) + prompt_line = rf"(?m:^{re.escape(self.base_prompt)}[>#][ \t]*$)" + self.read_until_pattern(pattern=prompt_line, read_timeout=20) + drain_read_channel(self, idle_reads=8, pause_sec=0.1) _CiscoIosCollectionSession.__name__ = ( f"CiscoIosCollection{getattr(base_cls, '__name__', 'Netmiko')}" diff --git a/tests/test_ne_session_interactive.py b/tests/test_ne_session_interactive.py index fabe6c4..6f8ffac 100644 --- a/tests/test_ne_session_interactive.py +++ b/tests/test_ne_session_interactive.py @@ -82,24 +82,29 @@ class CiscoCollectionPrepTests(unittest.TestCase): return sess @patch("netx_api.ne_netmiko.drain_read_channel") - def test_setup_only_establishes_hostname_prompt(self, drain: MagicMock): + def test_setup_drains_then_disables_paging_on_full_prompt_line(self, drain: MagicMock): sess = self.session() order: list[str] = [] drain.side_effect = lambda *_a, **_k: order.append("drain") sess.set_base_prompt.side_effect = lambda **kw: order.append("prompt") + sess.write_channel.side_effect = lambda data: order.append(f"write:{data}") + sess.read_until_pattern.side_effect = lambda **kw: order.append("read") sess.session_preparation() - self.assertEqual(order, ["prompt", "drain"]) + self.assertEqual(order, ["prompt", "drain", "write:terminal length 0\r", "read", "drain"]) sess.set_base_prompt.assert_called_once() initial = sess._test_channel_read.call_args.kwargs["pattern"] self.assertIsNotNone(re.search(initial, "R2.lab#")) self.assertIsNone(re.search(initial, "R2.lab#terminal length 0")) channel = SimpleNamespace(read_timeout_override=None, _read_buffer="", read_channel=lambda: "R2.lab#") self.assertEqual(BaseConnection.read_until_pattern(channel, pattern=initial), "R2.lab#") + prompt_line = sess.read_until_pattern.call_args.kwargs["pattern"] + self.assertIsNone(re.search(prompt_line, "R2.lab#terminal length 0\r\n")) + self.assertIsNone(re.search(prompt_line, "R2Xlab#")) + self.assertIsNotNone(re.search(prompt_line, "R2.lab#terminal length 0\r\nR2.lab#")) sess.set_terminal_width.assert_not_called() sess.disable_paging.assert_not_called() - sess.write_channel.assert_not_called() sess.send_command_timing.assert_not_called() - drain.assert_called_once() + self.assertEqual(drain.call_count, 2) self.assertEqual(drain.call_args.kwargs.get("idle_reads"), 8) @patch("netx_api.ne_netmiko.drain_read_channel") @@ -113,13 +118,13 @@ class CiscoCollectionPrepTests(unittest.TestCase): sess.send_command_timing.assert_not_called() @patch("netx_api.ne_netmiko.drain_read_channel") - def test_setup_never_sends_terminal_commands(self, drain: MagicMock): + def test_setup_sends_only_terminal_length_zero(self, drain: MagicMock): sess = self.session() sess.session_preparation() - sess.write_channel.assert_not_called() + sess.write_channel.assert_called_once_with("terminal length 0\r") sess.set_terminal_width.assert_not_called() sess.disable_paging.assert_not_called() - drain.assert_called_once_with(sess, idle_reads=8, pause_sec=0.1) + drain.assert_called_with(sess, idle_reads=8, pause_sec=0.1) class ZteCollectionPrepTests(unittest.TestCase):