

@mock.patch.object(task.LOG, 'warn')
def test_update_progress_upper_bound(self, mocked_warn):
    result = []

    def progress_callback(event_type, details):
        result.append(details.pop('progress'))
    a_task = ProgressTask()
    a_task.notifier.register(task.EVENT_UPDATE_PROGRESS, progress_callback)
    a_task.execute([1.0, 1.5, 2.0])
    self.assertEqual([1.0, 1.0, 1.0], result)
    self.assertEqual(2, mocked_warn.call_count)

dataset/ETHPy150Open openstack/taskflow/taskflow/tests/unit/test_task.py TaskTest.test_update_progress_upper_bound/original