Fix generator bug in send_config_set

netmiko/audiocode/audiocode_ssh.py

@@ -1,4 +1,4 @@
-from typing import Any, Optional, Sequence, TextIO, Union, List
+from typing import Any, Optional, Sequence, Iterator, TextIO, Union, List
 import time
 import re
 from netmiko.base_connection import BaseConnection
@@ -119,7 +119,7 @@ def exit_enable_mode(self, exit_command: str = "disable") -> str:
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         *,
         exit_config_mode: bool = True,
         read_timeout: Optional[float] = None,
@@ -359,7 +359,7 @@ def find_prompt(
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         *,
         exit_config_mode: bool = True,
         read_timeout: Optional[float] = None,

netmiko/base_connection.py

@@ -16,6 +16,7 @@
     cast,
     Type,
     Sequence,
+    Iterator,
     TextIO,
     Union,
     Tuple,
@@ -33,6 +34,7 @@
 from threading import Lock
 import functools
 import logging
+import itertools
 
 import paramiko
 import serial
@@ -2032,7 +2034,7 @@ def send_config_from_file(
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         *,
         exit_config_mode: bool = True,
         read_timeout: Optional[float] = None,
@@ -2129,8 +2131,10 @@ def send_config_set(
         # Set bypass_commands="" to force no-bypass (usually for testing)
         bypass_detected = False
         if bypass_commands:
+            # Make a copy of the iterator
+            config_commands, config_commands_tmp = itertools.tee(config_commands, 2)
             bypass_detected = any(
-                [True for cmd in config_commands if re.search(bypass_commands, cmd)]
+                [True for cmd in config_commands_tmp if re.search(bypass_commands, cmd)]
             )
         if bypass_detected:
             cmd_verify = False

netmiko/cdot/cdot_cros_ssh.py

@@ -7,7 +7,7 @@
 #
 # Purpose: Provide basic SSH connection to CROS based router products
 
-from typing import Optional, Union, Sequence, TextIO, Any
+from typing import Optional, Union, Sequence, Iterator, TextIO, Any
 import time
 import warnings
 from netmiko.no_enable import NoEnable
@@ -29,7 +29,7 @@ def session_preparation(self) -> None:
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = False,
         **kwargs: Any,
     ) -> str:

netmiko/cisco/cisco_viptela.py

@@ -1,5 +1,5 @@
 """Subclass specific to Cisco Viptela."""
-from typing import Union, Sequence, TextIO, Any
+from typing import Union, Sequence, Iterator, TextIO, Any
 import re
 
 from netmiko.cisco_base_connection import CiscoSSHConnection
@@ -38,7 +38,7 @@ def config_mode(
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = False,
         **kwargs: Any,
     ) -> str:

netmiko/cisco/cisco_wlc_ssh.py

@@ -1,5 +1,5 @@
 """Netmiko Cisco WLC support."""
-from typing import Any, Union, Sequence, TextIO
+from typing import Any, Union, Sequence, Iterator, TextIO
 import time
 import re
 import socket
@@ -225,7 +225,7 @@ def exit_config_mode(self, exit_config: str = "exit", pattern: str = "") -> str:
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = False,
         enter_config_mode: bool = False,
         **kwargs: Any,

netmiko/cisco/cisco_xr.py

@@ -1,4 +1,4 @@
-from typing import Optional, Any, Union, Sequence, TextIO
+from typing import Optional, Any, Union, Sequence, Iterator, TextIO
 import re
 import warnings
 from netmiko.base_connection import DELAY_FACTOR_DEPR_SIMPLE_MSG
@@ -43,7 +43,7 @@ def set_base_prompt(
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = False,
         **kwargs: Any,
     ) -> str:

netmiko/cloudgenix/cloudgenix_ion.py

@@ -1,4 +1,4 @@
-from typing import Any, Union, Sequence, TextIO, Optional
+from typing import Any, Union, Sequence, Iterator, TextIO, Optional
 from netmiko.no_config import NoConfig
 from netmiko.cisco_base_connection import CiscoSSHConnection
 
@@ -37,7 +37,7 @@ def save_config(self, *args: Any, **kwargs: Any) -> str:
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = False,
         **kwargs: Any
     ) -> str:

netmiko/ericsson/ericsson_ipos.py

@@ -1,7 +1,7 @@
 """
 Ericsson Ipos looks like it was RedBack equipment.
 """
-from typing import Optional, Any, Union, Sequence, TextIO
+from typing import Optional, Any, Union, Sequence, Iterator, TextIO
 import re
 import warnings
 
@@ -54,7 +54,7 @@ def exit_config_mode(self, exit_config: str = "end", pattern: str = "#") -> str:
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = False,
         **kwargs: Any,
     ) -> str:

