hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-09 04:20:45 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions 2

Disable target paging after hop login and keep CLI echo on timeouts.

Browse source 
LLDP discover and config collect were timing out on --More--; send vendor paging-off after nested/bastion auth, and attach session-log tails so failures show where the CLI stuck.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-08-29 12:30:33 +08:00
parent 3adf8d8b75
commit fbf73cbaaf
12 changed files with 334 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

35
tests/test_cli_hop_return.py
View file

@ -50,22 +50,25 @@ class CliHopReturnDetectionTests(unittest.TestCase):
self.assertFalse(should_close_cli_hop_session(text, "<HOP>", seen_other_prompt=False))
self.assertTrue(should_close_cli_hop_session(text, "<HOP>", 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 = ["<HOP>\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"], "<HOP>")
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 = ["<HOP>\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 = ["<HOP>\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__":

8
tests/test_ne_cli_errors.py
View file

@ -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 = (

6
tests/test_ne_session_interactive.py
View file

@ -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)

41
tests/test_paging_disable.py Normal file
View file

@ -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<HUAWEI>"
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()