diff options
Diffstat (limited to 'tests/test_massops.py')
| -rw-r--r-- | tests/test_massops.py | 164 |
1 files changed, 164 insertions, 0 deletions
diff --git a/tests/test_massops.py b/tests/test_massops.py new file mode 100644 index 0000000..ddc5242 --- /dev/null +++ b/tests/test_massops.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python +"""Large number of operations intended for measuring performance.""" + +import dbus +import decorator +import os +import tempfile +import time +import unittest + + +DS_DBUS_SERVICE = "org.laptop.sugar.DataStore" +DS_DBUS_INTERFACE = "org.laptop.sugar.DataStore" +DS_DBUS_PATH = "/org/laptop/sugar/DataStore" +NUM_RUNS = int(os.environ.get('MASSOPS_RUNS', '100')) +IGNORE_PROPERTIES = [ + 'checksum', + 'number', + 'timestamp', + 'uid', +] + + +@decorator.decorator +def repeat(func, *args, **kwargs): + """Run the decorated function NUM_RUNS times.""" + for i_ in range(NUM_RUNS): + func(*args, **kwargs) + + +class MassOpsTestCase(unittest.TestCase): + """Large number of operations intended for measuring performance.""" + + def setUp(self): + # pylint: disable-msg=C0103 + self._bus = dbus.SessionBus() + self._datastore = dbus.Interface(self._bus.get_object(DS_DBUS_SERVICE, + DS_DBUS_PATH), DS_DBUS_INTERFACE) + + _create_properties = { + 'title': 'DS test object', + 'mime_type': 'text/plain', + 'activity': 'org.sugarlabs.DataStoreTest1', + } + _create_content = 'Foo bar\n'*1000 + + def test_create(self): + """Run create() lots of times to create new objects.""" + for i in range(NUM_RUNS): + content_file = tempfile.NamedTemporaryFile() + content_file.write(self._create_content) + content_file.flush() + properties = self._create_properties.copy() + properties['number'] = str(i) + properties['timestamp'] = time.time() + self._datastore.create(properties, content_file.name, False) + content_file.close() + + @repeat + def test_find_all(self): + """Run find() to list all entries.""" + entries, total_count = self._datastore.find({}, ['number'], + byte_arrays=True) + self.assertEquals(total_count, NUM_RUNS) + self.assertEquals(total_count, len(entries)) + for position, entry in enumerate(entries): + self.assertEquals(int(entry['number']), NUM_RUNS-position-1) + + @repeat + def test_find_all_reverse_time(self): + """Run find() to list all entries in reverse chronological order.""" + entries, total_count = self._datastore.find({'order_by': + ['-timestamp']}, ['number'], byte_arrays=True) + self.assertEquals(total_count, NUM_RUNS) + self.assertEquals(total_count, len(entries)) + for position, entry in enumerate(entries): + self.assertEquals(int(entry['number']), position) + + @repeat + def test_find_all_title(self): + """Run find() to list all entries ordered by title.""" + entries, total_count = self._datastore.find({'order_by': + ['+title']}, ['tree_id'], byte_arrays=True) + self.assertEquals(total_count, NUM_RUNS) + self.assertEquals(total_count, len(entries)) + + @repeat + def test_find_all_reverse_title(self): + """Run find() to list all entries ordered by title (reversed).""" + entries, total_count = self._datastore.find({'order_by': + ['-title']}, ['tree_id'], byte_arrays=True) + self.assertEquals(total_count, NUM_RUNS) + self.assertEquals(total_count, len(entries)) + + @repeat + def test_find_all_chunked(self): + """Run find() to list all entries in small chunks.""" + chunk_size = 30 + for chunk_start in range(0, NUM_RUNS, 30): + entries, total_count = self._datastore.find({ + 'offset': chunk_start, 'limit': chunk_size}, + ['number'], byte_arrays=True) + self.assertEquals(len(entries), + min(chunk_size, NUM_RUNS-chunk_start)) + self.assertEquals(total_count, NUM_RUNS) + for position, entry in enumerate(entries): + self.assertEquals(int(entry['number']), + NUM_RUNS-(chunk_start+position)-1) + + def test_get_properties(self): + """Run get_properties() on all entries and verify result.""" + for entry in self._datastore.find({}, ['uid'], byte_arrays=True)[0]: + properties = self._datastore.get_properties(entry['uid'], + byte_arrays=True) + self._filter_properties(properties) + self.assertEquals(properties, self._create_properties) + + def test_get_filename(self): + """Run get_filename() on all entries and verify content.""" + for entry in self._datastore.find({}, ['uid'], byte_arrays=True)[0]: + filename = self._datastore.get_filename(entry['uid'], + byte_arrays=True) + try: + self.assertEquals(file(filename).read(), self._create_content) + finally: + os.remove(filename) + + _update_properties = { + 'title': 'DS test object (updated)', + 'mime_type': 'text/plain', + 'activity': 'org.sugarlabs.DataStoreTest1', + } + _update_content = 'Foo bar baz\n'*1000 + + def test_update(self): + """Update the content of all existing entries""" + content_file = tempfile.NamedTemporaryFile() + content_file.write(self._update_content) + content_file.flush() + for entry in self._datastore.find({}, ['uid'], byte_arrays=True)[0]: + self._datastore.update(entry['uid'], self._update_properties, + content_file.name, False) + + def test_update_verify(self): + """Verify test_update() has changed content and metadata of all entries.""" + for entry in self._datastore.find({}, [], byte_arrays=True)[0]: + filename = self._datastore.get_filename(entry['uid'], + byte_arrays=True) + self._filter_properties(entry) + try: + self.assertEquals(entry, self._update_properties) + self.assertEquals(file(filename).read(), self._update_content) + finally: + os.remove(filename) + + def _filter_properties(self, properties): + for key in IGNORE_PROPERTIES: + properties.pop(key, None) + + +def suite(): + test_suite = unittest.TestLoader().loadTestsFromTestCase(MassOpsTestCase) + test_suite.__doc__ = MassOpsTestCase.__doc__ + return test_suite |