diff --git a/netx_api/config_sync_runner.py b/netx_api/config_sync_runner.py index 8e91ad6..5248337 100644 --- a/netx_api/config_sync_runner.py +++ b/netx_api/config_sync_runner.py @@ -35,7 +35,7 @@ def _utcnow() -> datetime: def _format_error(exc: BaseException) -> str: - return format_cli_failure(exc, limit=1020) + return format_cli_failure(exc, limit=4000) def _pool_for_cycle(cycle_id: str, concurrency: int) -> ThreadPoolExecutor: @@ -124,16 +124,27 @@ def _collect_commands( *, conn_holder: dict[str, Any] | None = None, ) -> list[str]: + from .ne_netmiko import disable_target_paging + per_cmd = int(settings.ne_collect_read_timeout_sec or 120) session_timeout = per_cmd * max(1, len(commands)) + 60 log_buf = io.BytesIO() try: conn = open_netmiko_connection(creds, session_timeout=session_timeout, session_log=log_buf) except Exception as exc: - raise RuntimeError(format_cli_failure(exc, session_log_text(log_buf))) from exc + raise RuntimeError(format_cli_failure(exc, session_log_text(log_buf), limit=4000)) from exc if conn_holder is not None: conn_holder["conn"] = conn + conn_holder["session_log"] = log_buf try: + try: + disable_target_paging( + conn, + vendor=str(creds.get("vendor") or ""), + device_type=str(creds.get("device_type") or ""), + ) + except Exception: + pass outputs: list[str] = [] for command in commands: if conn_holder is not None and conn_holder.get("timed_out"): @@ -141,9 +152,13 @@ def _collect_commands( try: out = send_show_command(conn, command, read_timeout=per_cmd) except Exception as exc: - raise RuntimeError(format_cli_failure(exc, session_log_text(log_buf))) from exc + raise RuntimeError(format_cli_failure(exc, session_log_text(log_buf), limit=4000)) from exc outputs.append(str(out or "")) return outputs + except Exception as exc: + if isinstance(exc, TimeoutError): + raise RuntimeError(format_cli_failure(exc, session_log_text(log_buf), limit=4000)) from exc + raise finally: if conn_holder is not None: conn_holder.pop("conn", None) @@ -157,13 +172,18 @@ def _collect_with_timeout(creds: dict[str, Any], commands: list[str]) -> list[st cap = int(settings.ne_collect_run_timeout_cap_sec or 600) budget = min(cap, per_cmd * max(1, len(commands)) + 90) holder: dict[str, Any] = {} - return run_cli_with_timeout( - lambda: _collect_commands(creds, commands, conn_holder=holder), - timeout_sec=budget, - conn_holder=holder, - label="config_sync", - acquire_budget=True, - ) + try: + return run_cli_with_timeout( + lambda: _collect_commands(creds, commands, conn_holder=holder), + timeout_sec=budget, + conn_holder=holder, + label="config_sync", + acquire_budget=True, + ) + except TimeoutError as exc: + raise RuntimeError( + format_cli_failure(exc, session_log_text(holder.get("session_log")), limit=4000) + ) from exc def _history_keep(db) -> int: diff --git a/netx_api/ne_cli_errors.py b/netx_api/ne_cli_errors.py index ab321ff..076767f 100644 --- a/netx_api/ne_cli_errors.py +++ b/netx_api/ne_cli_errors.py @@ -89,7 +89,17 @@ def format_cli_failure(exc: BaseException | str, transcript: str = "", *, limit: else: msg = f"auth_rejected: {auth}" return msg[:limit] - return exc_text[:limit] + # Non-auth: keep a session-log tail so operators can see where the CLI stuck + # (More prompt, half-auth, wrong command echo, etc.). + tail = str(transcript or "").strip() + if not tail: + return exc_text[:limit] + sep = "\n--- session log ---\n" + budget = max(120, int(limit) - len(exc_text) - len(sep) - 8) + clipped = tail[-budget:] + if len(tail) > budget: + clipped = "…\n" + clipped + return f"{exc_text}{sep}{clipped}"[:limit] def session_log_text(session_log: Any) -> str: diff --git a/netx_api/ne_collect_runner.py b/netx_api/ne_collect_runner.py index 908455a..870fb0a 100644 --- a/netx_api/ne_collect_runner.py +++ b/netx_api/ne_collect_runner.py @@ -47,9 +47,12 @@ def shutdown_ne_collect_executor(*, wait: bool = False) -> None: def _format_run_error(exc: BaseException) -> str: head = f"{type(exc).__name__}: {exc}" + # RuntimeError from format_cli_failure already carries session log — keep it. + if isinstance(exc, RuntimeError) and "--- session log ---" in str(exc): + return str(exc)[:4000] tb = traceback.format_exc().strip() text = f"{head}\n{tb}" if tb else head - return text[:1020] + return text[:4000] def _safe_filename_part(text: str) -> str: @@ -64,23 +67,61 @@ def _collect_on_device( read_timeout_sec: int | None = None, conn_holder: dict[str, Any] | None = None, ) -> str: + import io + + from .ne_cli_errors import format_cli_failure, session_log_text + from .ne_netmiko import disable_target_paging + per_cmd = int(read_timeout_sec if read_timeout_sec is not None else (settings.ne_collect_read_timeout_sec or 120)) session_timeout = per_cmd * max(1, len(commands)) + 60 - conn = open_netmiko_connection(creds, session_timeout=session_timeout) - if conn_holder is not None: - conn_holder["conn"] = conn + log_buf = io.BytesIO() + chunks: list[str] = [] + conn = None try: + try: + conn = open_netmiko_connection( + creds, session_timeout=session_timeout, session_log=log_buf + ) + except Exception as exc: + raise RuntimeError(format_cli_failure(exc, session_log_text(log_buf), limit=4000)) from exc + if conn_holder is not None: + conn_holder["conn"] = conn + conn_holder["session_log"] = log_buf + # Belt-and-suspenders: hop/bastion nested CLIs and missed session_prep. + try: + disable_target_paging( + conn, + vendor=str(creds.get("vendor") or ""), + device_type=str(creds.get("device_type") or ""), + ) + except Exception: + _log.debug("collection paging disable failed", exc_info=True) prompt = str(conn.find_prompt() or "") - chunks: list[str] = [] for command in commands: if conn_holder is not None and conn_holder.get("timed_out"): raise TimeoutError("collection_aborted") ts = datetime.now().isoformat(timespec="seconds") chunks.append(f'>>> [{ts}] {{"String":"{command}", "Match":"{prompt}", "Timeout":0}}\n') - out = send_show_command(conn, command, read_timeout=per_cmd) + try: + out = send_show_command(conn, command, read_timeout=per_cmd) + except Exception as exc: + partial = "".join(chunks) + transcript = session_log_text(log_buf) or partial + raise RuntimeError( + format_cli_failure(exc, transcript, limit=4000) + ) from exc chunks.append(str(out or "")) chunks.append("\n") return "".join(chunks) + except Exception as exc: + # Surface echo for mid-command Netmiko failures not already wrapped. + if isinstance(exc, RuntimeError) and "--- session log ---" in str(exc): + raise + partial = "".join(chunks) + transcript = session_log_text(log_buf) or partial + if transcript: + raise RuntimeError(format_cli_failure(exc, transcript, limit=4000)) from exc + raise finally: if conn_holder is not None: conn_holder.pop("conn", None) @@ -94,20 +135,25 @@ def _collect_with_timeout( read_timeout_sec: int | None = None, ) -> str: from .cli_timeout import run_cli_with_timeout + from .ne_cli_errors import format_cli_failure, session_log_text per_cmd = int(read_timeout_sec if read_timeout_sec is not None else (settings.ne_collect_read_timeout_sec or 120)) cap = int(settings.ne_collect_run_timeout_cap_sec or 600) budget = min(cap, per_cmd * max(1, len(commands)) + 90) holder: dict[str, Any] = {} - return run_cli_with_timeout( - lambda: _collect_on_device( - creds, commands, read_timeout_sec=per_cmd, conn_holder=holder - ), - timeout_sec=budget, - conn_holder=holder, - label="collection", - acquire_budget=True, - ) + try: + return run_cli_with_timeout( + lambda: _collect_on_device( + creds, commands, read_timeout_sec=per_cmd, conn_holder=holder + ), + timeout_sec=budget, + conn_holder=holder, + label="collection", + acquire_budget=True, + ) + except TimeoutError as exc: + transcript = session_log_text(holder.get("session_log")) + raise RuntimeError(format_cli_failure(exc, transcript, limit=4000)) from exc def _update_run(run_id: str, **fields: Any) -> None: diff --git a/netx_api/ne_exec.py b/netx_api/ne_exec.py index e1b5667..62206c7 100644 --- a/netx_api/ne_exec.py +++ b/netx_api/ne_exec.py @@ -81,13 +81,16 @@ def execute_managed_ne_commands( try: output = _collect_on_device(creds, cmds, read_timeout_sec=read_timeout) except Exception as exc: + detail = str(exc)[:4000] return { "ok": False, "device": device, "commands": cmds, "read_timeout_sec": read_timeout, "error": type(exc).__name__, - "detail": str(exc)[:2000], + "detail": detail, + # Preserve whatever echo was captured so LLDP discover / ops can show it. + "output": detail, } if len(output) > _EXEC_MAX_OUTPUT: diff --git a/netx_api/ne_netmiko.py b/netx_api/ne_netmiko.py index 4c51c2a..f1acdd0 100644 --- a/netx_api/ne_netmiko.py +++ b/netx_api/ne_netmiko.py @@ -2,6 +2,7 @@ from __future__ import annotations +import time from typing import Any @@ -37,6 +38,63 @@ def is_zte_device_type(device_type: str) -> bool: return "zte" in str(device_type or "").strip().lower() +def is_huawei_device_type(device_type: str) -> bool: + return "huawei" in str(device_type or "").strip().lower() + + +def paging_disable_commands(*, vendor: str = "", device_type: str = "") -> list[str]: + """Vendor CLI commands to disable --More-- paging (collection / LLDP / sync).""" + dt = str(device_type or "").strip().lower() + v = str(vendor or "").strip().lower() + blob = f"{dt} {v}" + if "huawei" in blob or "华为" in v: + return ["screen-length 0 temporary"] + if "hp_comware" in blob or "h3c" in blob or "comware" in blob: + return ["screen-length disable"] + if "juniper" in blob or "junos" in blob: + return ["set cli screen-length 0"] + if "nokia" in blob or "alcatel_sros" in blob or "sros" in blob: + return ["environment no more"] + if "alcatel_aos" in blob: + return ["terminal length 0"] + # Cisco / ZTE / generic SSH CLIs + return ["terminal length 0"] + + +def disable_target_paging( + conn: Any, + *, + vendor: str = "", + device_type: str = "", +) -> str: + """Send paging-off on the *current* CLI (critical after CLI hop / bastion jump). + + Nested stelnet/telnet lands on the target without Netmiko ``session_preparation``, + so --More-- stays enabled and long ``display/show lldp`` / config dumps time out. + Uses timing reads so a wrong hop ``base_prompt`` does not block. + """ + cmds = paging_disable_commands(vendor=vendor, device_type=device_type) + chunks: list[str] = [] + for cmd in cmds: + try: + out = conn.send_command_timing(cmd, read_timeout=15) + chunks.append(str(out or "")) + continue + except Exception: + pass + try: + ret = getattr(conn, "RETURN", None) or "\n" + conn.write_channel(str(cmd) + ret) + time.sleep(0.35) + if hasattr(conn, "read_channel"): + part = conn.read_channel() + if part: + chunks.append(str(part)) + except Exception: + pass + return "".join(chunks) + + def send_show_command(conn: Any, command: str, *, read_timeout: int = 120) -> str: """Send a show/display command via ``send_command`` (wait for device prompt). diff --git a/netx_api/ne_session_connect.py b/netx_api/ne_session_connect.py index 5b3b8c2..932d164 100644 --- a/netx_api/ne_session_connect.py +++ b/netx_api/ne_session_connect.py @@ -360,14 +360,60 @@ def _zte_collection_driver_class(base_cls: type) -> type: return _ZteCollectionSession +def _huawei_collection_driver_class(base_cls: type) -> type: + """Huawei VRP collection: ensure screen-length is off before long display cmds.""" + + class _HuaweiCollectionSession(base_cls): # type: ignore[misc,valid-type] + def session_preparation(self) -> None: + prompt_pat = r"(?:<[^>\r\n]{1,64}>|\[[^\]\r\n]{1,64}\])" + try: + self._test_channel_read(pattern=prompt_pat) + except Exception: + try: + self.write_channel(self.RETURN) + except Exception: + pass + try: + self._test_channel_read(pattern=prompt_pat) + except Exception: + # Fall back to stock prompt family. + self._test_channel_read(pattern=r"[>\]]") + try: + self.set_base_prompt() + except Exception: + pass + try: + self.disable_paging( + command="screen-length 0 temporary", + cmd_verify=False, + pattern=prompt_pat, + ) + except Exception: + try: + self.send_command_timing("screen-length 0 temporary", read_timeout=15) + except Exception: + pass + try: + self.clear_buffer() + except Exception: + pass + + _HuaweiCollectionSession.__name__ = ( + f"HuaweiCollection{getattr(base_cls, '__name__', 'Netmiko')}" + ) + return _HuaweiCollectionSession + + def _collection_driver_class(device_type: str, base_cls: type) -> type: """Vendor-specific collection session prep (non-interactive CLI / LLDP / sync).""" - from .ne_netmiko import is_cisco_ios_device_type, is_zte_device_type + from .ne_netmiko import is_cisco_ios_device_type, is_huawei_device_type, is_zte_device_type if is_cisco_ios_device_type(device_type): return _cisco_ios_collection_driver_class(base_cls) if is_zte_device_type(device_type): return _zte_collection_driver_class(base_cls) + if is_huawei_device_type(device_type): + return _huawei_collection_driver_class(base_cls) return base_cls @@ -910,6 +956,20 @@ def _connect_via_cli_hop( progress_cb=progress_cb, emit_raw=emit_raw, ) + # Nested target CLI never got Netmiko session_preparation — turn off paging now. + if not interactive: + from .ne_netmiko import disable_target_paging + + try: + paging_out = disable_target_paging( + conn, + vendor=str(creds.get("vendor") or ""), + device_type=str(creds.get("device_type") or ""), + ) + if paging_out and emit_raw: + _emit_progress(progress_cb, paging_out) + except Exception: + _log.debug("cli hop target paging disable failed", exc_info=True) _attach_cli_hop_guard( conn, hop_prompt=hop_prompt, @@ -1060,6 +1120,21 @@ def _connect_via_bastion( except Exception: pass raise + # After secondary auth (or bastion-managed landing), ensure target paging is off. + # session_preparation may have run too early against proxy banners. + if not interactive: + from .ne_netmiko import disable_target_paging + + try: + paging_out = disable_target_paging( + conn, + vendor=str(creds.get("vendor") or ""), + device_type=str(creds.get("device_type") or ""), + ) + if paging_out and not teed: + _emit_progress(progress_cb, paging_out) + except Exception: + _log.debug("bastion target paging disable failed", exc_info=True) return conn diff --git a/netx_api/port_traffic_runner.py b/netx_api/port_traffic_runner.py index 16e1d06..d13315b 100644 --- a/netx_api/port_traffic_runner.py +++ b/netx_api/port_traffic_runner.py @@ -260,10 +260,20 @@ def _sample_targets_shared_session(device_id: str, target_ids: list[str]) -> tup holder: dict[str, Any] = {} def _run_session() -> tuple[int, str]: + from .ne_netmiko import disable_target_paging + conn = open_netmiko_connection(creds, session_timeout=budget) holder["conn"] = conn local_errors = 0 try: + try: + disable_target_paging( + conn, + vendor=str(creds.get("vendor") or ""), + device_type=str(creds.get("device_type") or ""), + ) + except Exception: + pass for tid, ifname in ifaces: if holder.get("timed_out"): raise TimeoutError("port_traffic_aborted") diff --git a/netx_api/topology_discover_scan.py b/netx_api/topology_discover_scan.py index 8d8d624..ae8f428 100644 --- a/netx_api/topology_discover_scan.py +++ b/netx_api/topology_discover_scan.py @@ -134,12 +134,16 @@ def _discover_one_target( "error": detail, } if not exec_out.get("ok"): - err = str(exec_out.get("error") or exec_out.get("detail") or "exec_failed")[:500] + raw = str(exec_out.get("output") or exec_out.get("detail") or "") + err = str(exec_out.get("detail") or exec_out.get("error") or "exec_failed")[:4000] return { **base, "ok": False, "command": cmd, + "parser_key": pkey, + "parser_stub": bool(is_stub), "error": err, + "raw_preview": _raw_preview(raw or err), } raw = str(exec_out.get("output") or "") diff --git a/tests/test_cli_hop_return.py b/tests/test_cli_hop_return.py index 4d1eb1e..413252c 100644 --- a/tests/test_cli_hop_return.py +++ b/tests/test_cli_hop_return.py @@ -50,22 +50,25 @@ class CliHopReturnDetectionTests(unittest.TestCase): self.assertFalse(should_close_cli_hop_session(text, "", seen_other_prompt=False)) self.assertTrue(should_close_cli_hop_session(text, "", seen_other_prompt=True)) + @patch("netx_api.ne_netmiko.disable_target_paging") @patch("netx_api.ne_session_connect._interactive_target_auth") @patch("netx_api.ne_session_connect._read_channel") - @patch("netx_api.ne_session_connect.ConnectHandler") + @patch("netx_api.ne_session_connect._build_netmiko_connection") def test_connect_attaches_cli_hop_guard( self, - mock_ch: MagicMock, + mock_build: MagicMock, mock_read: MagicMock, mock_auth: MagicMock, + mock_paging: MagicMock, ) -> None: from netx_api.ne_session_factory import _connect_via_cli_hop conn = MagicMock() conn.remote_conn = MagicMock() - mock_ch.return_value = conn + mock_build.return_value = conn mock_read.side_effect = ["\n", ""] mock_auth.return_value = None + mock_paging.return_value = "" creds = { "hop_host": "10.0.0.1", "hop_username": "admin", @@ -79,6 +82,8 @@ class CliHopReturnDetectionTests(unittest.TestCase): "password": "target-pass", "ip_address": "10.0.0.2", "port": 22, + "vendor": "Huawei", + "device_type": "huawei", } out = _connect_via_cli_hop(creds, cols=120, rows=40) self.assertIs(out, conn) @@ -90,23 +95,27 @@ class CliHopReturnDetectionTests(unittest.TestCase): self.assertTrue(guard["enabled"]) self.assertEqual(guard["hop_prompt"], "") self.assertEqual(guard["hop_vendor"], "huawei") + mock_paging.assert_called() + @patch("netx_api.ne_netmiko.disable_target_paging") @patch("netx_api.ne_session_connect._interactive_target_auth") @patch("netx_api.ne_session_connect._read_channel") - @patch("netx_api.ne_session_connect.ConnectHandler") + @patch("netx_api.ne_session_connect._build_netmiko_connection") def test_huawei_skips_system_view_by_default( self, - mock_ch: MagicMock, + mock_build: MagicMock, mock_read: MagicMock, mock_auth: MagicMock, + mock_paging: MagicMock, ) -> None: from netx_api.ne_session_factory import _connect_via_cli_hop conn = MagicMock() conn.remote_conn = MagicMock() - mock_ch.return_value = conn + mock_build.return_value = conn mock_read.side_effect = ["\n", ""] mock_auth.return_value = None + mock_paging.return_value = "" creds = { "hop_host": "10.0.0.1", "hop_username": "admin", @@ -121,29 +130,34 @@ class CliHopReturnDetectionTests(unittest.TestCase): "password": "target-pass", "ip_address": "10.0.0.2", "port": 22, + "vendor": "Huawei", + "device_type": "huawei", } _connect_via_cli_hop(creds, cols=80, rows=24) wrote = "".join(str(c.args[0]) for c in conn.write_channel.call_args_list) self.assertNotIn("system-view", wrote) self.assertIn("stelnet", wrote) + @patch("netx_api.ne_netmiko.disable_target_paging") @patch("netx_api.ne_session_connect._interactive_target_auth") @patch("netx_api.ne_session_connect._read_channel") - @patch("netx_api.ne_session_connect.ConnectHandler") + @patch("netx_api.ne_session_connect._build_netmiko_connection") def test_huawei_enters_system_view_when_flag_set( self, - mock_ch: MagicMock, + mock_build: MagicMock, mock_read: MagicMock, mock_auth: MagicMock, + mock_paging: MagicMock, ) -> None: from netx_api.ne_session_factory import _connect_via_cli_hop conn = MagicMock() conn.remote_conn = MagicMock() - mock_ch.return_value = conn + mock_build.return_value = conn # First read: user-view prompt; later reads: system-view prompt after system-view. mock_read.side_effect = ["\n", "[~HOP]\n", "[~HOP]\n", ""] mock_auth.return_value = None + mock_paging.return_value = "" creds = { "hop_host": "10.0.0.1", "hop_username": "admin", @@ -158,6 +172,8 @@ class CliHopReturnDetectionTests(unittest.TestCase): "password": "target-pass", "ip_address": "10.0.0.2", "port": 22, + "vendor": "Huawei", + "device_type": "huawei", } _connect_via_cli_hop(creds, cols=80, rows=24) wrote = "".join(str(c.args[0]) for c in conn.write_channel.call_args_list) @@ -168,6 +184,7 @@ class CliHopReturnDetectionTests(unittest.TestCase): guard = get_cli_hop_guard(conn) assert guard is not None self.assertEqual(guard["hop_prompt"], "[~HOP]") + mock_paging.assert_called() if __name__ == "__main__": diff --git a/tests/test_ne_cli_errors.py b/tests/test_ne_cli_errors.py index c308284..9243032 100644 --- a/tests/test_ne_cli_errors.py +++ b/tests/test_ne_cli_errors.py @@ -20,11 +20,13 @@ class CliAuthClassifyTests(unittest.TestCase): self.assertIn("Permission denied", msg) self.assertIn("prompt_timeout", msg) - def test_plain_timeout_unchanged(self): + def test_plain_timeout_includes_session_log(self): exc = RuntimeError("ReadTimeout: Pattern not detected: '[>#]' in output.") - msg = format_cli_failure(exc, "show running-config\n...still dumping...\n") + msg = format_cli_failure(exc, "show running-config\n...still dumping...\n---- More ----\n") self.assertIn("ReadTimeout", msg) self.assertFalse(msg.startswith("auth_rejected:")) + self.assertIn("session log", msg) + self.assertIn("More", msg) def test_authentication_exception(self): class AuthenticationException(Exception): @@ -53,6 +55,8 @@ class CliAuthClassifyTests(unittest.TestCase): self.assertIsNone(find_auth_failure_snippet(text)) msg = format_cli_failure("ReadTimeout: Pattern not detected", text) self.assertFalse(msg.startswith("auth_rejected:")) + # Still attach transcript for diagnostics (Huawei banner is not auth_rejected). + self.assertIn("session log", msg) def test_real_auth_failure_still_detected_near_banner(self): text = ( diff --git a/tests/test_ne_session_interactive.py b/tests/test_ne_session_interactive.py index 0638338..f31296f 100644 --- a/tests/test_ne_session_interactive.py +++ b/tests/test_ne_session_interactive.py @@ -57,6 +57,12 @@ class ZteCollectionPrepTests(unittest.TestCase): self.assertIsNot(wrapped, base) self.assertTrue(wrapped.__name__.startswith("ZteCollection")) + def test_collection_driver_selects_huawei_wrapper(self) -> None: + base = _netmiko_driver_class("huawei") + wrapped = _collection_driver_class("huawei", base) + self.assertIsNot(wrapped, base) + self.assertTrue(wrapped.__name__.startswith("HuaweiCollection")) + def test_prompt_timeout_sends_return_then_retries(self) -> None: base = _netmiko_driver_class("zte_zxros_ssh") cls = _zte_collection_driver_class(base) diff --git a/tests/test_paging_disable.py b/tests/test_paging_disable.py new file mode 100644 index 0000000..d4030ec --- /dev/null +++ b/tests/test_paging_disable.py @@ -0,0 +1,41 @@ +"""Tests for vendor paging-disable helpers used by LLDP / collect / hop.""" + +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock + +from netx_api.ne_netmiko import disable_target_paging, paging_disable_commands + + +class PagingDisableTests(unittest.TestCase): + def test_commands_by_vendor(self) -> None: + self.assertEqual( + paging_disable_commands(vendor="Huawei", device_type="huawei"), + ["screen-length 0 temporary"], + ) + self.assertEqual( + paging_disable_commands(vendor="ZTE", device_type="zte_zxros"), + ["terminal length 0"], + ) + self.assertEqual( + paging_disable_commands(vendor="Cisco", device_type="cisco_ios"), + ["terminal length 0"], + ) + self.assertEqual( + paging_disable_commands(vendor="H3C", device_type="hp_comware"), + ["screen-length disable"], + ) + + def test_disable_target_paging_uses_timing(self) -> None: + conn = MagicMock() + conn.send_command_timing.return_value = "ok\n" + out = disable_target_paging(conn, vendor="Huawei", device_type="huawei") + conn.send_command_timing.assert_called_once_with( + "screen-length 0 temporary", read_timeout=15 + ) + self.assertIn("ok", out) + + +if __name__ == "__main__": + unittest.main()