Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use describe_workflow_type method in localstack

Best Python code snippet using localstack_python

test_workflow.py

Source:test_workflow.py

...47    def test_workflow_type__diff_with_identical_workflow_type(self):48        with patch.object(49            Layer1, "describe_workflow_type", mock_describe_workflow_type,50        ):51            mocked = mock_describe_workflow_type()52            workflow_type = WorkflowType(53                self.domain,54                name=mocked["typeInfo"]["workflowType"]["name"],55                version=mocked["typeInfo"]["workflowType"]["version"],56                status=mocked["typeInfo"]["status"],57                creation_date=mocked["typeInfo"]["creationDate"],58                deprecation_date=mocked["typeInfo"]["deprecationDate"],59                task_list=mocked["configuration"]["defaultTaskList"]["name"],60                child_policy=mocked["configuration"]["defaultChildPolicy"],61                execution_timeout=mocked["configuration"][62                    "defaultExecutionStartToCloseTimeout"63                ],64                decision_tasks_timeout=mocked["configuration"][65                    "defaultTaskStartToCloseTimeout"66                ],67                description=mocked["typeInfo"]["description"],68            )69            diffs = workflow_type._diff()70            self.assertLength(diffs, 0)71    def test_exists_with_existing_workflow_type(self):72        with patch.object(Layer1, "describe_workflow_type"):73            self.assertTrue(self.wt.exists)74    def test_exists_with_non_existent_workflow_type(self):75        with patch.object(self.wt.connection, "describe_workflow_type") as mock:76            mock.side_effect = SWFResponseError(77                400,78                "Bad Request:",79                {80                    "__type": "com.amazonaws.swf.base.model#UnknownResourceFault",81                    "message": "Unknown type: WorkflowType=[workflowId=blah, runId=test]",82                },83                "UnknownResourceFault",84            )85            self.assertFalse(self.wt.exists)86    # TODO: fix test when no network (probably hits real SWF endpoints)87    @unittest.skip("Skip it in case there's no network connection.")88    def test_workflow_type_exists_with_whatever_error(self):89        with patch.object(self.wt.connection, "describe_workflow_type") as mock:90            with self.assertRaises(ResponseError):91                mock.side_effect = SWFResponseError(92                    400,93                    "mocking exception",94                    {"__type": "WhateverError", "message": "Whatever"},95                )96                dummy = self.domain.exists97    def test_is_synced_with_unsynced_workflow_type(self):98        pass99    def test_is_synced_with_synced_workflow_type(self):100        pass101    def test_is_synced_over_non_existent_workflow_type(self):102        with patch.object(103            Layer1, "describe_workflow_type", mock_describe_workflow_type104        ):105            workflow_type = WorkflowType(106                self.domain,107                "non-existent-workflow-type",108                version="non-existent-version",109            )110            self.assertFalse(workflow_type.is_synced)111    def test_changes_with_different_workflow_type(self):112        with patch.object(113            Layer1, "describe_workflow_type", mock_describe_workflow_type,114        ):115            workflow_type = WorkflowType(116                self.domain,117                "different-workflow-type-type",118                version="different-workflow-type-type-version",119            )120            diffs = workflow_type.changes121            self.assertIsNotNone(diffs)122            self.assertLength(diffs, 6)123            self.assertTrue(hasattr(diffs[0], "attr"))124            self.assertTrue(hasattr(diffs[0], "local"))125            self.assertTrue(hasattr(diffs[0], "upstream"))126    def test_workflow_type_changes_with_identical_workflow_type(self):127        with patch.object(128            Layer1, "describe_workflow_type", mock_describe_workflow_type,129        ):130            mocked = mock_describe_workflow_type()131            workflow_type = WorkflowType(132                self.domain,133                name=mocked["typeInfo"]["workflowType"]["name"],134                version=mocked["typeInfo"]["workflowType"]["version"],135                status=mocked["typeInfo"]["status"],136                creation_date=mocked["typeInfo"]["creationDate"],137                deprecation_date=mocked["typeInfo"]["deprecationDate"],138                task_list=mocked["configuration"]["defaultTaskList"]["name"],139                child_policy=mocked["configuration"]["defaultChildPolicy"],140                execution_timeout=mocked["configuration"][141                    "defaultExecutionStartToCloseTimeout"142                ],143                decision_tasks_timeout=mocked["configuration"][144                    "defaultTaskStartToCloseTimeout"...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.