Serialisation deserialisation tests for python sdk

Author: orkes-harshil

requirements.txt

@@ -5,4 +5,5 @@ requests >= 2.31.0
 typing-extensions >= 4.2.0
 astor >= 0.8.1
 shortuuid >= 1.0.11
-dacite >= 1.8.1
+dacite >= 1.8.1
+deprecated>=1.2.13

tests/serdesertest/__init__.py

@@ -0,0 +1,64 @@
+import unittest
+import json
+from serdesertest.util.serdeser_json_resolver_utility import JsonTemplateResolver
+
+# Import the classes being tested
+from conductor.client.http.models.authorization_request import AuthorizationRequest
+
+
+class TestAuthorizationRequestSerDes(unittest.TestCase):
+    """
+    Unit tests for serialization and deserialization of AuthorizationRequest model.
+    """
+
+    def setUp(self):
+        """Set up test fixtures."""
+        # Load the template JSON for testing
+        self.server_json_str = JsonTemplateResolver.get_json_string("AuthorizationRequest")
+        self.server_json = json.loads(self.server_json_str)
+
+    def test_serialization_deserialization(self):
+        """Test complete serialization and deserialization process."""
+        # Create the authorization request object directly from JSON
+        # The model's __init__ should handle the nested objects
+        auth_request = AuthorizationRequest(
+            subject=self.server_json.get('subject'),
+            target=self.server_json.get('target'),
+            access=self.server_json.get('access')
+        )
+
+        # Verify model is properly initialized
+        self.assertIsNotNone(auth_request, "Deserialized object should not be null")
+
+        # Verify access list
+        self.assertIsNotNone(auth_request.access, "Access list should not be null")
+        self.assertTrue(all(access in ["CREATE", "READ", "UPDATE", "DELETE", "EXECUTE"]
+                            for access in auth_request.access))
+
+        # Verify subject and target are present
+        self.assertIsNotNone(auth_request.subject, "Subject should not be null")
+        self.assertIsNotNone(auth_request.target, "Target should not be null")
+
+        # Serialize back to dictionary
+        result_dict = auth_request.to_dict()
+
+        # Verify structure matches the original
+        self.assertEqual(
+            set(self.server_json.keys()),
+            set(result_dict.keys()),
+            "Serialized JSON should have the same keys as the original"
+        )
+
+        # Convert both to JSON strings and compare (similar to objectMapper.readTree)
+        original_json_normalized = json.dumps(self.server_json, sort_keys=True)
+        result_json_normalized = json.dumps(result_dict, sort_keys=True)
+
+        self.assertEqual(
+            original_json_normalized,
+            result_json_normalized,
+            "Serialized JSON should match the original SERVER_JSON"
+        )
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/serdesertest/authorization_request_serdeser_test.py

