mirror/bacpypes
1
0
Fork 0
mirror of https://github.com/JoelBender/bacpypes synced 2025-09-28 22:15:23 +08:00
Code Issues Releases Wiki Activity

include time limit in tests, add some tests

Browse Source
This commit is contained in:
Joel Bender 2015-08-09 12:30:47 -04:00
parent 1a20f27c8f
commit 816c652bbf
1 changed files with 65 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
tests/test_utilities/test_time_machine.py
View File

@ -62,7 +62,7 @@ class SampleOneShotTask(OneShotTask):
self.process_task_called = 0
def process_task(self):
if _debug: SampleOneShotTask._debug("process_task")
if _debug: SampleOneShotTask._debug("process_task @ %r", time_machine.current_time)
self.process_task_called += 1
@ -75,36 +75,54 @@ sample_task_function_called = 0
@bacpypes_debugging
def sample_task_function(*args, **kwargs):
if _debug: sample_task_function._debug("sample_task_function %r %r", args, kwargs)
if _debug: sample_task_function._debug("sample_task_function %r %r @ %r", args, kwargs, time_machine.current_time)
global sample_task_function_called
# bump the counter
sample_task_function_called += 1
@bacpypes_debugging
class TestTaskManager(unittest.TestCase):
#
# SampleRecurringTask
#
def test_manager_exists(self):
if _debug: TestTaskManager._debug("test_manager_exists")
@bacpypes_debugging
class SampleRecurringTask(RecurringTask):
def __init__(self):
if _debug: SampleRecurringTask._debug("__init__")
RecurringTask.__init__(self)
self.process_task_called = 0
def process_task(self):
if _debug: SampleRecurringTask._debug("process_task @ %r", time_machine.current_time)
self.process_task_called += 1
@bacpypes_debugging
class TestTimeMachine(unittest.TestCase):
def test_time_machine_exists(self):
if _debug: TestTimeMachine._debug("test_time_machine_exists")
# time machine created by setUpPackage
assert time_machine is not None
def test_empty_run(self):
if _debug: TestTaskManager._debug("test_empty_run")
if _debug: TestTimeMachine._debug("test_empty_run")
# reset the time machine
reset_time_machine()
# let it run
run_time_machine()
run_time_machine(60.0)
# no time has passed
assert time_machine.current_time == 0.0
def test_one_shot_immediate(self):
if _debug: TestTaskManager._debug("test_one_shot_immediate")
if _debug: TestTimeMachine._debug("test_one_shot_immediate")
# create a function task
ft = SampleOneShotTask()
@ -112,14 +130,14 @@ class TestTaskManager(unittest.TestCase):
# reset the time machine, install the task, let it run
reset_time_machine()
ft.install_task(0.0)
run_time_machine()
run_time_machine(60.0)
# function called, no time has passed
assert ft.process_task_called == 1
assert time_machine.current_time == 0.0
def test_function_task_immediate(self):
if _debug: TestTaskManager._debug("test_function_task_immediate")
if _debug: TestTimeMachine._debug("test_function_task_immediate")
global sample_task_function_called
# create a function task
@ -129,14 +147,14 @@ class TestTaskManager(unittest.TestCase):
# reset the time machine, install the task, let it run
reset_time_machine()
ft.install_task(0.0)
run_time_machine()
run_time_machine(60.0)
# function called, no time has passed
assert sample_task_function_called == 1
assert time_machine.current_time == 0.0
def test_function_task_delay(self):
if _debug: TestTaskManager._debug("test_function_task_delay")
if _debug: TestTimeMachine._debug("test_function_task_delay")
global sample_task_function_called
sample_delay = 10.0
@ -148,8 +166,41 @@ class TestTaskManager(unittest.TestCase):
# reset the time machine, install the task, let it run
reset_time_machine()
ft.install_task(sample_delay)
run_time_machine()
run_time_machine(60.0)
# function called, no time has passed
assert sample_task_function_called == 1
assert time_machine.current_time == sample_delay
def test_recurring_task_1(self):
if _debug: TestTimeMachine._debug("test_recurring_task_1")
# create a function task
ft = SampleRecurringTask()
# reset the time machine, install the task, let it run
reset_time_machine()
ft.install_task(1000.0)
run_time_machine(5.0)
# function called, no time has passed
assert ft.process_task_called == 4
assert time_machine.current_time == 5.0
def test_recurring_task_2(self):
if _debug: TestTimeMachine._debug("test_recurring_task_2")
# create a function task
ft1 = SampleRecurringTask()
ft2 = SampleRecurringTask()
# reset the time machine, install the task, let it run
reset_time_machine()
ft1.install_task(1000.0)
ft2.install_task(1500.0)
run_time_machine(5.0)
# function called, no time has passed
assert ft1.process_task_called == 4
assert ft2.process_task_called == 3
assert time_machine.current_time == 5.0