From 09ad2eebe4c5bea1efc83e875cac01d43541d881 Mon Sep 17 00:00:00 2001 From: Renat Akhmerov Date: Mon, 26 Nov 2018 14:51:53 +0700 Subject: [PATCH] Fix the tests for workflow "started_at" and "finished_at" times * There was a race condition in the test that expected different times for two workflow runs (the second one is a rerun) although there was a chance that they could be the same. A delay between two runs was added to fix that. * Delays of some workflow tasks in the tests were too long. No need to do. It only increases test execution times w/o adding any real value. * Workflow definitions were moved into the tests where they're used. It improves readability. * Fixed style (blank lines, naming, etc.) Change-Id: Id61bc612014fde600366b1230ff073930a7970ce --- .../engine/test_task_started_finished_at.py | 183 ++++++++++-------- 1 file changed, 104 insertions(+), 79 deletions(-) diff --git a/mistral/tests/unit/engine/test_task_started_finished_at.py b/mistral/tests/unit/engine/test_task_started_finished_at.py index ae4d7656d..68280a520 100644 --- a/mistral/tests/unit/engine/test_task_started_finished_at.py +++ b/mistral/tests/unit/engine/test_task_started_finished_at.py @@ -10,150 +10,175 @@ # See the License for the specific language governing permissions and # limitations under the License. +import eventlet + from mistral.tests.unit.engine import base from mistral.db.v2 import api as db_api from mistral.services import workflows as wf_service -WF_FAIL_WITH_WAIT = """ -version: "2.0" - -wf_fail: - tasks: - task1: - action: std.fail - wait-before: 2 -""" - -WF_JOIN_ALL = """ -version: "2.0" - -wf_join: - tasks: - wait_2: - action: std.sleep seconds=2 - on-success: finish_wf - wait_4: - action: std.sleep seconds=4 - on-success: finish_wf - wait_6: - action: std.sleep seconds=6 - on-success: finish_wf - finish_wf: - join: all - action: std.sleep seconds=2 -""" - -WF_WITH_RETRIES = """ -version: "2.0" - -wf_retry: - tasks: - task1: - action: std.fail - retry: - delay: 1 - count: 5 -""" - -WF_WAIT_BEFORE_AFTER = """ -version: "2.0" - -wf_wait: - tasks: - task1: - action: std.noop - wait-before: 3 - wait-after: 3 -""" - class TaskStartedFinishedAtTest(base.EngineTestCase): - def setUp(self): super(TaskStartedFinishedAtTest, self).setUp() def test_started_finished_fields_updated_after_rerun(self): - wf_service.create_workflows(WF_FAIL_WITH_WAIT) - wf_ex = self.engine.start_workflow('wf_fail') + wf_text = """ + version: '2.0' + + wf: + tasks: + task1: + action: std.fail + wait-before: 2 + """ + + wf_service.create_workflows(wf_text) + + wf_ex = self.engine.start_workflow('wf') + self.await_workflow_error(wf_ex.id) - task_ex = self._get_task_from_wf(wf_ex.id) + task_ex = self._extract_task_ex(wf_ex.id) + started_1st, finished_1st = self._get_started_finished(task_ex) + # Make sure to rerun the workflow after a certain delay so that + # times for the first run are different from times in the second run. + eventlet.sleep(1) + wf_ex = self.engine.rerun_workflow(task_ex.id) - task_ex = self._get_task_from_wf(wf_ex.id) - self.assertIsNone(task_ex.finished_at) self.await_workflow_error(wf_ex.id) - task_ex = self._get_task_from_wf(wf_ex.id) + task_ex = self._extract_task_ex(wf_ex.id) + started_2nd, finished_2nd = self._get_started_finished(task_ex) self.assertNotEqual(started_1st, started_2nd) self.assertNotEqual(finished_1st, finished_2nd) def test_correct_duration_in_case_of_join_all(self): - wf_service.create_workflows(WF_JOIN_ALL) - wf_ex = self.engine.start_workflow('wf_join') + wf_text = """ + version: '2.0' + + wf: + tasks: + task1: + action: std.sleep seconds=1 + on-success: join_task + + task2: + action: std.sleep seconds=2 + on-success: join_task + + task3: + action: std.sleep seconds=3 + on-success: join_task + + join_task: + join: all + action: std.sleep seconds=1 + """ + + wf_service.create_workflows(wf_text) + + wf_ex = self.engine.start_workflow('wf') + self.await_workflow_success(wf_ex.id) - wait_2 = self._get_task_from_wf(wf_ex.id, 'wait_2') - wait_4 = self._get_task_from_wf(wf_ex.id, 'wait_4') - wait_6 = self._get_task_from_wf(wf_ex.id, 'wait_6') - finish_wf = self._get_task_from_wf(wf_ex.id, 'finish_wf') + task1_ex = self._extract_task_ex(wf_ex.id, 'task1') + task2_ex = self._extract_task_ex(wf_ex.id, 'task2') + task3_ex = self._extract_task_ex(wf_ex.id, 'task3') - self._check_was_started_after(finish_wf, wait_2) - self._check_was_started_after(finish_wf, wait_4) - self._check_was_started_after(finish_wf, wait_6) + join_task_ex = self._extract_task_ex(wf_ex.id, 'join_task') + + self._check_started_after(join_task_ex, task1_ex) + self._check_started_after(join_task_ex, task2_ex) + self._check_started_after(join_task_ex, task3_ex) def test_retries_do_not_update_created_at(self): - wf_service.create_workflows(WF_WITH_RETRIES) - wf_ex = self.engine.start_workflow('wf_retry') + wf_text = """ + version: '2.0' + + wf: + tasks: + task1: + action: std.fail + retry: + delay: 1 + count: 5 + """ + + wf_service.create_workflows(wf_text) + + wf_ex = self.engine.start_workflow('wf') + self.await_workflow_error(wf_ex.id) - task_ex = self._get_task_from_wf(wf_ex.id) + task_ex = self._extract_task_ex(wf_ex.id) + created_at = task_ex.created_at started_at = self._get_started_finished(task_ex)[0] self.assertEqual(created_at, started_at) def test_wait_before_after_are_included_to_duration(self): - wf_service.create_workflows(WF_WAIT_BEFORE_AFTER) - wf_ex = self.engine.start_workflow('wf_wait') + wf_text = """ + version: '2.0' + + wf: + tasks: + task1: + action: std.noop + wait-before: 1 + wait-after: 2 + """ + + wf_service.create_workflows(wf_text) + + wf_ex = self.engine.start_workflow('wf') + self.await_workflow_success(wf_ex.id) - task_ex = self._get_task_from_wf(wf_ex.id) + task_ex = self._extract_task_ex(wf_ex.id) + started, finished = self._get_started_finished(task_ex) + duration = self._get_task_duration(started, finished) self._check_duration_more_than(duration, 1) - def _get_task_from_wf(self, wf_ex_id, name='task1'): + def _extract_task_ex(self, wf_ex_id, name='task1'): with db_api.transaction(): wf_ex = db_api.get_workflow_execution(wf_ex_id) + task_execs = wf_ex.task_executions + return self._assert_single_item(task_execs, name=name) def _get_started_finished(self, task_ex): started_at = task_ex.started_at finished_at = task_ex.finished_at + self.assertIsNotNone(started_at) self.assertIsNotNone(finished_at) + return started_at, finished_at def _get_task_duration(self, start_time, finish_time): return (finish_time - start_time).total_seconds() - def _check_was_started_after(self, task_started, task_finished): - first_finished = self._get_started_finished(task_finished)[1] - second_started = self._get_started_finished(task_started)[0] + def _check_started_after(self, task_ex1, task_ex2): + first_finished = self._get_started_finished(task_ex2)[1] + second_started = self._get_started_finished(task_ex1)[0] + delta = self._get_task_duration(first_finished, second_started) self.assertTrue( delta >= 0, "Expected {} was started after {} was finished".format( - task_started.name, task_finished.name) + task_ex1.name, task_ex2.name) ) def _check_duration_more_than(self, duration, time):