utilities: Replace telnetlib with socket.

telnetlib is getting removed from standard python library in python 3.13[0].
While there are 3rd party replacements, it was used only to check if a port is
open. We can use built-in socket library for that.

[0] https://docs.python.org/3.13/library/telnetlib.html

Signed-off-by: Martin Kalcok <[email protected]>

Author: mkalcok

unit_tests/utilities/test_zaza_utilities_generic.py

@@ -415,20 +415,31 @@ def test_get_ubuntu_release(self):
             generic_utils.get_ubuntu_release(bad_name)
 
     def test_is_port_open(self):
+        mock_sock_instance = mock.MagicMock()
         self.patch(
-            'zaza.openstack.utilities.generic.telnetlib.Telnet',
-            new_callable=mock.MagicMock(),
-            name='telnet'
+            'zaza.openstack.utilities.generic.socket.socket',
+            return_value=mock_sock_instance,
+            name='mock_socket'
         )
 
         _port = "80"
         _addr = "10.5.254.20"
 
+        # Test successful connection
         self.assertTrue(generic_utils.is_port_open(_port, _addr))
-        self.telnet.assert_called_with(_addr, _port)
+        self.mock_socket.assert_called_with(
+            generic_utils.socket.AF_INET,
+            generic_utils.socket.SOCK_STREAM
+        )
+        mock_sock_instance.settimeout.assert_called_with(5)
+        mock_sock_instance.connect.assert_called_with((_addr, 80))
+        mock_sock_instance.close.assert_called_once()
 
-        self.telnet.side_effect = generic_utils.socket.error
+        # Test failed connection - socket should still be closed
+        mock_sock_instance.reset_mock()
+        mock_sock_instance.connect.side_effect = generic_utils.socket.error
         self.assertFalse(generic_utils.is_port_open(_port, _addr))
+        mock_sock_instance.close.assert_called_once()
 
     def test_get_unit_hostnames(self):
         self.patch(

zaza/openstack/utilities/generic.py

@@ -19,7 +19,6 @@
 import os
 import socket
 import subprocess
-import telnetlib
 import tempfile
 import yaml
 
@@ -583,20 +582,24 @@ def get_file_contents(unit, f):
                              "cat {}".format(f))['Stdout']
 
 
-def is_port_open(port, address):
+def is_port_open(port, address, timeout=5):
     """Determine if TCP port is accessible.
 
-    Connect to the MySQL port on the VIP.
+    Connect to a TCP port to check if it is accessible.
 
     :param port: Port number
-    :type port: str
-    :param address: IP address
-    :type port: str
+    :type port: str or int
+    :param address: IP address or hostname
+    :type address: str
+    :param timeout: Connection timeout in seconds
+    :type timeout: int
     :returns: True if port is reachable
     :rtype: boolean
     """
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.settimeout(timeout)
     try:
-        telnetlib.Telnet(address, port)
+        sock.connect((address, int(port)))
         return True
     except socket.error as e:
         if e.errno == 113:
@@ -606,6 +609,8 @@ def is_port_open(port, address):
             logging.error("connection refused connecting"
                           " to {}:{}".format(address, port))
         return False
+    finally:
+        sock.close()
 
 
 def port_knock_units(units, port=22, expect_success=True):