@@ -0,0 +1,103 @@
+import unittest
+import json
+
+from conductor.client.http.models import BulkResponse
+from serdesertest.util.serdeser_json_resolver_utility import JsonTemplateResolver
+
+
+class TestBulkResponseSerDeser(unittest.TestCase):
+    """Test serialization and deserialization for BulkResponse class"""
+
+    def setUp(self):
+        """Set up test fixtures"""
+        # Load test data from template
+        self.server_json_str = JsonTemplateResolver.get_json_string("BulkResponse")
+        # Parse into dictionary for comparisons
+        self.server_json_dict = json.loads(self.server_json_str)
+
+    def test_bulk_response_serialization_deserialization(self):
+        """Comprehensive test for serialization and deserialization of BulkResponse"""
+        # 1. Deserialize JSON into model object
+        bulk_response = BulkResponse(
+            bulk_error_results=self.server_json_dict['bulkErrorResults'],
+            bulk_successful_results=self.server_json_dict['bulkSuccessfulResults']
+        )
+
+        # 2. Verify BulkResponse object properties and types
+        self.assertIsInstance(bulk_response, BulkResponse)
+        self.assertIsInstance(bulk_response.bulk_error_results, dict)
+        self.assertIsInstance(bulk_response.bulk_successful_results, list)
+
+        # 3. Validate content of properties
+        for key, value in bulk_response.bulk_error_results.items():
+            self.assertIsInstance(key, str)
+            self.assertIsInstance(value, str)
+
+        # Validate the structure of items in bulk_successful_results
+        # This adapts to the actual structure found in the template
+        for item in bulk_response.bulk_successful_results:
+            # If the items are dictionaries with 'value' keys
+            if isinstance(item, dict) and 'value' in item:
+                self.assertIsInstance(item['value'], str)
+            # If the items are strings
+            elif isinstance(item, str):
+                pass
+            else:
+                self.fail(f"Unexpected item type in bulk_successful_results: {type(item)}")
+
+        # 4. Verify values match original source
+        self.assertEqual(bulk_response.bulk_error_results, self.server_json_dict['bulkErrorResults'])
+        self.assertEqual(bulk_response.bulk_successful_results, self.server_json_dict['bulkSuccessfulResults'])
+
+        # 5. Test serialization back to dictionary
+        result_dict = bulk_response.to_dict()
+        self.assertIn('bulk_error_results', result_dict)
+        self.assertIn('bulk_successful_results', result_dict)
+        self.assertEqual(result_dict['bulk_error_results'], self.server_json_dict['bulkErrorResults'])
+        self.assertEqual(result_dict['bulk_successful_results'], self.server_json_dict['bulkSuccessfulResults'])
+
+        # 6. Test serialization to JSON-compatible dictionary (with camelCase keys)
+        json_compatible_dict = {
+            'bulkErrorResults': result_dict['bulk_error_results'],
+            'bulkSuccessfulResults': result_dict['bulk_successful_results']
+        }
+
+        # 7. Normalize dictionaries for comparison (handles differences in ordering)
+        normalized_original = json.loads(json.dumps(self.server_json_dict, sort_keys=True))
+        normalized_result = json.loads(json.dumps(json_compatible_dict, sort_keys=True))
+        self.assertEqual(normalized_original, normalized_result)
+
+        # 8. Test full serialization/deserialization round trip
+        bulk_response_2 = BulkResponse(
+            bulk_error_results=result_dict['bulk_error_results'],
+            bulk_successful_results=result_dict['bulk_successful_results']
+        )
+        self.assertEqual(bulk_response.bulk_error_results, bulk_response_2.bulk_error_results)
+        self.assertEqual(bulk_response.bulk_successful_results, bulk_response_2.bulk_successful_results)
+
+        # 9. Test with missing fields
+        bulk_response_errors_only = BulkResponse(
+            bulk_error_results={"id1": "error1"}
+        )
+        self.assertEqual(bulk_response_errors_only.bulk_error_results, {"id1": "error1"})
+        self.assertIsNone(bulk_response_errors_only.bulk_successful_results)
+
+        # Create a structure similar to what's in the template
+        sample_successful_result = [{"value": "success1"}]
+        bulk_response_success_only = BulkResponse(
+            bulk_successful_results=sample_successful_result
+        )
+        self.assertIsNone(bulk_response_success_only.bulk_error_results)
+        self.assertEqual(bulk_response_success_only.bulk_successful_results, sample_successful_result)
+
+        # 10. Test with empty fields
+        bulk_response_empty = BulkResponse(
+            bulk_error_results={},
+            bulk_successful_results=[]
+        )
+        self.assertEqual(bulk_response_empty.bulk_error_results, {})
+        self.assertEqual(bulk_response_empty.bulk_successful_results, [])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/serdesertest/bulk_response_serdeser_test.py

