Fix default ports.

.github/workflows/ci.yml

@@ -83,7 +83,7 @@ jobs:
     strategy:
       fail-fast: true
       matrix:
-        python-version: ["3.7", "3.8", "3.9"]  # "3.10" add back after pyeapi new release.
+        python-version: ["3.8", "3.9", "3.10", "3.11"]  # "3.10" add back after pyeapi new release.
     runs-on: "ubuntu-20.04"
     env:
       PYTHON_VER: "${{ matrix.python-version }}"
@@ -156,7 +156,7 @@ jobs:
     strategy:
       fail-fast: true
       matrix:
-        python-version: ["3.7", "3.8", "3.9"]  # "3.10" add back after pyeapi new release.
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
     runs-on: "ubuntu-20.04"
     env:
       PYTHON_VER: "${{ matrix.python-version }}"

Dockerfile

@@ -15,7 +15,7 @@ RUN apt-get update && \
 RUN pip install --upgrade pip
 
 RUN curl -sSL https://install.python-poetry.org -o /tmp/install-poetry.py && \
-    python /tmp/install-poetry.py --version 1.2.0 && \
+    python /tmp/install-poetry.py --version 1.6.0 && \
     rm -f /tmp/install-poetry.py
 
 # Add poetry install location to the $PATH

pyntc/devices/aireos_device.py

@@ -70,7 +70,7 @@ class AIREOSDevice(BaseDevice):
     active_redundancy_states = {None, "active"}
 
     def __init__(  # nosec  # pylint: disable=too-many-arguments
-        self, host, username, password, secret="", port=22, confirm_active=True, **kwargs
+        self, host, username, password, secret="", port=None, confirm_active=True, **kwargs
     ):  # noqa: D403
         """
         PyNTC Device implementation for Cisco WLC.
@@ -80,13 +80,13 @@ def __init__(  # nosec  # pylint: disable=too-many-arguments
             username (str): The username to authenticate with the device.
             password (str): The password to authenticate with the device.
             secret (str): The password to escalate privilege on the device.
-            port (int): The port to use to establish the connection.
+            port (int): The port to use to establish the connection. Defaults to 22.
             confirm_active (bool): Determines if device's high availability state should be validated before leaving connection open.
         """
         super().__init__(host, username, password, device_type="cisco_aireos_ssh")
         self.native = None
         self.secret = secret
-        self.port = int(port)
+        self.port = int(port) if port else 22
         self.global_delay_factor = kwargs.get("global_delay_factor", 1)
         self.delay_factor = kwargs.get("delay_factor", 1)
         self._connected = False

pyntc/devices/asa_device.py

@@ -38,22 +38,22 @@ class ASADevice(BaseDevice):
     vendor = "cisco"
     active_redundancy_states = {None, "active"}
 
-    def __init__(self, host: str, username: str, password: str, secret="", port=22, **kwargs):  # nosec
+    def __init__(self, host: str, username: str, password: str, secret="", port=None, **kwargs):  # nosec
         """
         Pyntc Device constructor for Cisco ASA.
 
         Args:
             host (str): The address of the network device.
             username (str): The username to authenticate to the device.
             password (str): The password to authenticate to the device.
-            secret (str, optional): The password to escalate privilege on the device. Defaults to "".
+            secret (str, optional): The password to escalate privilege on the device. Defaults to 22.
             port (int, optional): Port used to establish connection. Defaults to 22.
         """
         super().__init__(host, username, password, device_type="cisco_asa_ssh")
 
         self.native: Optional[CiscoAsaSSH] = None
         self.secret = secret
-        self.port = int(port)
+        self.port = int(port) if port else 22
         self.kwargs = kwargs
         self.global_delay_factor: int = kwargs.get("global_delay_factor", 1)
         self.delay_factor: int = kwargs.get("delay_factor", 1)

pyntc/devices/ios_device.py

@@ -44,7 +44,7 @@ class IOSDevice(BaseDevice):
     active_redundancy_states = {None, "active"}
 
     def __init__(  # nosec
-        self, host, username, password, secret="", port=22, confirm_active=True, fast_cli=True, **kwargs
+        self, host, username, password, secret="", port=None, confirm_active=True, fast_cli=True, **kwargs
     ):  # noqa: D403
         """
         PyNTC Device implementation for Cisco IOS.
@@ -54,15 +54,15 @@ def __init__(  # nosec
             username (str): The username to authenticate with the device.
             password (str): The password to authenticate with the device.
             secret (str): The password to escalate privilege on the device.
-            port (int): The port to use to establish the connection.
+            port (int): The port to use to establish the connection. Defaults to 22.
             confirm_active (bool): Determines if device's high availability state should be validated before leaving connection open.
             fast_cli (bool): Fast CLI mode for Netmiko, it is recommended to use False when opening the client on code upgrades
         """
         super().__init__(host, username, password, device_type="cisco_ios_ssh")
 
         self.native = None
         self.secret = secret
