From 98b4fd2daf761897f92d87d270a17380b0d99a3c Mon Sep 17 00:00:00 2001 From: Sascha Silbe Date: Sun, 08 Nov 2009 11:16:47 +0000 Subject: add test for version 1 -> version 2 data store migration --- diff --git a/tests/test_migration_v1_v2.py b/tests/test_migration_v1_v2.py new file mode 100644 index 0000000..82c4cac --- /dev/null +++ b/tests/test_migration_v1_v2.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python +"""Test datastore migration from version 1 to version 2.""" + +import dbus +import decorator +import hashlib +import os +import tempfile +import time +import unittest +import uuid + + +DS_DBUS_SERVICE = "org.laptop.sugar.DataStore" +DS_DBUS_INTERFACE = "org.laptop.sugar.DataStore" +DS_DBUS_PATH = "/org/laptop/sugar/DataStore" +IGNORE_PROPERTIES = [ + 'activity_id', + 'checksum', + 'ctime', + 'mtime', + 'number', + 'timestamp', + 'uid', +] + + +class MigrationV1V2TestCase(unittest.TestCase): + """Test datastore migration from version 1 to version 2.""" + + def __init__(self, *args, **kwargs): + unittest.TestCase.__init__(self, *args, **kwargs) + self._templates = self._v1_properties*10 + + def setUp(self): + # pylint: disable-msg=C0103 + profile = os.environ.get('SUGAR_PROFILE', 'default') + base_dir = os.path.join(os.path.expanduser('~'), '.sugar', profile) + self._root_path = os.path.join(base_dir, 'datastore') + if not os.path.exists(self._root_path): + self._create_v1_datastore() + + self._bus = dbus.SessionBus() + self._datastore = dbus.Interface(self._bus.get_object(DS_DBUS_SERVICE, + DS_DBUS_PATH), DS_DBUS_INTERFACE) + + _v1_properties = [ + { + 'title': lambda number: 'DS test object %d' % (number, ), + 'mime_type': 'text/plain', + }, + { + 'title': lambda number: 'DS test object %d' % (number, ), + 'mime_type': 'text/html', + }, + { + 'title': lambda number: 'DS test object %d' % (number, ), + 'title_set_by_user': '1', + 'keep': '1', + 'mime_type': 'text/html', + 'activity': 'org.sugarlabs.DataStoreTest2', + 'activity_id': lambda *args: str(uuid.uuid4()), + 'timestamp': lambda *args: time.time(), + 'icon-color': '#00ff00,#0000ff', + 'buddies': '{}', + 'description': 'DS migration test object', + 'tags': lambda number: 'test tag%d' % (number, ), + 'preview': dbus.ByteArray(''.join([chr(i) for i in range(255)])), + }, + { + 'title': lambda number: 'DS test object %d' % (number, ), + 'activity': 'org.sugarlabs.DataStoreTest3', + 'activity_id': lambda *args: str(uuid.uuid4()), + 'ctime': lambda *args: time.strftime('%Y-%m-%dT%H:%M:%S'), + }, + { + 'title': lambda number: 'DS test object %d' % (number, ), + 'activity': 'org.sugarlabs.DataStoreTest4', + 'activity_id': lambda *args: str(uuid.uuid4()), + 'mtime': lambda *args: time.strftime('%Y-%m-%dT%H:%M:%S'), + }, + {}, + ] + def _v1_content(self, num): + return ('Foo bar %d\n' % (num, ))*1000 + + def _create_v1_datastore(self): + """Create a version 1 datastore on disk.""" + os.makedirs(self._root_path) + file(os.path.join(self._root_path, 'version'), 'w').write('1') + for i, template in enumerate(self._templates): + metadata = self._fill_template(template, i) + data = self._v1_content(i) + tree_id = str(uuid.uuid4()) + metadata['uid'] = tree_id + metadata['number'] = i + + self._create_v1_entry(tree_id, metadata, data) + + def _fill_template(self, template, i): + metadata = {} + for (key, value) in template.items(): + if callable(value): + value = value(i) + + metadata[key] = value + + return metadata + + def _create_v1_entry(self, tree_id, metadata, data): + """Create a single version 1 datastore entry.""" + checksum = hashlib.md5(data).hexdigest() + entry_dir = os.path.join(self._root_path, tree_id[:2], tree_id) + os.makedirs(entry_dir) + file(os.path.join(entry_dir, 'data'), 'w').write(data) + self._write_v1_metadata(os.path.join(entry_dir, 'metadata'), metadata) + checksum_dir = os.path.join(self._root_path, 'checksums', checksum) + os.makedirs(checksum_dir) + file(os.path.join(checksum_dir, tree_id), 'w').close() + + def _write_v1_metadata(self, directory, metadata): + os.makedirs(directory) + for key, value in metadata.items(): + file(os.path.join(directory, key), 'w').write(str(value)) + + def test_find_all(self): + """Run find() to list all migrated entries.""" + entries, count = self._find({}, ['uid']) + self.assertEquals(count, len(self._templates)) + + def test_get_properties(self): + """Run get_properties() on all entries and verify result.""" + for entry in self._find({}, ['uid'])[0]: + properties = self._datastore.get_properties(entry['uid'], + byte_arrays=True) + number = int(properties['number']) + expected = self._fill_template(self._templates[number], + number) + self._filter_properties(properties) + self._filter_properties(expected) + self.assertEquals(properties, expected) + + def test_get_filename(self): + """Run get_filename() on all entries and verify content.""" + for entry in self._find({}, ['number', 'uid'])[0]: + filename = self._datastore.get_filename(entry['uid'], + byte_arrays=True) + content = file(filename).read() + os.remove(filename) + number = int(entry['number']) + expected = self._v1_content(number) + self.assertEquals(content, expected) + + def _find(self, query, properties): + return self._datastore.find(dbus.Dictionary(query, signature='sv'), + properties, byte_arrays=True) + + def _filter_properties(self, properties): + for key in IGNORE_PROPERTIES: + properties.pop(key, None) + + +def suite(): + test_suite = unittest.TestLoader().loadTestsFromTestCase(MigrationV1V2TestCase) + test_suite.__doc__ = MigrationV1V2TestCase.__doc__ + return test_suite -- cgit v0.9.1