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

Add explicit Huawei hop system-view option and startup column safety-net.

Browse source 
Some hops only accept stelnet in system-view; make it a saved yes/no setting (default off) instead of auto-retry, and always ALTER the new column on API boot when Alembic skips legacy DDL.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-08-28 23:58:05 +08:00
parent 0a550abcb0
commit b36174bfa4
20 changed files with 242 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

78
tests/test_cli_hop_return.py
View file

@ -91,6 +91,84 @@ class CliHopReturnDetectionTests(unittest.TestCase):
self.assertEqual(guard["hop_prompt"], "<HOP>")
self.assertEqual(guard["hop_vendor"], "huawei")
@patch("netx_api.ne_session_connect._interactive_target_auth")
@patch("netx_api.ne_session_connect._read_channel")
@patch("netx_api.ne_session_connect.ConnectHandler")
def test_huawei_skips_system_view_by_default(
self,
mock_ch: MagicMock,
mock_read: MagicMock,
mock_auth: 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_read.side_effect = ["<HOP>\n", ""]
mock_auth.return_value = None
creds = {
"hop_host": "10.0.0.1",
"hop_username": "admin",
"hop_password": "hop-pass",
"hop_protocol": "ssh",
"hop_vendor": "huawei",
"hop_port": 22,
"hop_vrf": "",
"hop_command_template": "stelnet {target_ip}",
"hop_enter_system_view": False,
"username": "target",
"password": "target-pass",
"ip_address": "10.0.0.2",
"port": 22,
}
_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_session_connect._interactive_target_auth")
@patch("netx_api.ne_session_connect._read_channel")
@patch("netx_api.ne_session_connect.ConnectHandler")
def test_huawei_enters_system_view_when_flag_set(
self,
mock_ch: MagicMock,
mock_read: MagicMock,
mock_auth: MagicMock,
) -> None:
from netx_api.ne_session_factory import _connect_via_cli_hop
conn = MagicMock()
conn.remote_conn = MagicMock()
mock_ch.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
creds = {
"hop_host": "10.0.0.1",
"hop_username": "admin",
"hop_password": "hop-pass",
"hop_protocol": "ssh",
"hop_vendor": "huawei",
"hop_port": 22,
"hop_vrf": "",
"hop_command_template": "stelnet {target_ip}",
"hop_enter_system_view": True,
"username": "target",
"password": "target-pass",
"ip_address": "10.0.0.2",
"port": 22,
}
_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.assertIn("system-view", wrote)
self.assertIn("stelnet", wrote)
# system-view must come before stelnet
self.assertLess(wrote.find("system-view"), wrote.find("stelnet"))
guard = get_cli_hop_guard(conn)
assert guard is not None
self.assertEqual(guard["hop_prompt"], "[~HOP]")
if __name__ == "__main__":
unittest.main()

32
tests/test_schema_patches.py
View file

@ -13,6 +13,7 @@ import netx_api.models # noqa: F401
from netx_api.schema_patches import (
apply_auth_schema_patches,
apply_domain_schema_patches,
apply_hop_schema_safety_net,
apply_topology_schema_safety_net,
)
@ -60,6 +61,37 @@ class SchemaPatchesTests(unittest.TestCase):
self.assertIn("a_ifname", cols)
self.assertIn("z_ifname", cols)
def test_hop_safety_net_adds_enter_system_view(self) -> None:
with self.engine.begin() as conn:
conn.execute(text("DROP TABLE IF EXISTS managed_ne"))
conn.execute(
text(
"""
CREATE TABLE managed_ne (
id VARCHAR(64) PRIMARY KEY,
hop_enabled BOOLEAN DEFAULT 0
)
"""
)
)
conn.execute(text("DROP TABLE IF EXISTS cli_connect_profile"))
conn.execute(
text(
"""
CREATE TABLE cli_connect_profile (
id VARCHAR(64) PRIMARY KEY,
hop_enabled BOOLEAN DEFAULT 0
)
"""
)
)
apply_hop_schema_safety_net(conn)
apply_hop_schema_safety_net(conn)
ne_cols = {c["name"] for c in inspect(self.engine).get_columns("managed_ne")}
profile_cols = {c["name"] for c in inspect(self.engine).get_columns("cli_connect_profile")}
self.assertIn("hop_enter_system_view", ne_cols)
self.assertIn("hop_enter_system_view", profile_cols)
def test_domain_patches_do_not_raise(self) -> None:
with self.engine.begin() as conn:
apply_domain_schema_patches(conn)