blob: e35e1514d26ee903ff9b1a26dacb0f9e5377251e [file] [log] [blame]
#
# Copyright 2017 the original author or authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from unittest import TestCase, main
from nose.tools import raises
from twisted.internet import defer
from twisted.internet.defer import inlineCallbacks, returnValue, CancelledError
from mock.mock_task import SimpleTask
from nose.twistedtools import deferred
from pyvoltha.adapters.extensions.omci.tasks.task_runner import TaskRunner
DEVICE_ID = 'omci-unit-tests'
class TestTaskRunner(TestCase):
"""
Test the Task Runner Object
"""
def setUp(self):
# defer.setDebugging(True)
self.runner = TaskRunner(DEVICE_ID)
def tearDown(self):
r, self.runner = self.runner, None
r.stop()
def test_default_init(self):
self.assertFalse(self.runner.active)
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
self.assertEqual(self.runner.successful_tasks_completed, 0)
self.assertEqual(self.runner.failed_tasks, 0)
def test_start_stop(self):
self.assertFalse(self.runner.active)
self.runner.start()
self.assertTrue(self.runner.active)
self.runner.stop()
self.assertFalse(self.runner.active)
def unexpected_error(self, _failure):
self.assertEqual('Should not be here, expected success', _failure)
def unexpected_success(self, _results):
self.assertEqual('Should not be here, expected a failure', _results)
def test_simple_task_init(self):
t = SimpleTask(None, DEVICE_ID,
exclusive=True, priority=0,
success=True, value=0, delay=0)
self.assertEqual(t.priority, 0)
self.assertGreater(t.task_id, 0)
self.assertTrue(t.exclusive)
self.assertFalse(t.deferred.called)
def test_task_defaults_and_bound(self):
# Make sure no one has changed some declared defaults/max/min values
from pyvoltha.adapters.extensions.omci.tasks.task import Task
self.assertEqual(128, Task.DEFAULT_PRIORITY)
self.assertEqual(0, Task.MIN_PRIORITY)
self.assertEqual(255, Task.MAX_PRIORITY)
self.assertEqual(10, Task.DEFAULT_WATCHDOG_SECS)
self.assertEqual(3, Task.MIN_WATCHDOG_SECS)
self.assertEqual(60, Task.MAX_WATCHDOG_SECS)
@raises(AssertionError)
def test_task_priority_min(self):
from pyvoltha.adapters.extensions.omci.tasks.task import Task
_ = SimpleTask(None, DEVICE_ID, priority=Task.MIN_PRIORITY - 1)
@raises(AssertionError)
def test_task_priority_max(self):
from pyvoltha.adapters.extensions.omci.tasks.task import Task
_ = SimpleTask(None, DEVICE_ID, priority=Task.MAX_PRIORITY + 1)
@raises(AssertionError)
def test_task_watchdog_min(self):
from pyvoltha.adapters.extensions.omci.tasks.task import Task
_ = SimpleTask(None, DEVICE_ID, watchdog_timeout=Task.MIN_WATCHDOG_SECS - 0.000001)
@raises(AssertionError)
def test_task_watchdog_max(self):
from pyvoltha.adapters.extensions.omci.tasks.task import Task
_ = SimpleTask(None, DEVICE_ID, watchdog_timeout=Task.MAX_WATCHDOG_SECS + 0.000001)
@deferred(timeout=5)
def test_simple_success(self):
expected_result = 123
t = SimpleTask(None, DEVICE_ID,
exclusive=True, priority=0,
success=True, value=expected_result, delay=0)
d = self.runner.queue_task(t)
self.assertEqual(self.runner.pending_tasks, 1)
self.assertEqual(self.runner.running_tasks, 0)
self.runner.start()
def check_results(results):
self.assertEqual(results, expected_result)
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
self.assertEqual(self.runner.successful_tasks_completed, 1)
self.assertEqual(self.runner.failed_tasks, 0)
self.assertTrue(self.runner.active)
return results
d.addCallbacks(check_results, self.unexpected_error)
return d
@raises(Exception)
@deferred(timeout=5)
def test_simple_failure(self):
self.expected_failure = Exception('Testing a task failure')
t = SimpleTask(None, DEVICE_ID,
exclusive=True, priority=0,
success=False, value=self.expected_failure,
delay=0)
d = self.runner.queue_task(t)
self.assertEqual(self.runner.pending_tasks, 1)
self.assertEqual(self.runner.running_tasks, 0)
self.runner.start()
def expected_failure(failure):
self.assertEqual(failure, self.expected_failure)
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
self.assertEqual(self.runner.successful_tasks_completed, 0)
self.assertEqual(self.runner.failed_tasks, 1)
self.assertTrue(self.runner.active)
return failure
d.addCallbacks(self.unexpected_success, expected_failure)
return d
@deferred(timeout=5)
def test_priority(self):
self.last_value_set = 0
t1 = SimpleTask(None, DEVICE_ID,
exclusive=True, priority=1,
success=True, value=1, delay=0)
t2 = SimpleTask(None, DEVICE_ID,
exclusive=True, priority=2, # Should finish first
success=True, value=2, delay=0)
d1 = self.runner.queue_task(t1)
d2 = self.runner.queue_task(t2)
def set_last_value(results):
self.last_value_set = results
d1.addCallbacks(set_last_value, self.unexpected_error)
d2.addCallbacks(set_last_value, self.unexpected_error)
self.assertEqual(self.runner.pending_tasks, 2)
self.assertEqual(self.runner.running_tasks, 0)
d = defer.gatherResults([d1, d2], consumeErrors=True)
def check_results(_):
self.assertEqual(self.last_value_set, 1)
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
self.assertEqual(self.runner.successful_tasks_completed, 2)
d.addCallbacks(check_results, self.unexpected_error)
self.runner.start()
return d
@inlineCallbacks
def check_that_t1_t2_running_and_last_is_not(self, results):
from pyvoltha.common.utils.asleep import asleep
yield asleep(0.1)
self.assertEqual(self.runner.pending_tasks, 1)
self.assertEqual(self.runner.running_tasks, 2)
self.assertEqual(self.runner.successful_tasks_completed, 1)
returnValue(results)
@deferred(timeout=10)
def test_concurrent(self):
blocker = SimpleTask(None, DEVICE_ID,
exclusive=True, priority=10,
success=True, value=1, delay=0.5)
t1 = SimpleTask(None, DEVICE_ID,
exclusive=False, priority=9,
success=True, value=1, delay=2)
t2 = SimpleTask(None, DEVICE_ID,
exclusive=False, priority=9,
success=True, value=1, delay=2)
last = SimpleTask(None, DEVICE_ID,
exclusive=True, priority=8,
success=True, value=1, delay=0)
d0 = self.runner.queue_task(blocker)
d0.addCallbacks(self.check_that_t1_t2_running_and_last_is_not,
self.unexpected_error)
d1 = self.runner.queue_task(t1)
d2 = self.runner.queue_task(t2)
d3 = self.runner.queue_task(last)
self.assertEqual(self.runner.pending_tasks, 4)
self.assertEqual(self.runner.running_tasks, 0)
d = defer.gatherResults([d0, d1, d2, d3], consumeErrors=True)
def check_final_results(_):
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
self.assertEqual(self.runner.successful_tasks_completed, 4)
self.assertEqual(self.runner.failed_tasks, 0)
d.addCallbacks(check_final_results, self.unexpected_error)
self.runner.start()
return d
@raises(CancelledError)
@deferred(timeout=2)
def test_cancel_queued(self):
t = SimpleTask(None, DEVICE_ID,
exclusive=True, priority=9,
success=True, value=1, delay=0)
d = self.runner.queue_task(t)
self.assertEqual(self.runner.pending_tasks, 1)
self.assertEqual(self.runner.running_tasks, 0)
self.runner.cancel_task(t.task_id)
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
return d
@raises(CancelledError)
@deferred(timeout=2)
def test_task_stop_queued(self):
t = SimpleTask(None, DEVICE_ID,
exclusive=True, priority=9,
success=True, value=1, delay=0)
d = self.runner.queue_task(t)
self.assertEqual(self.runner.pending_tasks, 1)
self.assertEqual(self.runner.running_tasks, 0)
t.stop()
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
return d
def test_task_stop_not_queued(self):
t = SimpleTask(None, DEVICE_ID,
exclusive=True, priority=9,
success=True, value=1, delay=0)
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
t.stop()
self.assertFalse(t.running)
@deferred(timeout=3)
def test_task_runner_cancel_running(self):
# Both task run in parallel but t1 will finish first and
# will request t2 to terminate by calling the TaskRunner's
# cancel task method
t1 = SimpleTask(None, DEVICE_ID,
exclusive=False, priority=9,
success=True, value=1, delay=0.5)
t2 = SimpleTask(None, DEVICE_ID,
exclusive=False, priority=9,
success=True, value=1, delay=200)
d1 = self.runner.queue_task(t1)
d2 = self.runner.queue_task(t2)
self.assertEqual(self.runner.pending_tasks, 2)
self.assertEqual(self.runner.running_tasks, 0)
def kill_task_t2(_, task_2_id):
# Called on successful completion of task t1
self.assertIsInstance(task_2_id, int)
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 1)
# Cancel task runner and t2 task ID
self.runner.cancel_task(task_2_id)
self.assertEqual(self.runner.running_tasks, 0)
d1.addCallbacks(kill_task_t2, self.unexpected_error,
callbackArgs=[t2.task_id])
def expected_error(failure):
self.assertTrue(isinstance(failure.value, CancelledError))
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
self.assertEqual(self.runner.successful_tasks_completed, 1)
self.assertEqual(self.runner.failed_tasks, 1)
# T2 should not finish successfully, should get a cancel error
d2.addCallbacks(self.unexpected_success, expected_error)
# Run it
self.runner.start()
return defer.gatherResults([d1, d2], consumeErrors=True)
@deferred(timeout=3)
def test_task_stop_running(self):
# Run two tasks where T1 completes first and requests T2 to be
# canceled by calling T2's stop method
t1 = SimpleTask(None, DEVICE_ID,
exclusive=False, priority=9,
success=True, value=1, delay=0.5)
t2 = SimpleTask(None, DEVICE_ID,
exclusive=False, priority=9,
success=True, value=1, delay=200)
d1 = self.runner.queue_task(t1)
d2 = self.runner.queue_task(t2)
self.assertEqual(self.runner.pending_tasks, 2)
self.assertEqual(self.runner.running_tasks, 0)
def kill_task_t2(_, task_2):
# Called on successful completion of task t1
self.assertIsInstance(task_2, SimpleTask)
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 1)
# Cancel by telling the task to stop itself
task_2.stop()
self.assertEqual(self.runner.running_tasks, 0)
d1.addCallbacks(kill_task_t2, self.unexpected_error,
callbackArgs=[t2])
def expected_error(failure):
self.assertTrue(isinstance(failure.value, CancelledError))
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
self.assertEqual(self.runner.successful_tasks_completed, 1)
self.assertEqual(self.runner.failed_tasks, 1)
# T2 should not finish successfully, should get a cancel error
d2.addCallbacks(self.unexpected_success, expected_error)
# Run it
self.runner.start()
return defer.gatherResults([d1, d2], consumeErrors=True)
@deferred(timeout=3)
def test_task_cancel_not_queued(self):
t = SimpleTask(None, DEVICE_ID,
exclusive=True, priority=9,
success=True, value=1, delay=0)
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
def expected_error(failure):
self.assertTrue(isinstance(failure.value, CancelledError))
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
self.assertEqual(self.runner.successful_tasks_completed, 0)
self.assertEqual(self.runner.failed_tasks, 0)
# self.fail(msg='made it here') # Uncomment to verify called
t.deferred.addCallbacks(self.unexpected_success, expected_error)
self.runner.start()
t.deferred.cancel()
self.assertFalse(t.running)
return t.deferred
@deferred(timeout=3)
def test_task_deferred_cancel_running(self):
# Run two tasks where T1 completes first and requests T2 to be
# canceled by doing a 'cancel' on T2's deferred
t1 = SimpleTask(None, DEVICE_ID,
exclusive=False, priority=9,
success=True, value=1, delay=0.5)
t2 = SimpleTask(None, DEVICE_ID,
exclusive=False, priority=9,
success=True, value=1, delay=200)
d1 = self.runner.queue_task(t1)
d2 = self.runner.queue_task(t2)
self.assertEqual(self.runner.pending_tasks, 2)
self.assertEqual(self.runner.running_tasks, 0)
def kill_task_t2(_, deferred_2):
# Called on successful completion of task t1
self.assertIsInstance(deferred_2, defer.Deferred)
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 1)
# Cancel the deferred for T2
deferred_2.cancel()
self.assertEqual(self.runner.running_tasks, 0)
d1.addCallbacks(kill_task_t2, self.unexpected_error,
callbackArgs=[t2.deferred])
def expected_error(failure):
self.assertTrue(isinstance(failure.value, CancelledError))
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
self.assertEqual(self.runner.successful_tasks_completed, 1)
self.assertEqual(self.runner.failed_tasks, 1)
# self.fail(msg='made it here') # Uncomment to verify called
# T2 should not finish successfully, should get a cancel error
d2.addCallbacks(self.unexpected_success, expected_error)
# Run it
self.runner.start()
return defer.gatherResults([d1, d2], consumeErrors=True)
@deferred(timeout=3)
def test_watchdog_timeout(self):
t = SimpleTask(None, DEVICE_ID, delay=2)
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
self.assertEqual(self.runner.watchdog_timeouts, 0)
# Actual watchdog minimum is probably to long for an automated test, reach
# around and force ti to something smaller (kids, don't try this at home)
t._watchdog_timeout = 0.1
self.runner.queue_task(t)
self.assertEqual(self.runner.pending_tasks, 1)
self.assertEqual(self.runner.running_tasks, 0)
def expected_error(failure):
from pyvoltha.adapters.extensions.omci.tasks.task import WatchdogTimeoutFailure
self.assertTrue(isinstance(failure.value, WatchdogTimeoutFailure))
self.assertEqual(self.runner.pending_tasks, 0)
self.assertEqual(self.runner.running_tasks, 0)
self.assertEqual(self.runner.successful_tasks_completed, 0)
self.assertEqual(self.runner.failed_tasks, 1)
self.assertEqual(self.runner.watchdog_timeouts, 1)
self.assertEqual(self.runner.last_watchdog_failure_task, t.name)
# self.fail(msg='made it here') # Uncomment to verify called
t.deferred.addCallbacks(self.unexpected_success, expected_error)
# Run it
self.runner.start()
return t.deferred
if __name__ == '__main__':
main()