Commit b2395c98 authored by Philip ABBET's avatar Philip ABBET
Browse files

[unittests] Update to the latest beat.backend.python version

parent dfca1e82
......@@ -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)