MrBlake/NetBox-VM-Import-Desktop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

fix: synchronize delayed switch CLI output

Browse Source 
Wait for the initial AOS-CX prompt and require each command response to end at the current prompt before sending the next command. Also recognize Aruba 1GbT port speeds.
This commit is contained in:
MrBlake
2026-07-30 11:03:13 +02:00
parent fff69277ab
commit 1b51784748
1 changed files with 19 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
switch_sync.py
+19 -17
View File
@@ -45,6 +45,7 @@ LLDP_CONTROL = {
ANSI_ESCAPE = re.compile(r"\x1b(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") ANSI_ESCAPE = re.compile(r"\x1b(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
MAC_PATTERN = re.compile(r"(?i)\b(?:[0-9a-f]{2}[:-]){5}[0-9a-f]{2}\b|\b[0-9a-f]{4}(?:-[0-9a-f]{4}){2}\b") MAC_PATTERN = re.compile(r"(?i)\b(?:[0-9a-f]{2}[:-]){5}[0-9a-f]{2}\b|\b[0-9a-f]{4}(?:-[0-9a-f]{4}){2}\b")
CLI_PROMPT = re.compile(r"(?:^|\n)[^\r\n]*[>#][ \t]*(?:\n)?\Z")
def _clean_output(value): def _clean_output(value):
@@ -53,26 +54,24 @@ def _clean_output(value):
return value return value
def _read_available(channel, quiet_seconds=0.8, timeout=20): def _read_available(channel, timeout=20):
chunks = [] chunks = []
started = time.monotonic() started = time.monotonic()
last_data = started
while time.monotonic() - started < timeout: while time.monotonic() - started < timeout:
if channel.recv_ready(): if channel.recv_ready():
chunks.append(channel.recv(65535).decode("utf-8", errors="replace")) chunks.append(channel.recv(65535).decode("utf-8", errors="replace"))
last_data = time.monotonic()
# Interactive AOS-CX/AOS-S/Comware commands are complete when the # Interactive AOS-CX/AOS-S/Comware commands are complete when the
# device prompt returns. This avoids truncating output at a short # device prompt returns. This avoids truncating output at a short
# pause while also avoiding a fixed delay after every command. # pause while also avoiding a fixed delay after every command.
if re.search(r"(?m)^\s*[^\r\n]+[>#]\s*$", "".join(chunks)): current = _clean_output("".join(chunks))
break if CLI_PROMPT.search(current):
elif chunks and time.monotonic() - last_data >= quiet_seconds:
break
elif not chunks and time.monotonic() - started >= min(timeout, 5):
break break
else: else:
time.sleep(0.05) time.sleep(0.05)
return _clean_output("".join(chunks)) output = _clean_output("".join(chunks))
if not CLI_PROMPT.search(output):
raise TimeoutError("Der Switch hat die CLI-Ausgabe nicht mit einem vollständigen Prompt abgeschlossen.")
return output
def run_switch_commands(config): def run_switch_commands(config):
@@ -94,7 +93,7 @@ def run_switch_commands(config):
banner_timeout=12, banner_timeout=12,
) )
channel = client.invoke_shell(width=240, height=1000) channel = client.invoke_shell(width=240, height=1000)
_read_available(channel, quiet_seconds=0.5, timeout=5) _read_available(channel, timeout=30)
pager_command, interface_command, neighbor_command = COMMANDS[platform] pager_command, interface_command, neighbor_command = COMMANDS[platform]
channel.send(pager_command + "\n") channel.send(pager_command + "\n")
_read_available(channel) _read_available(channel)
@@ -123,20 +122,20 @@ def run_switch_commands(config):
time.sleep(2) time.sleep(2)
channel.send(interface_command + "\n") channel.send(interface_command + "\n")
interfaces = _read_available(channel, quiet_seconds=1.2, timeout=30) interfaces = _read_available(channel, timeout=30)
ip_outputs = [] ip_outputs = []
for command in IP_COMMANDS[platform]: for command in IP_COMMANDS[platform]:
channel.send(command + "\n") channel.send(command + "\n")
ip_outputs.append(_read_available(channel, quiet_seconds=1.0, timeout=25)) ip_outputs.append(_read_available(channel, timeout=25))
channel.send(neighbor_command + "\n") channel.send(neighbor_command + "\n")
neighbors = _read_available(channel, quiet_seconds=1.5, timeout=35) neighbors = _read_available(channel, timeout=35)
if platform == "aruba_cx" and re.search(r"(?i)invalid input|unknown command|unrecognized", neighbors): if platform == "aruba_cx" and re.search(r"(?i)invalid input|unknown command|unrecognized", neighbors):
channel.send("show lldp neighbor-info\n") channel.send("show lldp neighbor-info\n")
neighbors = _read_available(channel, quiet_seconds=1.5, timeout=35) neighbors = _read_available(channel, timeout=35)
cdp_neighbors = "" cdp_neighbors = ""
if platform == "aruba_cx": if platform == "aruba_cx":
channel.send("show cdp neighbor-info\n") channel.send("show cdp neighbor-info\n")
cdp_summary = _read_available(channel, quiet_seconds=1.2, timeout=25) cdp_summary = _read_available(channel, timeout=25)
cdp_ports = [] cdp_ports = []
for line in cdp_summary.splitlines(): for line in cdp_summary.splitlines():
match = re.match(r"^\s*(\d+/\d+/\d+)\s+\S+", line) match = re.match(r"^\s*(\d+/\d+/\d+)\s+\S+", line)
@@ -145,7 +144,7 @@ def run_switch_commands(config):
details = [] details = []
for port in cdp_ports: for port in cdp_ports:
channel.send(f"show cdp neighbor-info {port}\n") channel.send(f"show cdp neighbor-info {port}\n")
details.append(_read_available(channel, quiet_seconds=1.0, timeout=20)) details.append(_read_available(channel, timeout=20))
cdp_neighbors = "\n".join([cdp_summary, *details]) cdp_neighbors = "\n".join([cdp_summary, *details])
return { return {
"interfaces": interfaces, "interfaces": interfaces,
@@ -169,7 +168,10 @@ def _normalize_mac(value):
def _speed_mbps(line): def _speed_mbps(line):
matches = re.findall(r"(?i)\b(\d+(?:\.\d+)?)\s*(T|G|M)(?:b(?:it)?|bps)?\b", line) matches = re.findall(
r"(?i)\b(\d+(?:\.\d+)?)\s*(T|G|M)(?:b(?:it)?|bps)?(?:T|X|SR|LR|CR)?\b",
line,
)
if not matches: if not matches:
duplex = re.search(r"(?i)\b(\d+)(?:FDx|HDx)\b", line) duplex = re.search(r"(?i)\b(\d+)(?:FDx|HDx)\b", line)
if duplex: if duplex: