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

fix(webcrt): resize hop PTY before nested CLI jump

Browse source 
Nested stelnet/telnet captures hop TTY width at start; Netmiko's default 511 cols made mid-line edit redraws wrap and garble in ~80-col WebCRT.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
hansjone 2026-07-30 02:34:25 +00:00
parent b690914f18
commit b540ea21c1
6 changed files with 88 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

5
tests/test_cli_hop_return.py
View file

@ -62,6 +62,7 @@ class CliHopReturnDetectionTests(unittest.TestCase):
from netx_api.ne_session_factory import _connect_via_cli_hop
conn = MagicMock()
conn.remote_conn = MagicMock()
mock_ch.return_value = conn
mock_read.side_effect = ["<HOP>\n", ""]
mock_auth.return_value = None
@ -79,8 +80,10 @@ class CliHopReturnDetectionTests(unittest.TestCase):
"ip_address": "10.0.0.2",
"port": 22,
}
out = _connect_via_cli_hop(creds)
out = _connect_via_cli_hop(creds, cols=120, rows=40)
self.assertIs(out, conn)
# Nested stelnet must see WebCRT geometry, not Netmiko's default 511x1000.
conn.remote_conn.resize_pty.assert_called_with(width=120, height=40)
guard = get_cli_hop_guard(conn)
self.assertIsNotNone(guard)
assert guard is not None

42
tests/test_webcrt.py
View file

@ -122,6 +122,7 @@ class WebcrtServiceTests(unittest.TestCase):
with patch.object(svc.settings, "webcrt_max_sessions", 1):
out = svc.create_session(db, ne_id="ne-a", cols=80, rows=24, client="test")
self.assertIn("session_id", out)
self.assertFalse(out.get("cli_hop"))
with self.assertRaises(HTTPException) as ctx:
svc.create_session(db, ne_id="ne-a", cols=80, rows=24, client="test")
self.assertEqual(ctx.exception.status_code, 429)
@ -179,6 +180,7 @@ class WebcrtServiceTests(unittest.TestCase):
self.assertIn("session_log", mock_open.call_args.kwargs)
fake.remote_conn.resize_pty.assert_called()
self.assertEqual(out["ne_id"], "ne-hop")
self.assertFalse(out.get("cli_hop")) # bastion hop is not vendor CLI hop guard
sess = svc.get_session(out["session_id"])
assert sess is not None
boot = sess.bootstrap_output.decode("utf-8", errors="replace")
@ -190,6 +192,46 @@ class WebcrtServiceTests(unittest.TestCase):
self.assertEqual(fake.written[len(before) :], ["\n"])
svc.close_session(out["session_id"], reason="test")
@patch.object(svc, "_audit")
@patch.object(svc, "get_cli_hop_guard")
@patch.object(svc, "open_netmiko_connection")
@patch("netx_api.cli_resolve.resolve_cli_target")
def test_create_session_reports_cli_hop(
self,
mock_resolve: MagicMock,
mock_open: MagicMock,
mock_guard: MagicMock,
_mock_audit: MagicMock,
) -> None:
mock_resolve.return_value = (
{
"username": "u",
"password": "p",
"hop_enabled": True,
"hop_vendor": "huawei",
"ip_address": "10.0.0.3",
"protocol": "ssh",
},
{
"id": "ne-cli-hop",
"name": "C",
"ip_address": "10.0.0.3",
"protocol": "ssh",
"source": "managed",
},
)
mock_open.side_effect = lambda *a, **k: _FakeConn()
mock_guard.return_value = {"hop_prompt": "<HOP>", "hop_vendor": "huawei", "hop_host": "1.1.1.1"}
out = svc.create_session(MagicMock(), ne_id="ne-cli-hop", cols=100, rows=30, client="test")
self.assertTrue(out.get("cli_hop"))
self.assertEqual(mock_open.call_args.kwargs.get("cols"), 100)
self.assertEqual(mock_open.call_args.kwargs.get("rows"), 30)
sess = svc.get_session(out["session_id"])
assert sess is not None
self.assertTrue(sess.cli_hop_guard)
self.assertEqual(sess.cli_hop_prompt, "<HOP>")
svc.close_session(out["session_id"], reason="test")
@patch.object(svc, "_audit")
@patch.object(svc, "open_netmiko_connection")
@patch("netx_api.cli_resolve.resolve_cli_target")