netmiko/hp/hp_comware.py

@@ -1,5 +1,5 @@
 import re
-from typing import Union, Sequence, TextIO, Any, Optional
+from typing import Union, Sequence, Iterator, TextIO, Any, Optional
 
 from netmiko.cisco_base_connection import CiscoSSHConnection
 
@@ -44,7 +44,7 @@ def check_config_mode(
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = True,
         read_timeout: Optional[float] = None,
         delay_factor: Optional[float] = None,

netmiko/huawei/huawei.py

@@ -1,4 +1,4 @@
-from typing import Optional, Any, Union, Sequence, TextIO
+from typing import Optional, Any, Union, Sequence, Iterator, TextIO
 import time
 import re
 import warnings
@@ -196,7 +196,7 @@ def telnet_login(
 class HuaweiVrpv8SSH(HuaweiSSH):
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = False,
         **kwargs: Any,
     ) -> str:

netmiko/linux/linux_ssh.py

@@ -1,4 +1,4 @@
-from typing import Any, Optional, TYPE_CHECKING, Union, Sequence, TextIO
+from typing import Any, Optional, TYPE_CHECKING, Union, Sequence, Iterator, TextIO
 import os
 import re
 
@@ -61,7 +61,7 @@ def set_base_prompt(
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = True,
         **kwargs: Any,
     ) -> str:

netmiko/nokia/nokia_srl.py

@@ -7,7 +7,7 @@
 #   https://github.com/ktbyers/netmiko/blob/develop/LICENSE
 
 import re
-from typing import Any, Optional, Sequence, TextIO, Union
+from typing import Any, Optional, Sequence, Iterator, TextIO, Union
 from netmiko import log
 from netmiko.no_enable import NoEnable
 from netmiko.base_connection import BaseConnection
@@ -122,7 +122,7 @@ def exit_config_mode(self, exit_config: str = "", pattern: str = "") -> str:
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = False,
         **kwargs: Any,
     ) -> str:

netmiko/nokia/nokia_sros.py

@@ -9,7 +9,7 @@
 import re
 import os
 import time
-from typing import Any, Optional, Union, Sequence, TextIO, Callable
+from typing import Any, Optional, Union, Sequence, Iterator, TextIO, Callable
 
 from netmiko import log
 from netmiko.base_connection import BaseConnection
@@ -174,7 +174,7 @@ def save_config(self, *args: Any, **kwargs: Any) -> str:
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = None,
         **kwargs: Any,
     ) -> str:

netmiko/vyos/vyos_ssh.py

@@ -1,4 +1,4 @@
-from typing import Optional, Union, Sequence, TextIO, Any
+from typing import Optional, Union, Sequence, Iterator, TextIO, Any
 import time
 import warnings
 import re
@@ -123,7 +123,7 @@ def set_base_prompt(
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = False,
         **kwargs: Any,
     ) -> str:

netmiko/zyxel/zyxel_ssh.py

@@ -1,4 +1,4 @@
-from typing import Any, Sequence, TextIO, Union
+from typing import Any, Sequence, Iterator, TextIO, Union
 from netmiko.cisco_base_connection import CiscoSSHConnection
 from netmiko.no_enable import NoEnable
 from netmiko.no_config import NoConfig
