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

Skip CLI startup when device credentials are incomplete.

Browse source 
Reject non-interactive CLI work early across exec, LLDP, config sync, collection, traffic, and connect-test flows so tasks record a clear reason instead of launching a doomed session.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-08-11 15:11:59 +08:00
parent dcdf4147a6
commit b6a579ec02
12 changed files with 317 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

119
tests/test_cli_creds.py Normal file
View file

@ -0,0 +1,119 @@
from __future__ import annotations
import unittest
from netx_api.cli_creds import (
REASON_HOP_INCOMPLETE,
REASON_NO_PASSWORD,
REASON_USERNAME_REQUIRED,
cli_creds_ready,
cli_creds_skip_reason,
)
class CliCredsTests(unittest.TestCase):
def test_direct_ssh_requires_password(self) -> None:
creds = {
"ip_address": "10.0.0.1",
"protocol": "ssh",
"username": "admin",
"password": "",
"hop_enabled": False,
}
ready, reason = cli_creds_ready(creds, interactive=False)
self.assertFalse(ready)
self.assertEqual(reason, REASON_NO_PASSWORD)
self.assertEqual(cli_creds_skip_reason(creds), REASON_NO_PASSWORD)
def test_direct_ssh_ready(self) -> None:
creds = {
"ip_address": "10.0.0.1",
"protocol": "ssh",
"username": "admin",
"password": "secret",
"hop_enabled": False,
}
ready, reason = cli_creds_ready(creds, interactive=False)
self.assertTrue(ready)
self.assertEqual(reason, "")
self.assertIsNone(cli_creds_skip_reason(creds))
def test_bastion_managed_allows_empty_target_password(self) -> None:
creds = {
"ip_address": "10.0.0.2",
"protocol": "ssh",
"username": "ca-oper",
"password": "",
"hop_enabled": True,
"hop_vendor": "bastion",
"hop_target_auth_mode": "bastion_managed",
"hop_host": "10.34.145.27",
"hop_username": "jump",
"hop_password": "hop-secret",
}
ready, reason = cli_creds_ready(creds, interactive=False)
self.assertTrue(ready)
self.assertEqual(reason, "")
def test_cli_hop_requires_target_password(self) -> None:
creds = {
"ip_address": "10.0.0.3",
"protocol": "ssh",
"username": "admin",
"password": "",
"hop_enabled": True,
"hop_vendor": "zte",
"hop_target_auth_mode": "manual",
"hop_host": "10.1.1.1",
"hop_username": "hop",
"hop_password": "hop-secret",
}
ready, reason = cli_creds_ready(creds, interactive=False)
self.assertFalse(ready)
self.assertEqual(reason, REASON_NO_PASSWORD)
def test_hop_incomplete(self) -> None:
creds = {
"ip_address": "10.0.0.4",
"protocol": "ssh",
"username": "admin",
"password": "secret",
"hop_enabled": True,
"hop_vendor": "zte",
"hop_host": "",
"hop_username": "hop",
"hop_password": "",
}
ready, reason = cli_creds_ready(creds, interactive=False)
self.assertFalse(ready)
self.assertEqual(reason, REASON_HOP_INCOMPLETE)
def test_interactive_telnet_allows_empty_password(self) -> None:
creds = {
"ip_address": "10.0.0.5",
"protocol": "telnet",
"username": "",
"password": "",
"hop_enabled": False,
}
ready, reason = cli_creds_ready(creds, interactive=True)
self.assertTrue(ready)
ready_exec, reason_exec = cli_creds_ready(creds, interactive=False)
self.assertFalse(ready_exec)
self.assertEqual(reason_exec, REASON_USERNAME_REQUIRED)
def test_non_interactive_telnet_requires_password(self) -> None:
creds = {
"ip_address": "10.0.0.6",
"protocol": "telnet",
"username": "admin",
"password": "",
"hop_enabled": False,
}
ready, reason = cli_creds_ready(creds, interactive=False)
self.assertFalse(ready)
self.assertEqual(reason, REASON_NO_PASSWORD)
if __name__ == "__main__":
unittest.main()

16
tests/test_config_sync.py
View file

@ -124,7 +124,13 @@ class ConfigSyncSnapshotOverwriteTests(unittest.TestCase):
task.ne_ip = "1.1.1.1"
db.get.return_value = task
resolve.return_value = (
{"host": "1.1.1.1"},
{
"ip_address": "1.1.1.1",
"protocol": "ssh",
"username": "admin",
"password": "secret",
"hop_enabled": False,
},
{"vendor": "Cisco", "device_type": "ios", "name": "r1", "ip_address": "1.1.1.1"},
)
@ -158,7 +164,13 @@ class ConfigSyncSnapshotOverwriteTests(unittest.TestCase):
task.ne_ip = "1.1.1.1"
db.get.return_value = task
resolve.return_value = (
{"host": "1.1.1.1"},
{
"ip_address": "1.1.1.1",
"protocol": "ssh",
"username": "admin",
"password": "secret",
"hop_enabled": False,
},
{"vendor": "Juniper", "device_type": "junos", "name": "r1", "ip_address": "1.1.1.1"},
)

32
tests/test_ne_exec.py
View file

@ -8,6 +8,18 @@ from fastapi import HTTPException
from netx_api.ne_exec import _validate_command, execute_managed_ne_commands
def _ready_creds(**overrides) -> dict:
base = {
"ip_address": "1.1.1.1",
"protocol": "ssh",
"username": "admin",
"password": "secret",
"hop_enabled": False,
}
base.update(overrides)
return base
class NeExecValidationTests(unittest.TestCase):
def test_allows_show(self) -> None:
_validate_command("show ip interface brief")
@ -130,7 +142,7 @@ class NeExecRunTests(unittest.TestCase):
@patch("netx_api.ne_exec.resolve_cli_target")
def test_execute_success(self, resolve, _collect, _configured) -> None:
resolve.return_value = (
{"ip_address": "1.1.1.1"},
_ready_creds(),
{
"source": "managed",
"id": "ne-1",
@ -152,6 +164,20 @@ class NeExecRunTests(unittest.TestCase):
self.assertEqual(out["output"], "ok-output")
self.assertEqual(out["commands"], ["show version"])
@patch("netx_api.ne_exec.credentials_configured", return_value=True)
@patch("netx_api.ne_exec._collect_on_device", return_value="ok-output")
@patch("netx_api.ne_exec.resolve_cli_target")
def test_execute_skips_when_no_password(self, resolve, collect, _configured) -> None:
resolve.return_value = (
_ready_creds(password=""),
{"source": "managed", "id": "ne-1", "name": "R2", "ip_address": "192.168.0.128"},
)
db = MagicMock()
out = execute_managed_ne_commands(db, ["show version"], ne_id="ne-1")
self.assertFalse(out["ok"])
self.assertEqual(out["error"], "no_password")
collect.assert_not_called()
@patch("netx_api.ne_exec.credentials_configured", return_value=True)
@patch("netx_api.ne_exec._collect_on_device", return_value="ok-output")
@patch("netx_api.ne_exec.resolve_cli_target")
@ -181,7 +207,7 @@ class NeExecRunTests(unittest.TestCase):
collect.assert_not_called()
resolve.return_value = (
{"ip_address": "1.1.1.1"},
_ready_creds(),
{
"source": "managed",
"id": "ne-1",
@ -221,7 +247,7 @@ class NeExecBatchTests(unittest.TestCase):
session_local.return_value = MagicMock()
resolve.return_value = (
{"ip_address": "1.1.1.1"},
_ready_creds(),
{
"source": "managed",
"id": "ne-1",