add test for version 1 -> version 2 data store migration

1 files changed, 166 insertions, 0 deletions

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