Fix MikroTik command output reads without resending commands

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(

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

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"),