@@ -11,7 +11,7 @@ def disable_paging(self, *args: Any, **kwargs: Any) -> str:
 
     def send_config_set(
         self,
-        config_commands: Union[str, Sequence[str], TextIO, None] = None,
+        config_commands: Union[str, Sequence[str], Iterator[str], TextIO, None] = None,
         exit_config_mode: bool = False,
         enter_config_mode: bool = False,
         **kwargs: Any

requirements-genie.txt

@@ -1,2 +1,2 @@
-pyats==22.5
-genie==22.5
+pyats==22.7.1
+genie==22.7

tests/SLOG/cisco881_slog.log

@@ -1,3 +1,4 @@
+
 cisco1>terminal width 511
 cisco1>terminal length 0
 cisco1>

tests/SLOG/cisco881_slog_append.log

@@ -21,6 +21,7 @@ cisco1>
 Testing password and secret replacement
 This is my password ********
 This is my secret ********
+
 cisco1>terminal width 511
 cisco1>terminal length 0
 cisco1>

tests/SLOG/netmiko.log

@@ -35,17 +35,19 @@ cisco1#
 write_channel: b'exit\n'
 write_channel: b'terminal width 511\n'
 read_channel: 
-read_channel: cisco1>terminal width
+read_channel: 
+cisco1>terminal width
 read_channel:  511
 cisco1>
-Pattern found: (terminal width 511) cisco1>terminal width 511
+Pattern found: (terminal width 511) 
+cisco1>terminal width 511
 In disable_paging
 Command: terminal length 0
 
 write_channel: b'terminal length 0\n'
 read_channel: 
-read_channel: terminal length
-read_channel:  0
+read_channel: terminal lengt
+read_channel: h 0
 cisco1>
 Pattern found: (terminal\ length\ 0) 
 cisco1>terminal length 0
@@ -65,10 +67,12 @@ read_channel:
 [find_prompt()]: prompt is cisco1>
 write_channel: b'terminal width 511\n'
 read_channel: 
-read_channel: cisco1>terminal width
+read_channel: 
+cisco1>terminal width
 read_channel:  511
 cisco1>
-Pattern found: (terminal width 511) cisco1>terminal width 511
+Pattern found: (terminal width 511) 
+cisco1>terminal width 511
 In disable_paging
 Command: terminal length 0
 
@@ -102,8 +106,8 @@ read_channel:
 [find_prompt()]: prompt is cisco1>
 write_channel: b'show ip interface brief\n'
 read_channel: 
-read_channel: show ip interf
-read_channel: ace brief
+read_channel: show ip interfa
+read_channel: ce brief
 Interface                  IP-Address      OK? Method Status                Protocol
 FastEthernet0              unassigned      YES unset  down                  down    
 FastEthernet1              unassigned      YES unset  down                  down    

tests/test_netmiko_config.py

@@ -91,6 +91,38 @@ def test_config_set(net_connect, commands, expected_responses):
         assert config_commands[-1] in config_commands_output
 
 
+def test_config_set_generator(net_connect, commands, expected_responses):
+    """Test sending configuration commands as a generator."""
+
+    config_commands = commands["config"]
+    # Make a generator out of the config commands (to verify no issues with generators)
+    config_commands_gen = (cmd for cmd in config_commands)
+    support_commit = commands.get("support_commit")
+    config_verify = commands["config_verification"]
+
+    # Set to initial value and testing sending command as a string
+    net_connect.send_config_set(config_commands[0])
+    if support_commit:
+        net_connect.commit()
+    cmd_response = expected_responses.get("cmd_response_init")
+    config_commands_output = net_connect.send_command(config_verify)
+    if cmd_response:
+        assert cmd_response in config_commands_output
+    else:
+        assert config_commands[0] in config_commands_output
+
+    # Send the config commands as a generator
+    net_connect.send_config_set(config_commands_gen)
+    if support_commit:
+        net_connect.commit()
+    cmd_response = expected_responses.get("cmd_response_final")
+    config_commands_output = net_connect.send_command_expect(config_verify)
+    if cmd_response:
+        assert cmd_response in config_commands_output
+    else:
+        assert config_commands[-1] in config_commands_output
+
+
 def test_config_set_longcommand(net_connect, commands, expected_responses):
     """Test sending configuration commands using long commands"""
     config_commands = commands.get("config_long_command")