-        self.port = int(port)
+        self.port = int(port) if port else 22
         self.global_delay_factor = kwargs.get("global_delay_factor", 1)
         self.delay_factor = kwargs.get("delay_factor", 1)
         self._fast_cli = fast_cli

pyproject.toml

@@ -24,14 +24,11 @@ include = [
 ]
 
 [tool.poetry.dependencies]
-python = "^3.7"
-# Required for Python 3.7 for now. See: https://stackoverflow.com/a/73932581/194311
-importlib-metadata = "4.13.0"
+python = "^3.8"
 f5-sdk = "^3.0.21"
 junos-eznc = "^2.6"
 netmiko = "^4.0"
-# pyeapi doesn't support py3.10 yet in a release, and pypi doesn't allow direct dependencies. py3.10 to work. https://github.com/arista-eosplus/pyeapi/blob/236503162d1aa3ecc953678ec05380f1f605be02/pyeapi/api/abstract.py#L44
-pyeapi = "^0.8.4"
+pyeapi = "^1.0.2"
 pynxos = "^0.0.5"
 requests = "^2.28"
 scp = "^0.14"

tests/unit/test_devices/test_aireos_device.py

@@ -1671,3 +1671,13 @@ def test_uptime_string(mock_uptime_components, aireos_device):
 def test_wlans(aireos_show, aireos_expected_wlans):
     device = aireos_show(["show_wlan_summary.txt"])
     assert device.wlans == aireos_expected_wlans
+
+
+def test_port(aireos_device):
+    assert aireos_device.port == 22
+
+
+@mock.patch.object(AIREOSDevice, "open")
+def test_port_none(patch):
+    device = AIREOSDevice("host", "user", "pass", port=None)
+    assert device.port == 22

tests/unit/test_devices/test_asa_device.py

@@ -41,7 +41,6 @@
 class TestASADevice:
     def setup(self, api):
         with mock.patch("pyntc.devices.asa_device.ConnectHandler") as api:
-
             if not getattr(self, "device", None):
                 self.device = ASADevice("host", "user", "password")
 
@@ -66,6 +65,9 @@ def teardown(self):
         self.device.native.reset_mock()
         self.count_teardown += 1
 
+    def test_port(self):
+        assert self.device.port == 22
+
     @mock.patch.object(ASADevice, "_get_file_system", return_value="disk0:")
     def test_boot_options_dir(self, mock_boot):
         self.device.native.send_command_timing.side_effect = None
@@ -891,3 +893,9 @@ def test_vlan(mock_get_vlans, asa_device):
     mock_get_vlans.return_value = expected
     vlans = asa_device.vlans
     assert vlans == [10, 20]
+
+
+@mock.patch.object(ASADevice, "open")
+def test_port_none(patch):
+    device = ASADevice("host", "user", "pass", port=None)
+    assert device.port == 22

tests/unit/test_devices/test_eos_device.py

