Rewrite serdesertest/ to pytest (#316)

Author: IgorChvyrov-sm

tests/serdesertest/test_serdeser_action.py

@@ -1,131 +1,114 @@
-import unittest
+import json
+import re
+
+import pytest
+
 from conductor.client.http.models.action import Action
 from conductor.client.http.models.start_workflow import StartWorkflow
 from conductor.client.http.models.task_details import TaskDetails
 from conductor.client.http.models.terminate_workflow import TerminateWorkflow
-from conductor.client.http.models.update_workflow_variables import UpdateWorkflowVariables
+from conductor.client.http.models.update_workflow_variables import (
+    UpdateWorkflowVariables,
+)
 from tests.serdesertest.util.serdeser_json_resolver_utility import JsonTemplateResolver
-import json
-import re
-
-
-class TestActionSerDes(unittest.TestCase):
-    def setUp(self):
-        # Load the JSON template
-        self.server_json_str = JsonTemplateResolver.get_json_string("EventHandler.Action")
-        self.server_json = json.loads(self.server_json_str)
-
-    def _camel_to_snake(self, name):
-        """Convert camelCase to snake_case"""
-        # Insert underscore before uppercase letters and convert to lowercase
-        s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
-        return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()
-
-    def _create_model_object(self, model_class, json_data):
-        """Generic method to create model objects with proper camelCase to snake_case conversion"""
-        if not json_data:
-            return None
-
-        # Create an instance of the model class
-        obj = model_class()
-
-        # Iterate through the JSON data and set attributes on the model object
-        for key, value in json_data.items():
-            # Convert camelCase to snake_case
-            snake_key = self._camel_to_snake(key)
-
-            # Try to set the attribute if it exists on the model
-            if hasattr(obj, snake_key):
-                setattr(obj, snake_key, value)
-
-        return obj
-
-    def test_action_serdes(self):
-        # 1. Test deserialization of server JSON to SDK model
-        action_obj = Action(
-            action=self.server_json.get("action"),
-            start_workflow=self._create_model_object(StartWorkflow, self.server_json.get("start_workflow")),
-            complete_task=self._create_model_object(TaskDetails, self.server_json.get("complete_task")),
-            fail_task=self._create_model_object(TaskDetails, self.server_json.get("fail_task")),
-            expand_inline_json=self.server_json.get("expandInlineJSON"),
-            terminate_workflow=self._create_model_object(TerminateWorkflow, self.server_json.get("terminate_workflow")),
-            update_workflow_variables=self._create_model_object(UpdateWorkflowVariables,
-                                                                self.server_json.get("update_workflow_variables"))
-        )
-
-        # 2. Verify all fields are properly populated
-        self.assertEqual(self.server_json.get("action"), action_obj.action)
-
-        # Check if start_workflow exists in the JSON
-        if "start_workflow" in self.server_json:
-            self.assertIsNotNone(action_obj.start_workflow)
-
-        # Check if complete_task exists in the JSON
-        if "complete_task" in self.server_json:
-            self.assertIsNotNone(action_obj.complete_task)
-
-        # Check if fail_task exists in the JSON
-        if "fail_task" in self.server_json:
-            self.assertIsNotNone(action_obj.fail_task)
-
-        # Check if expandInlineJSON exists in the JSON
-        if "expandInlineJSON" in self.server_json:
-            self.assertEqual(self.server_json.get("expandInlineJSON"), action_obj.expand_inline_json)
-
-        # Check if terminate_workflow exists in the JSON
-        if "terminate_workflow" in self.server_json:
-            self.assertIsNotNone(action_obj.terminate_workflow)
-
-        # Check if update_workflow_variables exists in the JSON
-        if "update_workflow_variables" in self.server_json:
-            self.assertIsNotNone(action_obj.update_workflow_variables)
-
-        # 3. Verify the action enum value is valid
-        allowed_values = ["start_workflow", "complete_task", "fail_task", "terminate_workflow",
-                          "update_workflow_variables"]
-        self.assertIn(action_obj.action, allowed_values)
-
-        # 4. Test serialization back to JSON
-        result_json = action_obj.to_dict()
-
-        # 5. Verify the result JSON has the same keys and values as the original
-        for key in self.server_json:
-            if key == "expandInlineJSON":
-                # Handle camelCase to snake_case conversion
-                self.assertEqual(self.server_json[key], result_json["expand_inline_json"])
-            elif key in ["terminate_workflow", "start_workflow", "complete_task", "fail_task",
-                         "update_workflow_variables"]:
-                # For nested objects, verify they exist in result_json
-                if self.server_json[key] is not None:
-                    self.assertIsNotNone(result_json[key])
-
-                    # For terminate_workflow, check specific camelCase fields
-                    if key == "terminate_workflow" and key in result_json:
-                        term_json = self.server_json[key]
-                        result_term = result_json[key]
-
-                        # Maps the expected field names from server JSON to the result JSON
-                        if "workflowId" in term_json and "workflowId" in result_term:
-                            self.assertEqual(term_json["workflowId"], result_term["workflowId"])
-                        if "terminationReason" in term_json and "terminationReason" in result_term:
-                            self.assertEqual(term_json["terminationReason"], result_term["terminationReason"])
-
-                    # For update_workflow_variables, check specific camelCase fields
-                    if key == "update_workflow_variables" and key in result_json:
-                        update_json = self.server_json[key]
-                        result_update = result_json[key]
-
-                        # Maps the expected field names from server JSON to the result JSON
-                        if "workflowId" in update_json and "workflowId" in result_update:
-                            self.assertEqual(update_json["workflowId"], result_update["workflowId"])
-                        if "variables" in update_json and "variables" in result_update:
-                            self.assertEqual(update_json["variables"], result_update["variables"])
-                        if "appendArray" in update_json and "appendArray" in result_update:
-                            self.assertEqual(update_json["appendArray"], result_update["appendArray"])
-            elif key in result_json:
-                # Direct comparison for keys that match
-                self.assertEqual(self.server_json[key], result_json[key])
 
 
-if __name__ == '__main__':
-    unittest.main()
+def camel_to_snake(name):
+    s1 = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", name)
+    return re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1).lower()
+
+
+def create_model_object(model_class, json_data):
+    if not json_data:
+        return None
+    obj = model_class()
+    for key, value in json_data.items():
+        snake_key = camel_to_snake(key)
+        if hasattr(obj, snake_key):
+            setattr(obj, snake_key, value)
+    return obj
+
+
+@pytest.fixture
+def server_json():
+    return json.loads(JsonTemplateResolver.get_json_string("EventHandler.Action"))
+
+
+def test_action_serdes(server_json):
+    action_obj = Action(
+        action=server_json.get("action"),
+        start_workflow=create_model_object(
+            StartWorkflow, server_json.get("start_workflow")
+        ),
+        complete_task=create_model_object(
+            TaskDetails, server_json.get("complete_task")
+        ),
+        fail_task=create_model_object(TaskDetails, server_json.get("fail_task")),
+        expand_inline_json=server_json.get("expandInlineJSON"),
+        terminate_workflow=create_model_object(
+            TerminateWorkflow, server_json.get("terminate_workflow")
+        ),
+        update_workflow_variables=create_model_object(
+            UpdateWorkflowVariables, server_json.get("update_workflow_variables")
+        ),
+    )
+    assert server_json.get("action") == action_obj.action
+    if "start_workflow" in server_json:
+        assert action_obj.start_workflow is not None
+    if "complete_task" in server_json:
+        assert action_obj.complete_task is not None
+    if "fail_task" in server_json:
+        assert action_obj.fail_task is not None
+    if "expandInlineJSON" in server_json:
+        assert server_json.get("expandInlineJSON") == action_obj.expand_inline_json
+    if "terminate_workflow" in server_json:
+        assert action_obj.terminate_workflow is not None
+    if "update_workflow_variables" in server_json:
+        assert action_obj.update_workflow_variables is not None
+    allowed_values = [
+        "start_workflow",
+        "complete_task",
+        "fail_task",
+        "terminate_workflow",
+        "update_workflow_variables",
+    ]
+    assert action_obj.action in allowed_values
+    result_json = action_obj.to_dict()
+    for key in server_json:
+        if key == "expandInlineJSON":
+            assert server_json[key] == result_json["expand_inline_json"]
+        elif key in [
+            "terminate_workflow",
+            "start_workflow",
+            "complete_task",
+            "fail_task",
+            "update_workflow_variables",
+        ]:
+            if server_json[key] is not None:
+                assert result_json[key] is not None
+                if key == "terminate_workflow" and key in result_json:
+                    term_json = server_json[key]
+                    result_term = result_json[key]
+                    if "workflowId" in term_json and "workflowId" in result_term:
+                        assert term_json["workflowId"] == result_term["workflowId"]
+                    if (
+                        "terminationReason" in term_json
+                        and "terminationReason" in result_term
+                    ):
+                        assert (
+                            term_json["terminationReason"]
+                            == result_term["terminationReason"]
+                        )
+                if key == "update_workflow_variables" and key in result_json:
+                    update_json = server_json[key]
+                    result_update = result_json[key]
+                    if "workflowId" in update_json and "workflowId" in result_update:
+                        assert update_json["workflowId"] == result_update["workflowId"]
+                    if "variables" in update_json and "variables" in result_update:
+                        assert update_json["variables"] == result_update["variables"]
+                    if "appendArray" in update_json and "appendArray" in result_update:
+                        assert (
+                            update_json["appendArray"] == result_update["appendArray"]
+                        )
+        elif key in result_json:
+            assert server_json[key] == result_json[key]

