From ce9a78b5ec0bdb9db7f179ac60a365fccb8943c9 Mon Sep 17 00:00:00 2001 From: oliver Date: Tue, 13 Oct 2026 23:17:18 +0800 Subject: [PATCH] Fix MikroTik command output reads without resending commands --- netx_api/ne_netmiko.py | 20 ++++++++++++++++ tests/test_ne_netmiko_show.py | 45 +++++++++++++++++++++++++++++++++++ tests/test_paging_disable.py | 6 +++++ 3 files changed, 71 insertions(+) diff --git a/netx_api/ne_netmiko.py b/netx_api/ne_netmiko.py index 9b351e6..5353af1 100644 --- a/netx_api/ne_netmiko.py +++ b/netx_api/ne_netmiko.py @@ -47,6 +47,9 @@ def paging_disable_commands(*, vendor: str = "", device_type: str = "") -> list[ dt = str(device_type or "").strip().lower() v = str(vendor or "").strip().lower() blob = f"{dt} {v}" + # RouterOS has no pager; Cisco's command leaves an extra error/prompt redraw. + if "mikrotik" in blob or "routeros" in blob or "switchos" in blob: + return [] 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: @@ -134,6 +137,23 @@ def drain_read_channel( def _send_command_expect_prompt(conn: Any, cmd: str, *, read_timeout: int) -> str: + device_type = getattr(conn, "device_type", "") + if isinstance(device_type, str) and device_type.lower().startswith("mikrotik"): + # find_prompt sends an extra Enter *after* our channel drain. RouterOS + # can repaint that old prompt after send_command starts, ending the read + # before the command's output arrives. Use its complete prompt directly. + # Keep echo verification off: long RouterOS scripts wrap at PTY width. + prompt_pattern = r"(?m)^\[[^\r\n\]]+\](?:[ \t]+/[^\r\n>]*)?[ \t]*>[ \t]*$" + return str( + conn.send_command( + command_string=cmd, + read_timeout=read_timeout, + auto_find_prompt=False, + expect_string=prompt_pattern, + cmd_verify=False, + ) + or "" + ) try: return str( conn.send_command( diff --git a/tests/test_ne_netmiko_show.py b/tests/test_ne_netmiko_show.py index 2add1bd..5c57bb4 100644 --- a/tests/test_ne_netmiko_show.py +++ b/tests/test_ne_netmiko_show.py @@ -3,6 +3,7 @@ from __future__ import annotations import unittest +import re from unittest.mock import MagicMock, patch from netx_api.ne_netmiko import drain_read_channel, send_show_command @@ -23,6 +24,50 @@ class DrainReadChannelTests(unittest.TestCase): class SendShowCommandTests(unittest.TestCase): + @patch("netx_api.ne_netmiko.drain_read_channel") + def test_mikrotik_does_not_probe_prompt_after_drain(self, drain: MagicMock) -> None: + conn = MagicMock() + conn.device_type = "mikrotik_routeros_ssh" + + def delayed_routeros_output(**kwargs: object) -> str: + # A prompt probe queues an old prompt before the actual output. + if kwargs.get("auto_find_prompt", True): + return "" + return "name: MikroTik" + + conn.send_command.side_effect = delayed_routeros_output + self.assertEqual(send_show_command(conn, "/system identity print"), "name: MikroTik") + conn.send_command.assert_called_once() + conn.find_prompt.assert_not_called() + conn.send_command_timing.assert_not_called() + drain.assert_called_once_with(conn) + pattern = conn.send_command.call_args.kwargs["expect_string"] + for prompt in ("[admin@MikroTik] >", "[ops@edge-router] /ip/firewall>"): + self.assertIsNotNone(re.search(pattern, prompt)) + for echo in ("[admin@MikroTik]", "[admin@MikroTik] > /system resource print"): + self.assertIsNone(re.search(pattern, echo)) + + @patch("netx_api.ne_netmiko.drain_read_channel") + def test_mikrotik_wrapped_script_does_not_require_exact_echo(self, drain: MagicMock) -> None: + conn = MagicMock() + conn.device_type = "mikrotik_routeros" + conn.send_command.return_value = "finished" + command = ':put "' + "x" * 600 + '"' + self.assertEqual(send_show_command(conn, command, read_timeout=45), "finished") + conn.send_command.assert_called_once() + self.assertFalse(conn.send_command.call_args.kwargs["cmd_verify"]) + self.assertEqual(conn.send_command.call_args.kwargs["read_timeout"], 45) + + @patch("netx_api.ne_netmiko.drain_read_channel") + def test_mikrotik_timeout_never_resends_write(self, drain: MagicMock) -> None: + conn = MagicMock() + conn.device_type = "mikrotik_routeros" + conn.send_command.side_effect = TimeoutError("late prompt") + with self.assertRaises(TimeoutError): + send_show_command(conn, '/ip address add address=192.0.2.1/32 interface=lo') + conn.send_command.assert_called_once() + conn.send_command_timing.assert_not_called() + @patch("netx_api.ne_netmiko.drain_read_channel") def test_drains_then_sends_once(self, drain: MagicMock) -> None: conn = MagicMock() diff --git a/tests/test_paging_disable.py b/tests/test_paging_disable.py index d4030ec..74a6e02 100644 --- a/tests/test_paging_disable.py +++ b/tests/test_paging_disable.py @@ -9,6 +9,12 @@ from netx_api.ne_netmiko import disable_target_paging, paging_disable_commands class PagingDisableTests(unittest.TestCase): + def test_routeros_does_not_send_cisco_paging_command(self) -> None: + conn = MagicMock() + self.assertEqual(disable_target_paging(conn, device_type="mikrotik_routeros"), "") + conn.send_command_timing.assert_not_called() + conn.write_channel.assert_not_called() + def test_commands_by_vendor(self) -> None: self.assertEqual( paging_disable_commands(vendor="Huawei", device_type="huawei"),