From b2395c98aa17a75a1f6202e6fbc24b0c9d502ab2 Mon Sep 17 00:00:00 2001
From: Philip ABBET <philip.abbet@idiap.ch>
Date: Mon, 16 Oct 2017 15:25:53 +0200
Subject: [PATCH] [unittests] Update to the latest beat.backend.python version
---
beat/core/test/test_inputs.py | 988 +++++++++++++++----------
beat/core/test/test_message_handler.py | 6 +-
2 files changed, 613 insertions(+), 381 deletions(-)
diff --git a/beat/core/test/test_inputs.py b/beat/core/test/test_inputs.py
index c0c8554e..021a0aa9 100644
--- a/beat/core/test/test_inputs.py
+++ b/beat/core/test/test_inputs.py
@@ -28,6 +28,7 @@
import six
import nose.tools
+import unittest
from .mocks import MockDataSource
@@ -37,556 +38,783 @@ from ..dataformat import DataFormat
from . import prefix
-def test_input_creation():
- data_source = MockDataSource([], [])
+class InputTest(unittest.TestCase):
- input = Input('test', 'mock', data_source)
+ def test_creation(self):
+ data_source = MockDataSource([], [])
- assert input.data is None
- assert input.group is None
- nose.tools.eq_(input.data_format, 'mock')
- nose.tools.eq_(input.data_index, -1)
- nose.tools.eq_(input.data_index_end, -1)
- nose.tools.eq_(input.data_source, data_source)
- nose.tools.eq_(input.name, 'test')
- nose.tools.eq_(input.nb_data_blocks_read, 0)
+ input = Input('test', 'mock', data_source)
+ self.assertTrue(input.data is None)
+ self.assertTrue(input.group is None)
+ self.assertEqual(input.data_format, 'mock')
+ self.assertEqual(input.data_index, -1)
+ self.assertEqual(input.data_index_end, -1)
+ self.assertTrue(input.data_same_as_previous)
+ self.assertEqual(input.data_source, data_source)
+ self.assertEqual(input.name, 'test')
+ self.assertEqual(input.nb_data_blocks_read, 0)
-def test_input_no_more_data():
- data_source = MockDataSource([], [])
+ def test_no_more_data(self):
+ data_source = MockDataSource([], [])
- input = Input('test', 'mock', data_source)
+ input = Input('test', 'mock', data_source)
- assert not input.hasMoreData()
+ self.assertFalse(input.hasMoreData())
-def test_input_next():
+ def test_has_more_data(self):
+ dataformat = DataFormat(prefix, 'user/single_integer/1')
- dataformat = DataFormat(prefix, 'user/single_integer/1')
- assert dataformat.valid
+ data_source = MockDataSource([ dataformat.type(value=10) ], [ (0, 0) ])
- data_source = MockDataSource([ dataformat.type(value=10) ], [ (0, 0) ])
- input = Input('test', 'mock', data_source)
+ input = Input('test', 'mock', data_source)
- assert input.hasMoreData()
+ self.assertTrue(input.hasMoreData())
- input.next()
- assert input.data is not None
- nose.tools.eq_(input.data.value, 10)
- nose.tools.eq_(input.data_index, 0)
- nose.tools.eq_(input.data_index_end, 0)
- nose.tools.eq_(input.nb_data_blocks_read, 1)
+ def test_retrieve_one_data_unit(self):
+ group = InputGroup('channel1', restricted_access=False)
- assert not input.hasMoreData()
+ dataformat = DataFormat(prefix, 'user/single_integer/1')
+ data_source = MockDataSource([ dataformat.type(value=10), dataformat.type(value=20) ],
+ [ (0, 0), (1, 1) ])
-def test_restricted_creation():
+ input = Input('test', 'mock', data_source)
- group = InputGroup('channel1')
+ group.add(input)
- assert group.restricted_access
- assert group.synchronization_listener is None
- nose.tools.eq_(group.channel, 'channel1')
- nose.tools.eq_(len(group), 0)
- nose.tools.eq_(group.data_index, -1)
- nose.tools.eq_(group.data_index_end, -1)
+ self.assertTrue(input.hasMoreData())
+ self.assertFalse(input.hasDataChanged())
+ self.assertTrue(input.isDataUnitDone())
+ input.next()
-def test_restricted_one_input():
+ self.assertTrue(input.data is not None)
+ self.assertEqual(input.data.value, 10)
+ self.assertEqual(input.data_index, 0)
+ self.assertEqual(input.data_index_end, 0)
+ self.assertFalse(input.data_same_as_previous)
+ self.assertEqual(input.nb_data_blocks_read, 1)
- data_source = MockDataSource([], [])
+ self.assertTrue(input.hasMoreData())
+ self.assertTrue(input.hasDataChanged())
+ self.assertTrue(input.isDataUnitDone())
- input = Input('input1', 'mock', data_source)
- group = InputGroup('channel1')
+ def test_retrieve_all_data_units(self):
+ group = InputGroup('channel1', restricted_access=False)
- group.add(input)
+ dataformat = DataFormat(prefix, 'user/single_integer/1')
- nose.tools.eq_(input.group, group)
- nose.tools.eq_(len(group), 1)
+ data_source = MockDataSource([ dataformat.type(value=10) ], [ (0, 0) ])
+ input = Input('test', 'mock', data_source)
- nose.tools.eq_(group['input1'], input)
+ group.add(input)
+ self.assertTrue(input.hasMoreData())
+ self.assertFalse(input.hasDataChanged())
+ self.assertTrue(input.isDataUnitDone())
-def test_restricted_two_inputs():
+ input.next()
- data_source1 = MockDataSource([], [])
- input1 = Input('input1', 'mock', data_source1)
+ self.assertTrue(input.data is not None)
+ self.assertEqual(input.data.value, 10)
+ self.assertEqual(input.data_index, 0)
+ self.assertEqual(input.data_index_end, 0)
+ self.assertFalse(input.data_same_as_previous)
+ self.assertEqual(input.nb_data_blocks_read, 1)
- data_source2 = MockDataSource([], [])
- input2 = Input('input2', 'mock', data_source2)
+ self.assertFalse(input.hasMoreData())
+ self.assertTrue(input.hasDataChanged())
+ self.assertTrue(input.isDataUnitDone())
- group = InputGroup('channel1')
- group.add(input1)
- group.add(input2)
+ def test_restricted_access(self):
+ group = InputGroup('channel1', restricted_access=True)
- nose.tools.eq_(input1.group, group)
- nose.tools.eq_(input2.group, group)
- nose.tools.eq_(len(group), 2)
+ dataformat = DataFormat(prefix, 'user/single_integer/1')
- nose.tools.eq_(group['input1'], input1)
- nose.tools.eq_(group['input2'], input2)
+ data_source = MockDataSource([ dataformat.type(value=10) ], [ (0, 0) ])
+ input = Input('test', 'mock', data_source)
+ group.add(input)
-def test_restricted_no_more_data():
- data_source1 = MockDataSource([], [])
- input1 = Input('input1', 'mock', data_source1)
+ self.assertTrue(input.hasMoreData())
+ self.assertFalse(input.hasDataChanged())
+ self.assertTrue(input.isDataUnitDone())
- data_source2 = MockDataSource([], [])
- input2 = Input('input2', 'mock', data_source2)
+ with self.assertRaises(RuntimeError):
+ input.next()
- group = InputGroup('channel1')
- group.add(input1)
- group.add(input2)
+#----------------------------------------------------------
- assert not group.hasMoreData()
+class RestrictedInputGroupTest(unittest.TestCase):
-@nose.tools.raises(RuntimeError)
-def test_restricted_cannot_call_next():
- group = InputGroup('channel1')
- group.next()
+ def test_creation(self):
+ group = InputGroup('channel1')
+ self.assertTrue(group.restricted_access)
+ self.assertTrue(group.synchronization_listener is None)
+ self.assertEqual(group.channel, 'channel1')
+ self.assertEqual(len(group), 0)
+ self.assertEqual(group.data_index, -1)
+ self.assertEqual(group.data_index_end, -1)
-def test_fullaccess_creation():
- group = InputGroup('channel1', restricted_access=False)
- assert not group.restricted_access
- assert group.synchronization_listener is None
- nose.tools.eq_(group.channel, 'channel1')
- nose.tools.eq_(len(group), 0)
- nose.tools.eq_(group.data_index, -1)
- nose.tools.eq_(group.data_index_end, -1)
+ def test_add_one_input(self):
+ data_source = MockDataSource([], [])
+ input = Input('input1', 'mock', data_source)
-def test_fullaccess_creation_with_listener():
- listener = SynchronizationListener()
+ group = InputGroup('channel1')
- group = InputGroup('channel1', synchronization_listener=listener,
- restricted_access=False)
+ group.add(input)
- assert not group.restricted_access
- nose.tools.eq_(group.synchronization_listener, listener)
- nose.tools.eq_(group.channel, 'channel1')
- nose.tools.eq_(len(group), 0)
- nose.tools.eq_(group.data_index, -1)
- nose.tools.eq_(group.data_index_end, -1)
+ self.assertEqual(input.group, group)
+ self.assertEqual(len(group), 1)
+ self.assertEqual(group['input1'], input)
- nose.tools.eq_(listener.data_index_start, -1)
- nose.tools.eq_(listener.data_index_end, -1)
+ def test_add_two_inputs(self):
+ data_source1 = MockDataSource([], [])
+ input1 = Input('input1', 'mock', data_source1)
-def test_fullaccess_one_input():
- data_source = MockDataSource([], [])
+ data_source2 = MockDataSource([], [])
+ input2 = Input('input2', 'mock', data_source2)
- input = Input('input1', 'mock', data_source)
+ group = InputGroup('channel1')
- group = InputGroup('channel1', restricted_access=False)
+ group.add(input1)
+ group.add(input2)
- group.add(input)
+ self.assertEqual(input1.group, group)
+ self.assertEqual(input2.group, group)
+ self.assertEqual(len(group), 2)
+ self.assertEqual(group['input1'], input1)
+ self.assertEqual(group['input2'], input2)
- nose.tools.eq_(input.group, group)
- nose.tools.eq_(len(group), 1)
- nose.tools.eq_(group['input1'], input)
+ def test_no_more_data(self):
+ data_source1 = MockDataSource([], [])
+ input1 = Input('input1', 'mock', data_source1)
+ data_source2 = MockDataSource([], [])
+ input2 = Input('input2', 'mock', data_source2)
-def test_fullaccess_two_inputs():
- data_source1 = MockDataSource([], [])
- input1 = Input('input1', 'mock', data_source1)
+ group = InputGroup('channel1')
- data_source2 = MockDataSource([], [])
- input2 = Input('input2', 'mock', data_source2)
+ group.add(input1)
+ group.add(input2)
- group = InputGroup('channel1', restricted_access=False)
+ self.assertFalse(group.hasMoreData())
- group.add(input1)
- group.add(input2)
- nose.tools.eq_(input1.group, group)
- nose.tools.eq_(input2.group, group)
- nose.tools.eq_(len(group), 2)
+ def test_has_more_data(self):
+ dataformat = DataFormat(prefix, 'user/single_integer/1')
- nose.tools.eq_(group['input1'], input1)
- nose.tools.eq_(group['input2'], input2)
+ data_source1 = MockDataSource([ dataformat.type(value=10) ], [ (0, 0) ])
+ input1 = Input('input1', 'mock', data_source1)
+ data_source2 = MockDataSource([ dataformat.type(value=10) ], [ (0, 0) ])
+ input2 = Input('input2', 'mock', data_source2)
-def test_fullaccess_no_more_data():
- data_source1 = MockDataSource([], [])
- input1 = Input('input1', 'mock', data_source1)
+ group = InputGroup('channel1')
- data_source2 = MockDataSource([], [])
- input2 = Input('input2', 'mock', data_source2)
+ group.add(input1)
+ group.add(input2)
- group = InputGroup('channel1', restricted_access=False)
+ self.assertTrue(group.hasMoreData())
- group.add(input1)
- group.add(input2)
- assert not group.hasMoreData()
+ def test_restricted_access(self):
+ group = InputGroup('channel1')
+ with self.assertRaises(RuntimeError):
+ group.next()
-def test_fullaccess_one_input_iteration():
+#----------------------------------------------------------
- dataformat = DataFormat(prefix, 'user/single_integer/1')
- assert dataformat.valid
- listener = SynchronizationListener()
+class InputGroupTest(unittest.TestCase):
- group = InputGroup('channel1', synchronization_listener=listener,
- restricted_access=False)
+ def test_creation(self):
+ group = InputGroup('channel1', restricted_access=False)
- data_source = MockDataSource(map(lambda x: dataformat.type(value=x),
- six.moves.range(0, 5)), [(0, 0), (1, 1), (2, 3), (4, 4), (5, 5)])
- input = Input('input1', 'mock', data_source)
+ self.assertFalse(group.restricted_access)
+ self.assertTrue(group.synchronization_listener is None)
+ self.assertEqual(group.channel, 'channel1')
+ self.assertEqual(len(group), 0)
+ self.assertEqual(group.data_index, -1)
+ self.assertEqual(group.data_index_end, -1)
- group.add(input)
- offset = 0
- for i in six.moves.range(0, 5):
- assert input.hasMoreData()
+ def test_creation_with_listener(self):
+ listener = SynchronizationListener()
- assert group.hasMoreData()
- group.next()
+ group = InputGroup('channel1', synchronization_listener=listener,
+ restricted_access=False)
- assert input.data is not None
- nose.tools.eq_(input.data.value, i)
- nose.tools.eq_(input.data_index, i + offset)
- nose.tools.eq_(input.nb_data_blocks_read, i + 1)
+ self.assertFalse(group.restricted_access)
+ self.assertEqual(group.synchronization_listener, listener)
+ self.assertEqual(group.channel, 'channel1')
+ self.assertEqual(len(group), 0)
+ self.assertEqual(group.data_index, -1)
+ self.assertEqual(group.data_index_end, -1)
- nose.tools.eq_(group.data_index, i + offset)
- nose.tools.eq_(listener.data_index_start, i + offset)
+ self.assertEqual(listener.data_index_start, -1)
+ self.assertEqual(listener.data_index_end, -1)
- if i == 2:
- offset += 1
- nose.tools.eq_(input.data_index_end, i + offset)
- nose.tools.eq_(group.data_index_end, i + offset)
- nose.tools.eq_(listener.data_index_end, i + offset)
+ def test_add_one_input(self):
+ data_source = MockDataSource([], [])
- assert not input.hasMoreData()
- assert not group.hasMoreData()
+ input = Input('input1', 'mock', data_source)
+ group = InputGroup('channel1', restricted_access=False)
-def test_fullaccess_three_inputs_iteration__same_frequency():
- dataformat = DataFormat(prefix, 'user/single_integer/1')
- assert dataformat.valid
+ group.add(input)
- listener = SynchronizationListener()
+ self.assertEqual(input.group, group)
+ self.assertEqual(len(group), 1)
+ self.assertEqual(group['input1'], input)
- group = InputGroup('channel1', synchronization_listener=listener,
- restricted_access=False)
- data_source1 = MockDataSource(map(lambda x: dataformat.type(value=x),
- six.moves.range(0, 5)), [(0, 0), (1, 1), (2, 2), (3, 3), (4, 4)])
- group.add(Input('input1', 'mock', data_source1))
+ def test_add_two_inputs(self):
+ data_source1 = MockDataSource([], [])
+ input1 = Input('input1', 'mock', data_source1)
- data_source2 = MockDataSource(map(lambda x: dataformat.type(value=x),
- six.moves.range(10, 15)), [(0, 0), (1, 1), (2, 2), (3, 3), (4, 4)])
- group.add(Input('input2', 'mock', data_source2))
+ data_source2 = MockDataSource([], [])
+ input2 = Input('input2', 'mock', data_source2)
- data_source3 = MockDataSource(map(lambda x: dataformat.type(value=x),
- six.moves.range(20, 25)), [(0, 0), (1, 1), (2, 2), (3, 3), (4, 4)])
- group.add(Input('input3', 'mock', data_source3))
+ group = InputGroup('channel1', restricted_access=False)
- nose.tools.eq_(group.data_index, -1)
- nose.tools.eq_(group.data_index_end, -1)
+ group.add(input1)
+ group.add(input2)
- for i in six.moves.range(0, 5):
- assert group.hasMoreData()
+ self.assertEqual(input1.group, group)
+ self.assertEqual(input2.group, group)
+ self.assertEqual(len(group), 2)
+ self.assertEqual(group['input1'], input1)
+ self.assertEqual(group['input2'], input2)
- group.next()
- nose.tools.eq_(group.data_index, i)
- nose.tools.eq_(group.data_index_end, i)
+ def test_no_more_data(self):
+ data_source1 = MockDataSource([], [])
+ input1 = Input('input1', 'mock', data_source1)
- nose.tools.eq_(listener.data_index_start, i)
- nose.tools.eq_(listener.data_index_end, i)
+ data_source2 = MockDataSource([], [])
+ input2 = Input('input2', 'mock', data_source2)
- input1 = group['input1']
- assert input1.data is not None
- nose.tools.eq_(input1.data.value, i)
- nose.tools.eq_(input1.data_index, i)
- nose.tools.eq_(input1.data_index_end, i)
- nose.tools.eq_(input1.nb_data_blocks_read, i + 1)
+ group = InputGroup('channel1', restricted_access=False)
- input2 = group['input2']
- assert input2.data is not None
- nose.tools.eq_(input2.data.value, 10 + i)
- nose.tools.eq_(input2.data_index, i)
- nose.tools.eq_(input2.data_index_end, i)
- nose.tools.eq_(input2.nb_data_blocks_read, i + 1)
+ group.add(input1)
+ group.add(input2)
- input3 = group['input3']
- assert input3.data is not None
- nose.tools.eq_(input3.data.value, 20 + i)
- nose.tools.eq_(input3.data_index, i)
- nose.tools.eq_(input3.data_index_end, i)
- nose.tools.eq_(input3.nb_data_blocks_read, i + 1)
+ self.assertFalse(group.hasMoreData())
- assert not group.hasMoreData()
+ def test_has_more_data(self):
+ dataformat = DataFormat(prefix, 'user/single_integer/1')
-def test_fullaccess_three_inputs_iteration__different_frequencies():
- dataformat = DataFormat(prefix, 'user/single_integer/1')
- assert dataformat.valid
+ data_source1 = MockDataSource([ dataformat.type(value=10) ], [ (0, 0) ])
+ input1 = Input('input1', 'mock', data_source1)
- listener = SynchronizationListener()
+ data_source2 = MockDataSource([ dataformat.type(value=10) ], [ (0, 0) ])
+ input2 = Input('input2', 'mock', data_source2)
- group = InputGroup('channel1', synchronization_listener=listener,
- restricted_access=False)
+ group = InputGroup('channel1', restricted_access=False)
- data_source1 = MockDataSource(map(lambda x: dataformat.type(value=x),
- six.moves.range(0, 6)), [(0, 0), (1, 1), (2, 2), (3, 3), (4, 4), (5, 5)])
- group.add(Input('input1', 'mock', data_source1))
+ group.add(input1)
+ group.add(input2)
- data_source2 = MockDataSource(map(lambda x: dataformat.type(value=x),
- six.moves.range(10, 13)), [(0, 1), (2, 3), (4, 5)])
- group.add(Input('input2', 'mock', data_source2))
+ self.assertTrue(group.hasMoreData())
- data_source3 = MockDataSource([dataformat.type(value=20)], [(0, 5)])
- group.add(Input('input3', 'mock', data_source3))
- nose.tools.eq_(group.data_index, -1)
- nose.tools.eq_(group.data_index_end, -1)
+ def test_retrieve_one_input_iteration(self):
+ indices = [ (0, 0), (1, 1), (2, 3), (4, 4), (5, 5) ]
- for i in six.moves.range(0, 6):
- assert group.hasMoreData()
+ expected_hasMoreData = [True, True, True, True, False]
+ expected_isDataUnitDone = [True, True, True, True, True]
+ expected_hasDataChanged = [True, True, True, True, True]
- group.next()
-
- nose.tools.eq_(group.data_index, i)
- nose.tools.eq_(group.data_index_end, i)
-
- nose.tools.eq_(listener.data_index_start, i)
- nose.tools.eq_(listener.data_index_end, i)
-
- input1 = group['input1']
- assert input1.data is not None
- nose.tools.eq_(input1.data.value, i)
- nose.tools.eq_(input1.data_index, i)
- nose.tools.eq_(input1.data_index_end, i)
- nose.tools.eq_(input1.nb_data_blocks_read, i + 1)
-
- input2 = group['input2']
- assert input2.data is not None
- nose.tools.eq_(input2.data.value, 10 + i // 2)
- nose.tools.eq_(input2.data_index, (i // 2) * 2)
- nose.tools.eq_(input2.data_index_end, (i // 2) * 2 + 1)
- nose.tools.eq_(input2.nb_data_blocks_read, (i + 2) // 2)
-
- input3 = group['input3']
- assert input3.data is not None
- nose.tools.eq_(input3.data.value, 20)
- nose.tools.eq_(input3.data_index, 0)
- nose.tools.eq_(input3.data_index_end, 5)
- nose.tools.eq_(input3.nb_data_blocks_read, 1)
- assert not group.hasMoreData()
+ dataformat = DataFormat(prefix, 'user/single_integer/1')
+ data_source = MockDataSource([ dataformat.type(value=x) for x in range(0, len(indices)) ],
+ indices)
+ input = Input('input', 'mock', data_source)
-def test_fullaccess_two_inputs_iteration__desynchronized_frequencies():
- dataformat = DataFormat(prefix, 'user/single_integer/1')
- assert dataformat.valid
+ group = InputGroup('channel1', restricted_access=False)
- listener = SynchronizationListener()
+ group.add(input)
- group = InputGroup('channel1', synchronization_listener=listener,
- restricted_access=False)
+ self.assertTrue(group.hasMoreData())
- data_source1 = MockDataSource(map(lambda x: dataformat.type(value=x),
- six.moves.range(0, 6)), [(0, 2), (3, 5), (6, 8),
- (9, 11), (12, 14), (15, 17)])
- group.add(Input('input1', 'mock', data_source1))
+ self.assertTrue(input.hasMoreData())
+ self.assertTrue(input.isDataUnitDone())
+ self.assertFalse(input.hasDataChanged())
- data_source2 = MockDataSource(map(lambda x: dataformat.type(value=x),
- six.moves.range(0, 3)), [(0, 4), (5, 10), (11, 17)])
- group.add(Input('input2', 'mock', data_source2))
+ for i in range(0, len(indices)):
+ group.next()
- nose.tools.eq_(group.data_index, -1)
- nose.tools.eq_(group.data_index_end, -1)
- nose.tools.eq_(listener.data_index_start, -1)
- nose.tools.eq_(listener.data_index_end, -1)
+ self.assertEqual(group.data_index, indices[i][0])
+ self.assertEqual(group.data_index_end, indices[i][1])
- assert group.hasMoreData()
- group.next()
- nose.tools.eq_(group.data_index, 0)
- nose.tools.eq_(group.data_index_end, 2)
- nose.tools.eq_(listener.data_index_start, 0)
- nose.tools.eq_(listener.data_index_end, 2)
+ self.assertTrue(input.data is not None)
+ self.assertEqual(input.data.value, i)
+ self.assertEqual(input.data_index, indices[i][0])
+ self.assertEqual(input.data_index_end, indices[i][1])
+ self.assertFalse(input.data_same_as_previous)
+ self.assertEqual(input.nb_data_blocks_read, i + 1)
- assert group.hasMoreData()
- group.next()
- nose.tools.eq_(group.data_index, 3)
- nose.tools.eq_(group.data_index_end, 4)
- nose.tools.eq_(listener.data_index_start, 3)
- nose.tools.eq_(listener.data_index_end, 4)
+ self.assertEqual(group.hasMoreData(), expected_hasMoreData[i])
+ self.assertEqual(input.hasMoreData(), expected_hasMoreData[i])
+ self.assertEqual(input.isDataUnitDone(), expected_isDataUnitDone[i])
+ self.assertEqual(input.hasDataChanged(), expected_hasDataChanged[i])
- assert group.hasMoreData()
- group.next()
- nose.tools.eq_(group.data_index, 5)
- nose.tools.eq_(group.data_index_end, 5)
- nose.tools.eq_(listener.data_index_start, 5)
- nose.tools.eq_(listener.data_index_end, 5)
- assert group.hasMoreData()
- group.next()
- nose.tools.eq_(group.data_index, 6)
- nose.tools.eq_(group.data_index_end, 8)
- nose.tools.eq_(listener.data_index_start, 6)
- nose.tools.eq_(listener.data_index_end, 8)
+ def test_retrieve_three_inputs_iteration__same_frequency(self):
+ indices = [ (0, 0), (1, 1), (2, 2), (3, 3), (4, 4) ]
- assert group.hasMoreData()
- group.next()
- nose.tools.eq_(group.data_index, 9)
- nose.tools.eq_(group.data_index_end, 10)
- nose.tools.eq_(listener.data_index_start, 9)
- nose.tools.eq_(listener.data_index_end, 10)
+ expected_hasMoreData = [True, True, True, True, False]
+ expected_isDataUnitDone = [True, True, True, True, True]
+ expected_hasDataChanged = [True, True, True, True, True]
- assert group.hasMoreData()
- group.next()
- nose.tools.eq_(group.data_index, 11)
- nose.tools.eq_(group.data_index_end, 11)
- nose.tools.eq_(listener.data_index_start, 11)
- nose.tools.eq_(listener.data_index_end, 11)
- assert group.hasMoreData()
- group.next()
- nose.tools.eq_(group.data_index, 12)
- nose.tools.eq_(group.data_index_end, 14)
- nose.tools.eq_(listener.data_index_start, 12)
- nose.tools.eq_(listener.data_index_end, 14)
+ dataformat = DataFormat(prefix, 'user/single_integer/1')
- assert group.hasMoreData()
- group.next()
- nose.tools.eq_(group.data_index, 15)
- nose.tools.eq_(group.data_index_end, 17)
- nose.tools.eq_(listener.data_index_start, 15)
- nose.tools.eq_(listener.data_index_end, 17)
+ data_source1 = MockDataSource([ dataformat.type(value=x) for x in range(0, len(indices)) ],
+ indices)
+ input1 = Input('input1', 'mock', data_source1)
- assert not group.hasMoreData()
+ data_source2 = MockDataSource([ dataformat.type(value=x + 10) for x in range(0, len(indices)) ],
+ indices)
+ input2 = Input('input2', 'mock', data_source2)
+ data_source3 = MockDataSource([ dataformat.type(value=x + 20) for x in range(0, len(indices)) ],
+ indices)
+ input3 = Input('input3', 'mock', data_source3)
-def test_list_creation():
- inputs = InputList()
+ group = InputGroup('channel1', restricted_access=False)
- assert inputs.main_group is None
- nose.tools.eq_(inputs.nbGroups(), 0)
- nose.tools.eq_(len(inputs), 0)
- assert not inputs.hasMoreData()
+ group.add(input1)
+ group.add(input2)
+ group.add(input3)
+ self.assertTrue(group.hasMoreData())
-def test_list_unkown_input_retrieval():
- inputs = InputList()
+ self.assertTrue(input1.hasMoreData())
+ self.assertTrue(input1.isDataUnitDone())
+ self.assertFalse(input1.hasDataChanged())
- assert inputs['unknown'] is None
+ self.assertTrue(input2.hasMoreData())
+ self.assertTrue(input2.isDataUnitDone())
+ self.assertFalse(input2.hasDataChanged())
+ self.assertTrue(input3.hasMoreData())
+ self.assertTrue(input3.isDataUnitDone())
+ self.assertFalse(input3.hasDataChanged())
-def test_list_group_of_unkown_input_retrieval():
- inputs = InputList()
+ for i in range(0, len(indices)):
+ group.next()
- assert inputs.groupOf('unknown') is None
+ self.assertEqual(group.data_index, indices[i][0])
+ self.assertEqual(group.data_index_end, indices[i][1])
+ self.assertTrue(input1.data is not None)
+ self.assertEqual(input1.data.value, i)
+ self.assertEqual(input1.data_index, indices[i][0])
+ self.assertEqual(input1.data_index_end, indices[i][1])
+ self.assertFalse(input1.data_same_as_previous)
+ self.assertEqual(input1.nb_data_blocks_read, i + 1)
-def test_list_one_group_one_input():
- inputs = InputList()
+ self.assertTrue(input2.data is not None)
+ self.assertEqual(input2.data.value, i + 10)
+ self.assertEqual(input2.data_index, indices[i][0])
+ self.assertEqual(input2.data_index_end, indices[i][1])
+ self.assertFalse(input2.data_same_as_previous)
+ self.assertEqual(input2.nb_data_blocks_read, i + 1)
- data_source = MockDataSource([], [])
- input = Input('input1', 'mock', data_source)
+ self.assertTrue(input3.data is not None)
+ self.assertEqual(input3.data.value, i + 20)
+ self.assertEqual(input3.data_index, indices[i][0])
+ self.assertEqual(input3.data_index_end, indices[i][1])
+ self.assertFalse(input3.data_same_as_previous)
+ self.assertEqual(input3.nb_data_blocks_read, i + 1)
- group = InputGroup('channel1')
- group.add(input)
+ self.assertEqual(group.hasMoreData(), expected_hasMoreData[i])
+ self.assertEqual(input1.hasMoreData(), expected_hasMoreData[i])
+ self.assertEqual(input2.hasMoreData(), expected_hasMoreData[i])
+ self.assertEqual(input3.hasMoreData(), expected_hasMoreData[i])
- inputs.add(group)
+ self.assertEqual(input1.isDataUnitDone(), expected_isDataUnitDone[i])
+ self.assertEqual(input2.isDataUnitDone(), expected_isDataUnitDone[i])
+ self.assertEqual(input3.isDataUnitDone(), expected_isDataUnitDone[i])
- nose.tools.eq_(inputs.main_group, group)
- nose.tools.eq_(inputs.nbGroups(), 1)
- nose.tools.eq_(len(inputs), 1)
+ self.assertEqual(input1.hasDataChanged(), expected_hasDataChanged[i])
+ self.assertEqual(input2.hasDataChanged(), expected_hasDataChanged[i])
+ self.assertEqual(input3.hasDataChanged(), expected_hasDataChanged[i])
- nose.tools.eq_(inputs['input1'], input)
- nose.tools.eq_(inputs[0], input)
- nose.tools.eq_(inputs.groupOf('input1'), group)
+ def test_retrieve_three_inputs_iteration__different_frequencies(self):
+ indices1 = [ (0, 0), (1, 1), (2, 2), (3, 3), (4, 4), (5, 5) ]
+ indices2 = [ (0, 1), (2, 3), (4, 5) ]
+ indices3 = [ (0, 5) ]
+ expected_hasMoreData_1 = [True, True, True, True, True, False]
+ expected_hasMoreData_2 = [True, True, True, True, False, False]
+ expected_hasMoreData_3 = [False, False, False, False, False, False]
-def test_list_one_group_two_inputs():
- inputs = InputList()
+ expected_isDataUnitDone_1 = [True, True, True, True, True, True]
+ expected_isDataUnitDone_2 = [False, True, False, True, False, True]
+ expected_isDataUnitDone_3 = [False, False, False, False, False, True]
- data_source1 = MockDataSource([], [])
- input1 = Input('input1', 'mock', data_source1)
+ expected_hasDataChanged_1 = [True, True, True, True, True, True]
+ expected_hasDataChanged_2 = [True, False, True, False, True, False]
+ expected_hasDataChanged_3 = [True, False, False, False, False, False]
- data_source2 = MockDataSource([], [])
- input2 = Input('input2', 'mock', data_source2)
- group = InputGroup('channel1')
+ dataformat = DataFormat(prefix, 'user/single_integer/1')
- group.add(input1)
- group.add(input2)
+ data_source1 = MockDataSource([ dataformat.type(value=x) for x in range(0, len(indices1)) ],
+ indices1)
+ input1 = Input('input1', 'mock', data_source1)
- inputs.add(group)
+ data_source2 = MockDataSource([ dataformat.type(value=x + 10) for x in range(0, len(indices2)) ],
+ indices2)
+ input2 = Input('input2', 'mock', data_source2)
- nose.tools.eq_(inputs.main_group, group)
- nose.tools.eq_(inputs.nbGroups(), 1)
- nose.tools.eq_(len(inputs), 2)
+ data_source3 = MockDataSource([ dataformat.type(value=x + 20) for x in range(0, len(indices3)) ],
+ indices3)
+ input3 = Input('input3', 'mock', data_source3)
- nose.tools.eq_(inputs['input1'], input1)
- nose.tools.eq_(inputs[0], input1)
+ group = InputGroup('channel1', restricted_access=False)
- nose.tools.eq_(inputs['input2'], input2)
- nose.tools.eq_(inputs[1], input2)
+ group.add(input1)
+ group.add(input2)
+ group.add(input3)
- nose.tools.eq_(inputs.groupOf('input1'), group)
- nose.tools.eq_(inputs.groupOf('input2'), group)
+ self.assertTrue(group.hasMoreData())
+ self.assertTrue(input1.hasMoreData())
+ self.assertTrue(input1.isDataUnitDone())
+ self.assertFalse(input1.hasDataChanged())
-def test_list_two_groups_three_inputs():
- inputs = InputList()
+ self.assertTrue(input2.hasMoreData())
+ self.assertTrue(input2.isDataUnitDone())
+ self.assertFalse(input2.hasDataChanged())
- group1 = InputGroup('channel1')
+ self.assertTrue(input3.hasMoreData())
+ self.assertTrue(input3.isDataUnitDone())
+ self.assertFalse(input3.hasDataChanged())
- data_source1 = MockDataSource([], [])
- input1 = Input('input1', 'mock', data_source1)
- group1.add(input1)
+ for i in range(0, len(indices1)):
+ group.next()
- data_source2 = MockDataSource([], [])
- input2 = Input('input2', 'mock', data_source2)
- group1.add(input2)
+ self.assertEqual(group.data_index, indices1[i][0])
+ self.assertEqual(group.data_index_end, indices1[i][1])
- inputs.add(group1)
+ self.assertTrue(input1.data is not None)
+ self.assertEqual(input1.data.value, i)
+ self.assertEqual(input1.data_index, indices1[i][0])
+ self.assertEqual(input1.data_index_end, indices1[i][1])
+ self.assertFalse(input1.data_same_as_previous)
+ self.assertEqual(input1.nb_data_blocks_read, i + 1)
- group2 = InputGroup('channel2', restricted_access=False)
+ self.assertTrue(input2.data is not None)
+ self.assertEqual(input2.data.value, (i // 2) + 10)
+ self.assertEqual(input2.data_index, indices2[i // 2][0])
+ self.assertEqual(input2.data_index_end, indices2[i // 2][1])
+ self.assertEqual(input2.nb_data_blocks_read, (i // 2) + 1)
- data_source3 = MockDataSource([], [])
- input3 = Input('input3', 'mock', data_source3)
- group2.add(input3)
+ self.assertTrue(input3.data is not None)
+ self.assertEqual(input3.data.value, 20)
+ self.assertEqual(input3.data_index, indices3[0][0])
+ self.assertEqual(input3.data_index_end, indices3[0][1])
+ self.assertEqual(input3.nb_data_blocks_read, 1)
- inputs.add(group2)
+ self.assertEqual(group.hasMoreData(), expected_hasMoreData_1[i])
+ self.assertEqual(input1.hasMoreData(), expected_hasMoreData_1[i])
+ self.assertEqual(input2.hasMoreData(), expected_hasMoreData_2[i])
+ self.assertEqual(input3.hasMoreData(), expected_hasMoreData_3[i])
- nose.tools.eq_(inputs.main_group, group1)
- nose.tools.eq_(inputs.nbGroups(), 2)
- nose.tools.eq_(len(inputs), 3)
+ self.assertEqual(input1.isDataUnitDone(), expected_isDataUnitDone_1[i])
+ self.assertEqual(input2.isDataUnitDone(), expected_isDataUnitDone_2[i])
+ self.assertEqual(input3.isDataUnitDone(), expected_isDataUnitDone_3[i])
- nose.tools.eq_(inputs['input1'], input1)
- nose.tools.eq_(inputs[0], input1)
+ self.assertEqual(input1.hasDataChanged(), expected_hasDataChanged_1[i])
+ self.assertEqual(input2.hasDataChanged(), expected_hasDataChanged_2[i])
+ self.assertEqual(input3.hasDataChanged(), expected_hasDataChanged_3[i])
- nose.tools.eq_(inputs['input2'], input2)
- nose.tools.eq_(inputs[1], input2)
- nose.tools.eq_(inputs['input3'], input3)
- nose.tools.eq_(inputs[2], input3)
+ def test_retrieve_two_inputs_iteration__desynchronized_frequencies(self):
+ indices1 = [ (0, 2), (3, 5), (6, 8), (9, 11), (12, 14), (15, 17) ]
+ indices2 = [ (0, 4), (5, 10), (11, 17) ]
- nose.tools.eq_(inputs.groupOf('input1'), group1)
- nose.tools.eq_(inputs.groupOf('input2'), group1)
- nose.tools.eq_(inputs.groupOf('input3'), group2)
+ expected_values_1 = [0, 1, 1, 2, 3, 3, 4, 5]
+ expected_values_2 = [0, 0, 1, 1, 1, 2, 2, 2]
- counter = 0
- for input in inputs:
- nose.tools.eq_(inputs[counter], input)
- counter += 1
+ expected_group_indices = [(0, 2), (3, 4), (5, 5), (6, 8), (9, 10), (11, 11), (12, 14), (15, 17)]
+ expected_indices_1 = [(0, 2), (3, 5), (3, 5), (6, 8), (9, 11), (9, 11), (12, 14), (15, 17)]
+ expected_indices_2 = [(0, 4), (0, 4), (5, 10), (5, 10), (5, 10), (11, 17), (11, 17), (11, 17)]
- nose.tools.eq_(counter, 3)
+ expected_hasMoreData_1 = [True, True, True, True, True, True, True, False]
+ expected_hasMoreData_2 = [True, True, True, True, True, False, False, False]
+
+ expected_isDataUnitDone_1 = [True, False, True, True, False, True, True, True]
+ expected_isDataUnitDone_2 = [False, True, False, False, True, False, False, True]
+
+ expected_hasDataChanged_1 = [True, True, False, True, True, False, True, True]
+ expected_hasDataChanged_2 = [True, False, True, False, False, True, False, False]
+
+
+ dataformat = DataFormat(prefix, 'user/single_integer/1')
+
+ data_source1 = MockDataSource([ dataformat.type(value=x) for x in range(0, len(indices1)) ],
+ indices1)
+ input1 = Input('input1', 'mock', data_source1)
+
+ data_source2 = MockDataSource([ dataformat.type(value=x) for x in range(0, len(indices2)) ],
+ indices2)
+ input2 = Input('input2', 'mock', data_source2)
+
+ group = InputGroup('channel1', restricted_access=False)
+
+ group.add(input1)
+ group.add(input2)
+
+ self.assertTrue(group.hasMoreData())
+
+ self.assertTrue(input1.hasMoreData())
+ self.assertTrue(input1.isDataUnitDone())
+ self.assertFalse(input1.hasDataChanged())
+
+ self.assertTrue(input2.hasMoreData())
+ self.assertTrue(input2.isDataUnitDone())
+ self.assertFalse(input2.hasDataChanged())
+
+ for i in range(0, len(expected_indices_1)):
+ group.next()
+
+ self.assertEqual(group.data_index, expected_group_indices[i][0])
+ self.assertEqual(group.data_index_end, expected_group_indices[i][1])
+
+ self.assertTrue(input1.data is not None)
+ self.assertEqual(input1.data.value, expected_values_1[i])
+ self.assertEqual(input1.data_index, expected_indices_1[i][0])
+ self.assertEqual(input1.data_index_end, expected_indices_1[i][1])
+
+ self.assertTrue(input2.data is not None)
+ self.assertEqual(input2.data.value, expected_values_2[i])
+ self.assertEqual(input2.data_index, expected_indices_2[i][0])
+ self.assertEqual(input2.data_index_end, expected_indices_2[i][1])
+
+ self.assertEqual(group.hasMoreData(), expected_hasMoreData_1[i])
+ self.assertEqual(input1.hasMoreData(), expected_hasMoreData_1[i])
+ self.assertEqual(input2.hasMoreData(), expected_hasMoreData_2[i])
+
+ self.assertEqual(input1.isDataUnitDone(), expected_isDataUnitDone_1[i])
+ self.assertEqual(input2.isDataUnitDone(), expected_isDataUnitDone_2[i])
+
+ self.assertEqual(input1.hasDataChanged(), expected_hasDataChanged_1[i])
+ self.assertEqual(input2.hasDataChanged(), expected_hasDataChanged_2[i])
+
+
+ def test_retrieve_two_inputs_iteration__different_frequencies(self):
+ indices1 = [ (0, 0), (1, 1), (2, 2), (3, 3), (4, 4), (5, 5), (6, 6), (7, 7) ]
+ indices2 = [ (0, 3), (4, 7) ]
+
+ expected_values_1 = [0, 1, 2, 3, 4, 5, 6, 7]
+ expected_values_2 = [0, 0, 0, 0, 1, 1, 1, 1]
+
+ expected_hasMoreData_1 = [True, True, True, True, True, True, True, False]
+ expected_hasMoreData_2 = [True, True, True, True, False, False, False, False]
+
+ expected_isDataUnitDone_1 = [True, True, True, True, True, True, True, True]
+ expected_isDataUnitDone_2 = [False, False, False, True, False, False, False, True]
+
+ expected_hasDataChanged_1 = [True, True, True, True, True, True, True, True]
+ expected_hasDataChanged_2 = [True, False, False, False, True, False, False, False]
+
+
+ dataformat = DataFormat(prefix, 'user/single_integer/1')
+
+ data_source1 = MockDataSource([ dataformat.type(value=x) for x in range(0, len(indices1)) ],
+ indices1)
+ input1 = Input('input1', 'mock', data_source1)
+
+ data_source2 = MockDataSource([ dataformat.type(value=x) for x in range(0, len(indices2)) ],
+ indices2)
+ input2 = Input('input2', 'mock', data_source2)
+
+ group = InputGroup('channel1', restricted_access=False)
+
+ group.add(input1)
+ group.add(input2)
+
+ self.assertTrue(group.hasMoreData())
+
+ self.assertTrue(input1.hasMoreData())
+ self.assertTrue(input1.isDataUnitDone())
+ self.assertFalse(input1.hasDataChanged())
+
+ self.assertTrue(input2.hasMoreData())
+ self.assertTrue(input2.isDataUnitDone())
+ self.assertFalse(input2.hasDataChanged())
+
+ for i in range(0, len(indices1)):
+ group.next()
+
+ self.assertEqual(group.data_index, indices1[i][0])
+ self.assertEqual(group.data_index_end, indices1[i][1])
+
+ self.assertTrue(input1.data is not None)
+ self.assertEqual(input1.data.value, expected_values_1[i])
+ self.assertEqual(input1.data_index, indices1[i][0])
+ self.assertEqual(input1.data_index_end, indices1[i][1])
+
+ self.assertTrue(input2.data is not None)
+ self.assertEqual(input2.data.value, expected_values_2[i])
+ self.assertEqual(input2.data_index, indices2[i // 4][0])
+ self.assertEqual(input2.data_index_end, indices2[i // 4][1])
+
+ self.assertEqual(group.hasMoreData(), expected_hasMoreData_1[i])
+ self.assertEqual(input1.hasMoreData(), expected_hasMoreData_1[i])
+ self.assertEqual(input2.hasMoreData(), expected_hasMoreData_2[i])
+
+ self.assertEqual(input1.isDataUnitDone(), expected_isDataUnitDone_1[i])
+ self.assertEqual(input2.isDataUnitDone(), expected_isDataUnitDone_2[i])
+
+ self.assertEqual(input1.hasDataChanged(), expected_hasDataChanged_1[i])
+ self.assertEqual(input2.hasDataChanged(), expected_hasDataChanged_2[i])
+
+
+#----------------------------------------------------------
+
+
+class InputListTest(unittest.TestCase):
+
+ def test_creation(self):
+ inputs = InputList()
+
+ self.assertTrue(inputs.main_group is None)
+ self.assertEqual(inputs.nbGroups(), 0)
+ self.assertEqual(len(inputs), 0)
+ self.assertFalse(inputs.hasMoreData())
+
+
+ def test_list_unkown_input_retrieval(self):
+ inputs = InputList()
+
+ self.assertTrue(inputs['unknown'] is None)
+
+
+ def test_list_group_of_unkown_input_retrieval(self):
+ inputs = InputList()
+
+ self.assertTrue(inputs.groupOf('unknown') is None)
+
+
+ def test_list_one_group_one_input(self):
+ inputs = InputList()
+
+ data_source = MockDataSource([], [])
+ input = Input('input1', 'mock', data_source)
+
+ group = InputGroup('channel1')
+ group.add(input)
+
+ inputs.add(group)
+
+ self.assertEqual(inputs.main_group, group)
+ self.assertEqual(inputs.nbGroups(), 1)
+ self.assertEqual(len(inputs), 1)
+
+ self.assertEqual(inputs['input1'], input)
+ self.assertEqual(inputs[0], input)
+
+ self.assertEqual(inputs.groupOf('input1'), group)
+
+
+ def test_list_one_group_two_inputs(self):
+ inputs = InputList()
+
+ data_source1 = MockDataSource([], [])
+ input1 = Input('input1', 'mock', data_source1)
+
+ data_source2 = MockDataSource([], [])
+ input2 = Input('input2', 'mock', data_source2)
+
+ group = InputGroup('channel1')
+
+ group.add(input1)
+ group.add(input2)
+
+ inputs.add(group)
+
+ self.assertEqual(inputs.main_group, group)
+ self.assertEqual(inputs.nbGroups(), 1)
+ self.assertEqual(len(inputs), 2)
+
+ self.assertEqual(inputs['input1'], input1)
+ self.assertEqual(inputs[0], input1)
+
+ self.assertEqual(inputs['input2'], input2)
+ self.assertEqual(inputs[1], input2)
+
+ self.assertEqual(inputs.groupOf('input1'), group)
+ self.assertEqual(inputs.groupOf('input2'), group)
+
+
+ def test_list_two_groups_three_inputs(self):
+ inputs = InputList()
+
+ group1 = InputGroup('channel1')
+
+ data_source1 = MockDataSource([], [])
+ input1 = Input('input1', 'mock', data_source1)
+ group1.add(input1)
+
+ data_source2 = MockDataSource([], [])
+ input2 = Input('input2', 'mock', data_source2)
+ group1.add(input2)
+
+ inputs.add(group1)
+
+ group2 = InputGroup('channel2', restricted_access=False)
+
+ data_source3 = MockDataSource([], [])
+ input3 = Input('input3', 'mock', data_source3)
+ group2.add(input3)
+
+ inputs.add(group2)
+
+ self.assertEqual(inputs.main_group, group1)
+ self.assertEqual(inputs.nbGroups(), 2)
+ self.assertEqual(len(inputs), 3)
+
+ self.assertEqual(inputs['input1'], input1)
+ self.assertEqual(inputs[0], input1)
+
+ self.assertEqual(inputs['input2'], input2)
+ self.assertEqual(inputs[1], input2)
+
+ self.assertEqual(inputs['input3'], input3)
+ self.assertEqual(inputs[2], input3)
+
+ self.assertEqual(inputs.groupOf('input1'), group1)
+ self.assertEqual(inputs.groupOf('input2'), group1)
+ self.assertEqual(inputs.groupOf('input3'), group2)
+
+ counter = 0
+ for input in inputs:
+ self.assertEqual(inputs[counter], input)
+ counter += 1
+
+ self.assertEqual(counter, 3)
diff --git a/beat/core/test/test_message_handler.py b/beat/core/test/test_message_handler.py
index 3e215510..0b7eb024 100644
--- a/beat/core/test/test_message_handler.py
+++ b/beat/core/test/test_message_handler.py
@@ -89,7 +89,7 @@ class TestMessageHandler(unittest.TestCase):
input_b = Input('b', 'user/single_integer/1', data_source_b)
- group = InputGroup('channel')
+ group = InputGroup('channel', restricted_access=False)
group.add(input_a)
group.add(input_b)
@@ -143,11 +143,15 @@ class TestMessageHandler(unittest.TestCase):
self.remote_input_a.next()
nose.tools.eq_(self.remote_input_a.data.value, 10)
+ assert self.remote_input_a.hasDataChanged()
assert self.remote_input_a.hasMoreData()
+ assert self.remote_input_a.isDataUnitDone()
self.remote_input_a.next()
nose.tools.eq_(self.remote_input_a.data.value, 20)
+ assert self.remote_input_a.hasDataChanged()
assert not self.remote_input_a.hasMoreData()
+ assert self.remote_input_a.isDataUnitDone()
def test_group_has_more_data(self):
--
GitLab