@@ -212,7 +212,6 @@ def test_file_copy_remote_exists_bad_md5(self, mock_open, mock_close, mock_ssh,
     @mock.patch.object(EOSDevice, "close")
     @mock.patch("netmiko.arista.arista.AristaSSH", autospec=True)
     def test_file_copy_remote_not_exist(self, mock_open, mock_close, mock_ssh, mock_ft):
-
         self.device.native_ssh = mock_open
         self.device.native_ssh.send_command_timing.side_effect = None
         self.device.native_ssh.send_command_timing.return_value = "flash: /dev/null"

tests/unit/test_devices/test_f5_device.py

@@ -132,7 +132,7 @@ def test_reboot(self):
 
         volume = VOLUME
         # skip the wait_for_device_reboot
-        with (mock.patch.object(self.device, "_wait_for_device_reboot", return_value=True)):
+        with mock.patch.object(self.device, "_wait_for_device_reboot", return_value=True):
             self.device.reboot(volume=volume)
 
         # # Check if _get_active_volume worked
@@ -146,7 +146,7 @@ def test_reboot_with_timer(self):
         api.tm.sys.software.volumes.volume.load.return_value.active = True
 
         # skipping timeout! It's too long!!
-        with (mock.patch.object(self.device, "_wait_for_device_reboot", timeout=0)):
+        with mock.patch.object(self.device, "_wait_for_device_reboot", timeout=0):
             self.device.reboot(volume=volume)
 
         # # Check if _get_active_volume worked
@@ -157,7 +157,7 @@ def test_reboot_with_timer(self):
     def test_reboot_no_volume(self):
         api = self.device.api_handler
 
-        with (mock.patch.object(self.device, "_wait_for_device_reboot", return_value=True)):
+        with mock.patch.object(self.device, "_wait_for_device_reboot", return_value=True):
             self.device.reboot()
 
         # Check if _reboot_to_volume worked
@@ -175,7 +175,7 @@ def test_set_boot_options(self):
         # Patching out _volume_exists for _image_install
         api.tm.sys.software.volumes.volume.exists.return_value = True
 
-        with (mock.patch.object(self.device, "_wait_for_image_installed", timeout=0, return_value=None)):
+        with mock.patch.object(self.device, "_wait_for_image_installed", timeout=0, return_value=None):
             self.device.set_boot_options(image_name=image_name, volume=volume)
 
         api.tm.util.bash.exec_cmd.assert_called()
@@ -194,7 +194,7 @@ def test_set_boot_options_no_image(self):
         # Patching out _volume_exists for _image_install
         api.tm.sys.software.volumes.volume.exists.return_value = False
 
-        with (mock.patch.object(self.device, "_wait_for_image_installed", timeout=0, return_value=None)):
+        with mock.patch.object(self.device, "_wait_for_image_installed", timeout=0, return_value=None):
             self.device.set_boot_options(image_name=image_name, volume=volume)
 
         api.tm.util.bash.exec_cmd.assert_called()
@@ -215,7 +215,7 @@ def test_set_boot_options_bad_boot(self):
         # Patching out _volume_exists for _image_install
         api.tm.sys.software.volumes.volume.exists.return_value = False
 
-        with (mock.patch.object(self.device, "_wait_for_image_installed", timeout=0, return_value=None)):
+        with mock.patch.object(self.device, "_wait_for_image_installed", timeout=0, return_value=None):
             with pytest.raises(NTCFileNotFoundError):
                 self.device.set_boot_options(image_name="bad_image", volume=volume)
 
@@ -248,7 +248,7 @@ def test_install_os(self):
         # Patching out _image_install
         api.tm.sys.software.volumes.volume.exists.return_value = True
 
-        with (mock.patch.object(self.device, "_wait_for_image_installed", timeout=0, return_value=None)):
+        with mock.patch.object(self.device, "_wait_for_image_installed", timeout=0, return_value=None):
             self.device.install_os(image_name=image_name, volume=volume)
 
         api.tm.util.bash.exec_cmd.assert_called()

tests/unit/test_devices/test_ios_device.py

@@ -78,6 +78,9 @@ def tearDown(self):
         # Reset the mock so we don't have transient test effects
         self.device.native.reset_mock()
 
+    def test_port(self):
+        self.assertEqual(self.device.port, 22)
+
     def test_bad_show(self):
         command = "show microsoft"
         self.device.native.send_command.return_value = "Error: Microsoft"
@@ -393,6 +396,12 @@ def test_install_os_error(self, mock_wait, mock_reboot, mock_set_boot, mock_imag
     unittest.main()
 
 
+@mock.patch.object(IOSDevice, "open")
+def test_port_none(patch):
+    device = IOSDevice("host", "user", "pass", port=None)
+    assert device.port == 22
+
+
 def test_check_command_output_for_errors(ios_device):
     command_passes = ios_device._check_command_output_for_errors("valid command", "valid output")
     assert command_passes is None
@@ -978,6 +987,7 @@ def test_set_boot_options_image_packages_conf_file(
 # TESTS FOR IOS INSTALL MODE METHOD
 #
 
+
 # Test install mode upgrade for install mode with latest method
 @mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock)
 @mock.patch.object(IOSDevice, "_image_booted")
@@ -1106,6 +1116,7 @@ def test_install_os_install_mode_no_upgrade(
 # FROM CISCO IOS EVEREST VERSION TESTS
 #
 
+
 # Test install mode upgrade for install mode with interim method on OS Version
 @mock.patch.object(IOSDevice, "os_version", new_callable=mock.PropertyMock)
 @mock.patch.object(IOSDevice, "_image_booted")