tests/serdesertest/test_serdeser_authorization_request.py

@@ -1,64 +1,36 @@
-import unittest
 import json
-from tests.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")
+import pytest
 
-        # 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"
-        )
+from conductor.client.http.models.authorization_request import AuthorizationRequest
+from tests.serdesertest.util.serdeser_json_resolver_utility import JsonTemplateResolver
 
 
-if __name__ == '__main__':
-    unittest.main()
+@pytest.fixture
+def server_json():
+    return json.loads(JsonTemplateResolver.get_json_string("AuthorizationRequest"))
+
+
+def test_serialization_deserialization(server_json):
+    auth_request = AuthorizationRequest(
+        subject=server_json.get("subject"),
+        target=server_json.get("target"),
+        access=server_json.get("access"),
+    )
+    assert auth_request is not None, "Deserialized object should not be null"
+    assert auth_request.access is not None, "Access list should not be null"
+    assert all(
+        access in ["CREATE", "READ", "UPDATE", "DELETE", "EXECUTE"]
+        for access in auth_request.access
+    )
+    assert auth_request.subject is not None, "Subject should not be null"
+    assert auth_request.target is not None, "Target should not be null"
+    result_dict = auth_request.to_dict()
+    assert set(server_json.keys()) == set(
+        result_dict.keys()
+    ), "Serialized JSON should have the same keys as the original"
+    original_json_normalized = json.dumps(server_json, sort_keys=True)
+    result_json_normalized = json.dumps(result_dict, sort_keys=True)
+    assert (
+        original_json_normalized == result_json_normalized
+    ), "Serialized JSON should match the original SERVER_JSON"