@@ -0,0 +1,46 @@
+import unittest
+import json
+
+from conductor.client.http.models import ConductorApplication
+from serdesertest.util.serdeser_json_resolver_utility import JsonTemplateResolver
+
+
+
+class TestConductorApplicationSerialization(unittest.TestCase):
+    """Test case for ConductorApplication serialization and deserialization."""
+
+    def setUp(self):
+        """Set up test fixtures before each test."""
+        # Load JSON template from the resolver utility
+        self.server_json_str = JsonTemplateResolver.get_json_string("ConductorApplication")
+        self.server_json = json.loads(self.server_json_str)
+
+    def test_serialization_deserialization(self):
+        """Test that validates the serialization and deserialization of ConductorApplication model."""
+
+        # Step 1: Deserialize server JSON into SDK model object
+        # Create model object using constructor with fields from the JSON
+        conductor_app = ConductorApplication(
+            id=self.server_json.get('id'),
+            name=self.server_json.get('name'),
+            created_by=self.server_json.get('createdBy')
+        )
+
+        # Step 2: Verify all fields are correctly populated
+        self.assertEqual(conductor_app.id, self.server_json.get('id'))
+        self.assertEqual(conductor_app.name, self.server_json.get('name'))
+        self.assertEqual(conductor_app.created_by, self.server_json.get('createdBy'))
+
+        # Step 3: Serialize the model back to JSON
+        serialized_json = conductor_app.to_dict()
+
+        # Step 4: Verify the serialized JSON matches the original
+        # Note: Field names in serialized_json will be in snake_case
+        self.assertEqual(serialized_json.get('id'), self.server_json.get('id'))
+        self.assertEqual(serialized_json.get('name'), self.server_json.get('name'))
+        # Handle the camelCase to snake_case transformation
+        self.assertEqual(serialized_json.get('created_by'), self.server_json.get('createdBy'))
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/serdesertest/conductor_application_serdeser_test.py

@@ -0,0 +1,104 @@
+import json
+import unittest
+
+from conductor.client.http.models import ConductorUser, Role, Group
+from serdesertest.util.serdeser_json_resolver_utility import JsonTemplateResolver
+
+
+class TestConductorUserSerDeSer(unittest.TestCase):
+    """Test serialization and deserialization of ConductorUser."""
+
+    def setUp(self):
+        # Load JSON template using the utility
+        self.server_json_str = JsonTemplateResolver.get_json_string("ConductorUser")
+        self.server_json = json.loads(self.server_json_str)
+
+    def test_conductor_user_serde(self):
+        """Test that ConductorUser can be deserialized from server JSON and serialized back without data loss."""
+
+        # 1. Deserialize server JSON into ConductorUser object
+        conductor_user = ConductorUser()
+        conductor_user_dict = self.server_json
+
+        # Set attributes from deserialized JSON
+        if 'id' in conductor_user_dict:
+            conductor_user.id = conductor_user_dict['id']
+        if 'name' in conductor_user_dict:
+            conductor_user.name = conductor_user_dict['name']
+        if 'roles' in conductor_user_dict:
+            # Assuming Role has a from_dict method or similar
+            roles_list = []
+            for role_data in conductor_user_dict['roles']:
+                role = Role()  # Create a Role object based on your actual implementation
+                # Set Role properties here
+                roles_list.append(role)
+            conductor_user.roles = roles_list
+        if 'groups' in conductor_user_dict:
+            # Assuming Group has a from_dict method or similar
+            groups_list = []
+            for group_data in conductor_user_dict['groups']:
+                group = Group()  # Create a Group object based on your actual implementation
+                # Set Group properties here
+                groups_list.append(group)
+            conductor_user.groups = groups_list
+        if 'uuid' in conductor_user_dict:
+            conductor_user.uuid = conductor_user_dict['uuid']
+        if 'applicationUser' in conductor_user_dict:
+            conductor_user.application_user = conductor_user_dict['applicationUser']
+        if 'encryptedId' in conductor_user_dict:
+            conductor_user.encrypted_id = conductor_user_dict['encryptedId']
+        if 'encryptedIdDisplayValue' in conductor_user_dict:
+            conductor_user.encrypted_id_display_value = conductor_user_dict['encryptedIdDisplayValue']
+
+        # 2. Verify all fields are properly populated
+        expected_id = self.server_json.get('id', None)
+        self.assertEqual(conductor_user.id, expected_id)
+
+        expected_name = self.server_json.get('name', None)
+        self.assertEqual(conductor_user.name, expected_name)
+
+        # Verify lists
+        if 'roles' in self.server_json:
+            self.assertEqual(len(conductor_user.roles), len(self.server_json['roles']))
+
+        if 'groups' in self.server_json:
+            self.assertEqual(len(conductor_user.groups), len(self.server_json['groups']))
+
+        expected_uuid = self.server_json.get('uuid', None)
+        self.assertEqual(conductor_user.uuid, expected_uuid)
+
+        expected_app_user = self.server_json.get('applicationUser', None)
+        self.assertEqual(conductor_user.application_user, expected_app_user)
+
+        expected_encrypted_id = self.server_json.get('encryptedId', None)
+        self.assertEqual(conductor_user.encrypted_id, expected_encrypted_id)
+
+        expected_encrypted_id_display = self.server_json.get('encryptedIdDisplayValue', None)
+        self.assertEqual(conductor_user.encrypted_id_display_value, expected_encrypted_id_display)
+
+        # 3. Serialize the object back to JSON
+        serialized_json = conductor_user.to_dict()
+
+        # 4. Verify the serialized JSON matches the original
+        # Handle camelCase to snake_case transformations
+        if 'applicationUser' in self.server_json:
+            self.assertEqual(serialized_json['application_user'], self.server_json['applicationUser'])
+        if 'encryptedId' in self.server_json:
+            self.assertEqual(serialized_json['encrypted_id'], self.server_json['encryptedId'])
+        if 'encryptedIdDisplayValue' in self.server_json:
+            self.assertEqual(serialized_json['encrypted_id_display_value'], self.server_json['encryptedIdDisplayValue'])
+
+        # Check common fields that don't need transformation
+        for field in ['id', 'name', 'uuid']:
+            if field in self.server_json:
+                self.assertEqual(serialized_json[field], self.server_json[field])
+
+        # Check lists length
+        if 'roles' in self.server_json:
+            self.assertEqual(len(serialized_json['roles']), len(self.server_json['roles']))
+        if 'groups' in self.server_json:
+            self.assertEqual(len(serialized_json['groups']), len(self.server_json['groups']))
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/serdesertest/conductor_user_serdeser_test.py

