(self)
| 24 | |
| 25 | class WiringTest(base.TestWorkflowExecution): |
| 26 | def test_sequential(self): |
| 27 | wf_name = "examples.orquesta-sequential" |
| 28 | wf_input = {"name": "Thanos"} |
| 29 | |
| 30 | expected_output = {"greeting": "Thanos, All your base are belong to us!"} |
| 31 | expected_result = {"output": expected_output} |
| 32 | |
| 33 | ex = self._execute_workflow(wf_name, wf_input) |
| 34 | ex = self._wait_for_completion(ex) |
| 35 | |
| 36 | self.assertEqual(ex.status, ac_const.LIVEACTION_STATUS_SUCCEEDED) |
| 37 | self.assertDictEqual(ex.result, expected_result) |
| 38 | |
| 39 | def test_join(self): |
| 40 | wf_name = "examples.orquesta-join" |
nothing calls this directly
no test coverage detected