unit tests

jneprz · jneprz · commit 5df8b3a9a87a · 2025-07-03T13:58:26.000-07:00
diff --git a/src/sftp/azext_sftp/_validators.py b/src/sftp/azext_sftp/_validators.py
@@ -13,7 +13,7 @@ def storage_account_name_or_id_validator(cmd, namespace):
     Validator for storage account name or resource ID.
     Converts storage account name to full resource ID if needed.
     """
-    if namespace.storage_account:
+    if hasattr(namespace, 'storage_account') and namespace.storage_account:
         if not is_valid_resource_id(namespace.storage_account):
             if not hasattr(namespace, 'resource_group_name') or not namespace.resource_group_name:
                 raise azclierror.RequiredArgumentMissingError(
diff --git a/src/sftp/azext_sftp/custom.py b/src/sftp/azext_sftp/custom.py
@@ -69,6 +69,7 @@ def sftp_cert(cmd, cert_path=None, public_key_file=None, ssh_client_folder=None)
         logger.warning("%s contains sensitive information (id_rsa, id_rsa.pub). "
                        "Please delete once this certificate is no longer being used.", keys_folder)
 
+    # pylint: disable=broad-except
     try:
         cert_expiration = sftp_utils.get_certificate_start_and_end_times(cert_file, ssh_client_folder)[1]
         print_styled_text((Style.SUCCESS,
@@ -103,15 +104,15 @@ def sftp_connect(cmd, storage_account, port=None, cert_file=None, private_key_fi
         delete_cert = True
         delete_keys = True
         credentials_folder = tempfile.mkdtemp(prefix="aadsftp")
-        
+
         try:
             profile = Profile(cli_ctx=cmd.cli_ctx)
             profile.get_subscription()
         except Exception:
             if credentials_folder and os.path.isdir(credentials_folder):
                 shutil.rmtree(credentials_folder)
             raise
-        
+
         print_styled_text((Style.ACTION, "Generating temporary credentials..."))
 
     if cert_file and public_key_file:
@@ -123,12 +124,9 @@ def sftp_connect(cmd, storage_account, port=None, cert_file=None, private_key_fi
                 None, None, credentials_folder, ssh_client_folder)
             cert_file, user = _get_and_write_certificate(cmd, public_key_file, None, ssh_client_folder)
         elif not cert_file:
-            try:
-                profile = Profile(cli_ctx=cmd.cli_ctx)
-                profile.get_subscription()
-            except Exception:
-                raise
-            
+            profile = Profile(cli_ctx=cmd.cli_ctx)
+            profile.get_subscription()
+
             public_key_file, private_key_file, _ = _check_or_create_public_private_files(
                 public_key_file, private_key_file, None, ssh_client_folder)
             print_styled_text((Style.ACTION, "Generating certificate..."))
diff --git a/src/sftp/azext_sftp/file_utils.py b/src/sftp/azext_sftp/file_utils.py
@@ -40,6 +40,7 @@ def mkdir_p(path):
 
 def delete_file(file_path, message, warning=False):
     if os.path.isfile(file_path):
+        # pylint: disable=broad-except
         try:
             os.remove(file_path)
         except Exception as e:
@@ -51,6 +52,7 @@ def delete_file(file_path, message, warning=False):
 
 def delete_folder(dir_path, message, warning=False):
     if os.path.isdir(dir_path):
+        # pylint: disable=broad-except
         try:
             os.rmdir(dir_path)
         except Exception as e:
diff --git a/src/sftp/azext_sftp/rsa_parser.py b/src/sftp/azext_sftp/rsa_parser.py
@@ -7,6 +7,7 @@
 import struct
 
 
+# pylint: disable=too-few-public-methods
 class RSAParser():
     RSAAlgorithm = 'ssh-rsa'
 
diff --git a/src/sftp/azext_sftp/sftp_info.py b/src/sftp/azext_sftp/sftp_info.py
@@ -10,6 +10,7 @@
 logger = log.get_logger(__name__)
 
 
+# pylint: disable=too-few-public-methods
 class ConnectionInfo:
     """Encapsulates connection-specific information."""
 
@@ -20,6 +21,7 @@ def __init__(self, storage_account, username=None, host=None, port=None):
         self.port = port
 
 
+# pylint: disable=too-few-public-methods
 class AuthenticationFiles:
     """Encapsulates authentication file paths."""
 
@@ -29,6 +31,7 @@ def __init__(self, public_key_file=None, private_key_file=None, cert_file=None):
         self.cert_file = os.path.abspath(os.path.expanduser(cert_file)) if cert_file else None
 
 
+# pylint: disable=too-few-public-methods
 class SessionConfiguration:
     """Encapsulates session configuration options."""
 
@@ -41,6 +44,7 @@ def __init__(self, sftp_args=None, ssh_client_folder=None, ssh_proxy_folder=None
         self.yes_without_prompt = yes_without_prompt
 
 
+# pylint: disable=too-few-public-methods
 class RuntimeState:
     """Encapsulates runtime state information."""
 
diff --git a/src/sftp/azext_sftp/sftp_utils.py b/src/sftp/azext_sftp/sftp_utils.py
@@ -69,7 +69,7 @@ def _execute_sftp_process(command, env, creationflags):
         return sftp_process, None
 
 
-def _attempt_connection(command, env, creationflags, op_info, attempt_num):
+def _attempt_connection(command, env, creationflags, op_info, attempt_num):  # pylint: disable=unused-argument
     """Attempt a single SFTP connection."""
     connection_start_time = time.time()
     try:
diff --git a/src/sftp/azext_sftp/tests/latest/test_connectivity_utils.py b/src/sftp/azext_sftp/tests/latest/test_connectivity_utils.py
@@ -0,0 +1,227 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import unittest
+import json
+import base64
+from unittest import mock
+
+from azext_sftp import connectivity_utils
+
+
+class SftpConnectivityUtilsTest(unittest.TestCase):
+    """Test suite for SFTP connectivity utilities.
+    
+    Owner: johnli1
+    """
+
+    def test_format_relay_info_string_success(self):
+        """Test format_relay_info_string with valid relay information."""
+        # Arrange
+        relay_info = {
+            'namespaceName': 'test-namespace',
+            'namespaceNameSuffix': 'servicebus.windows.net',
+            'hybridConnectionName': 'test-connection',
+            'accessKey': 'test-access-key',
+            'expiresOn': '2025-07-04T10:00:00Z',
+            'serviceConfigurationToken': 'test-token'
+        }
+        
+        # Act
+        result = connectivity_utils.format_relay_info_string(relay_info)
+        
+        # Assert
+        # Decode the base64 result to verify the structure
+        decoded_bytes = base64.b64decode(result.encode('ascii'))
+        decoded_string = decoded_bytes.decode('ascii')
+        parsed_result = json.loads(decoded_string)
+        
+        self.assertIn('relay', parsed_result)
+        relay_data = parsed_result['relay']
+        
+        self.assertEqual(relay_data['namespaceName'], 'test-namespace')
+        self.assertEqual(relay_data['namespaceNameSuffix'], 'servicebus.windows.net')
+        self.assertEqual(relay_data['hybridConnectionName'], 'test-connection')
+        self.assertEqual(relay_data['accessKey'], 'test-access-key')
+        self.assertEqual(relay_data['expiresOn'], '2025-07-04T10:00:00Z')
+        self.assertEqual(relay_data['serviceConfigurationToken'], 'test-token')
+
+    def test_format_relay_info_string_minimal_data(self):
+        """Test format_relay_info_string with minimal required fields."""
+        # Arrange
+        relay_info = {
+            'namespaceName': 'minimal',
+            'namespaceNameSuffix': 'suffix',
+            'hybridConnectionName': 'connection',
+            'accessKey': 'key',
+            'expiresOn': '2025-01-01T00:00:00Z',
+            'serviceConfigurationToken': 'token'
+        }
+        
+        # Act
+        result = connectivity_utils.format_relay_info_string(relay_info)
+        
+        # Assert
+        self.assertIsInstance(result, str)
+        self.assertTrue(len(result) > 0)
+        
+        # Verify it's valid base64
+        try:
+            decoded = base64.b64decode(result.encode('ascii'))
+            json.loads(decoded.decode('ascii'))
+        except (ValueError, json.JSONDecodeError):
+            self.fail("Result is not valid base64-encoded JSON")
+
+    def test_format_relay_info_string_special_characters(self):
+        """Test format_relay_info_string with special characters in data."""
+        # Arrange
+        relay_info = {
+            'namespaceName': 'test-namespace-with-dashes',
+            'namespaceNameSuffix': 'test.suffix.com',
+            'hybridConnectionName': 'connection_with_underscores',
+            'accessKey': 'key+with/special=chars',
+            'expiresOn': '2025-12-31T23:59:59.999Z',
+            'serviceConfigurationToken': 'token-with-various_special.chars'
+        }
+        
+        # Act
+        result = connectivity_utils.format_relay_info_string(relay_info)
+        
+        # Assert
+        decoded_bytes = base64.b64decode(result.encode('ascii'))
+        decoded_string = decoded_bytes.decode('ascii')
+        parsed_result = json.loads(decoded_string)
+        
+        relay_data = parsed_result['relay']
+        self.assertEqual(relay_data['namespaceName'], 'test-namespace-with-dashes')
+        self.assertEqual(relay_data['accessKey'], 'key+with/special=chars')
+
+    def test_format_relay_info_string_unicode_characters(self):
+        """Test format_relay_info_string with unicode characters."""
+        # Arrange
+        relay_info = {
+            'namespaceName': 'test-unicode-αβγ',
+            'namespaceNameSuffix': 'suffix-δεζ',
+            'hybridConnectionName': 'connection-ηθι',
+            'accessKey': 'key-κλμ',
+            'expiresOn': '2025-06-15T12:30:45Z',
+            'serviceConfigurationToken': 'token-νξο'
+        }
+        
+        # Act
+        result = connectivity_utils.format_relay_info_string(relay_info)
+        
+        # Assert
+        self.assertIsInstance(result, str)
+        
+        # Verify roundtrip encoding/decoding works
+        decoded_bytes = base64.b64decode(result.encode('ascii'))
+        decoded_string = decoded_bytes.decode('ascii')
+        parsed_result = json.loads(decoded_string)
+        
+        relay_data = parsed_result['relay']
+        self.assertEqual(relay_data['namespaceName'], 'test-unicode-αβγ')
+
+    def test_format_relay_info_string_empty_values(self):
+        """Test format_relay_info_string with empty string values."""
+        # Arrange
+        relay_info = {
+            'namespaceName': '',
+            'namespaceNameSuffix': '',
+            'hybridConnectionName': '',
+            'accessKey': '',
+            'expiresOn': '',
+            'serviceConfigurationToken': ''
+        }
+        
+        # Act
+        result = connectivity_utils.format_relay_info_string(relay_info)
+        
+        # Assert
+        decoded_bytes = base64.b64decode(result.encode('ascii'))
+        decoded_string = decoded_bytes.decode('ascii')
+        parsed_result = json.loads(decoded_string)
+        
+        relay_data = parsed_result['relay']
+        for key in relay_data:
+            self.assertEqual(relay_data[key], '')
+
+    def test_format_relay_info_string_large_values(self):
+        """Test format_relay_info_string with large string values."""
+        # Arrange
+        large_string = 'x' * 1000  # 1000 character string
+        relay_info = {
+            'namespaceName': large_string,
+            'namespaceNameSuffix': large_string,
+            'hybridConnectionName': large_string,
+            'accessKey': large_string,
+            'expiresOn': '2025-07-04T10:00:00Z',
+            'serviceConfigurationToken': large_string
+        }
+        
+        # Act
+        result = connectivity_utils.format_relay_info_string(relay_info)
+        
+        # Assert
+        self.assertIsInstance(result, str)
+        decoded_bytes = base64.b64decode(result.encode('ascii'))
+        decoded_string = decoded_bytes.decode('ascii')
+        parsed_result = json.loads(decoded_string)
+        
+        relay_data = parsed_result['relay']
+        self.assertEqual(relay_data['namespaceName'], large_string)
+
+    @mock.patch('json.dumps')
+    def test_format_relay_info_string_json_error_handling(self, mock_dumps):
+        """Test format_relay_info_string handles JSON serialization errors."""
+        # Arrange
+        mock_dumps.side_effect = TypeError("Object not serializable")
+        relay_info = {
+            'namespaceName': 'test',
+            'namespaceNameSuffix': 'test',
+            'hybridConnectionName': 'test',
+            'accessKey': 'test',
+            'expiresOn': 'test',
+            'serviceConfigurationToken': 'test'
+        }
+        
+        # Act & Assert
+        with self.assertRaises(TypeError):
+            connectivity_utils.format_relay_info_string(relay_info)
+
+    def test_format_relay_info_string_structure_validation(self):
+        """Test that format_relay_info_string produces the expected JSON structure."""
+        # Arrange
+        relay_info = {
+            'namespaceName': 'test-namespace',
+            'namespaceNameSuffix': 'servicebus.windows.net',
+            'hybridConnectionName': 'test-connection',
+            'accessKey': 'test-access-key',
+            'expiresOn': '2025-07-04T10:00:00Z',
+            'serviceConfigurationToken': 'test-token'
+        }
+        
+        # Act
+        result = connectivity_utils.format_relay_info_string(relay_info)
+        
+        # Assert - Verify the exact structure matches expectations
+        decoded_bytes = base64.b64decode(result.encode('ascii'))
+        decoded_string = decoded_bytes.decode('ascii')
+        parsed_result = json.loads(decoded_string)
+        
+        # Check top-level structure
+        self.assertEqual(list(parsed_result.keys()), ['relay'])
+        
+        # Check relay object has all required fields
+        relay_data = parsed_result['relay']
+        expected_keys = {
+            'namespaceName', 'namespaceNameSuffix', 'hybridConnectionName',
+            'accessKey', 'expiresOn', 'serviceConfigurationToken'
+        }
+        self.assertEqual(set(relay_data.keys()), expected_keys)
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/src/sftp/azext_sftp/tests/latest/test_file_utils.py b/src/sftp/azext_sftp/tests/latest/test_file_utils.py
diff --git a/src/sftp/azext_sftp/tests/latest/test_rsa_parser.py b/src/sftp/azext_sftp/tests/latest/test_rsa_parser.py
diff --git a/src/sftp/azext_sftp/tests/latest/test_validators.py b/src/sftp/azext_sftp/tests/latest/test_validators.py