@@ -0,0 +1,58 @@
+import unittest
+import json
+
+from conductor.client.http.models.correlation_ids_search_request import CorrelationIdsSearchRequest
+from serdesertest.util.serdeser_json_resolver_utility import JsonTemplateResolver
+
+class TestCorrelationIdsSearchRequest(unittest.TestCase):
+    """Test case for CorrelationIdsSearchRequest class."""
+
+    def setUp(self):
+        """Set up test fixtures."""
+        # Load the JSON template for CorrelationIdsSearchRequest
+        self.server_json_str = JsonTemplateResolver.get_json_string("CorrelationIdsSearchRequest")
+        self.server_json = json.loads(self.server_json_str)
+
+        # Convert camelCase to snake_case for initialization
+        self.python_format_json = {}
+        for key, value in self.server_json.items():
+            # Use the attribute_map to find the Python property name
+            python_key = next((k for k, v in CorrelationIdsSearchRequest.attribute_map.items() if v == key), key)
+            self.python_format_json[python_key] = value
+
+    def test_serdeser_correlation_ids_search_request(self):
+        """Test serialization and deserialization of CorrelationIdsSearchRequest."""
+        # 1. Server JSON can be correctly deserialized into SDK model object
+        model_obj = CorrelationIdsSearchRequest(**self.python_format_json)
+
+        # 2. All fields are properly populated during deserialization
+        # Check correlation_ids (list[str])
+        self.assertIsNotNone(model_obj.correlation_ids)
+        self.assertIsInstance(model_obj.correlation_ids, list)
+        for item in model_obj.correlation_ids:
+            self.assertIsInstance(item, str)
+
+        # Check workflow_names (list[str])
+        self.assertIsNotNone(model_obj.workflow_names)
+        self.assertIsInstance(model_obj.workflow_names, list)
+        for item in model_obj.workflow_names:
+            self.assertIsInstance(item, str)
+
+        # 3. The SDK model can be serialized back to JSON
+        serialized_dict = model_obj.to_dict()
+
+        # 4. The resulting JSON matches the original
+        # Convert serialized dict keys to camelCase for comparison
+        json_dict = {}
+        for attr, value in serialized_dict.items():
+            if attr in model_obj.attribute_map:
+                json_dict[model_obj.attribute_map[attr]] = value
+            else:
+                json_dict[attr] = value
+
+        # Compare with original JSON
+        self.assertEqual(self.server_json, json_dict)
+
+
+if __name__ == '__main__':
+    unittest.main()