-# Copyright (C) 2009-2010 W. Trevor King <wking@drexel.edu>
+# Copyright (C) 2009-2012 Chris Ball <cjb@laptop.org>
+# W. Trevor King <wking@tremily.us>
#
-# This program is free software; you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation; either version 2 of the License, or
-# (at your option) any later version.
+# This file is part of Bugs Everywhere.
#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
+# Bugs Everywhere is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 2 of the License, or (at your option) any
+# later version.
#
-# You should have received a copy of the GNU General Public License along
-# with this program; if not, write to the Free Software Foundation, Inc.,
-# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+# Bugs Everywhere is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# Bugs Everywhere. If not, see <http://www.gnu.org/licenses/>.
"""
Abstract bug repository data storage to easily support multiple backends.
class InvalidID (KeyError):
def __init__(self, id=None, revision=None, msg=None):
- if msg == None and id != None:
- msg = id
- KeyError.__init__(self, msg)
+ KeyError.__init__(self, id)
+ self.msg = msg
self.id = id
self.revision = revision
+ def __str__(self):
+ if self.msg == None:
+ return '%s in revision %s' % (self.id, self.revision)
+ return self.msg
+
class InvalidRevision (KeyError):
pass
def __init__(self):
Exception.__init__(self, 'No changes to commit')
+class _EMPTY (object):
+ """Entry has been added but has no user-set value."""
+ pass
class Entry (Tree):
- def __init__(self, id, value=None, parent=None, directory=False,
+ def __init__(self, id, value=_EMPTY, parent=None, directory=False,
children=None):
if children == None:
Tree.__init__(self)
"""Add an entry"""
if self.is_writeable() == False:
raise NotWriteable('Cannot add entry to unwriteable storage.')
- try: # Maybe we've already added that id?
- self.get(id)
- pass # yup, no need to add another
- except InvalidID:
+ if not self.exists(id):
self._add(id, *args, **kwargs)
def _add(self, id, parent=None, directory=False):
p = self._data[parent]
self._data[id] = Entry(id, parent=p, directory=directory)
+ def exists(self, *args, **kwargs):
+ """Check an entry's existence"""
+ if self.is_readable() == False:
+ raise NotReadable('Cannot check entry existence in unreadable storage.')
+ return self._exists(*args, **kwargs)
+
+ def _exists(self, id, revision=None):
+ return id in self._data
+
def remove(self, *args, **kwargs):
"""Remove an entry."""
if self.is_writeable() == False:
for entry in reversed(list(self._data[id].traverse())):
self._remove(entry.id)
+ def ancestors(self, *args, **kwargs):
+ """Return a list of the specified entry's ancestors' ids."""
+ if self.is_readable() == False:
+ raise NotReadable('Cannot list parents with unreadable storage.')
+ return self._ancestors(*args, **kwargs)
+
+ def _ancestors(self, id=None, revision=None):
+ if id == None:
+ return []
+ ancestors = []
+ stack = [id]
+ while len(stack) > 0:
+ id = stack.pop(0)
+ parent = self._data[id].parent
+ if parent != None and not parent.id.startswith('__'):
+ ancestor = parent.id
+ ancestors.append(ancestor)
+ stack.append(ancestor)
+ return ancestors
+
def children(self, *args, **kwargs):
"""Return a list of specified entry's children's ids."""
if self.is_readable() == False:
return value
def _get(self, id, default=InvalidObject, revision=None):
- if id in self._data:
+ if id in self._data and self._data[id].value != _EMPTY:
return self._data[id].value
elif default == InvalidObject:
raise InvalidID(id)
p = self._data[-1][parent]
self._data[-1][id] = Entry(id, parent=p, directory=directory)
+ def _exists(self, id, revision=None):
+ if revision == None:
+ revision = -1
+ else:
+ revision = int(revision)
+ return id in self._data[revision]
+
def _remove(self, id):
if self._data[-1][id].directory == True \
and len(self.children(id)) > 0:
for entry in reversed(list(self._data[-1][id].traverse())):
self._remove(entry.id)
+ def _ancestors(self, id=None, revision=None):
+ if id == None:
+ return []
+ if revision == None:
+ revision = -1
+ else:
+ revision = int(revision)
+ ancestors = []
+ stack = [id]
+ while len(stack) > 0:
+ id = stack.pop(0)
+ parent = self._data[revision][id].parent
+ if parent != None and not parent.id.startswith('__'):
+ ancestor = parent.id
+ ancestors.append(ancestor)
+ stack.append(ancestor)
+ return ancestors
+
def _children(self, id=None, revision=None):
if id == None:
id = '__ROOT__'
if revision == None:
revision = -1
+ else:
+ revision = int(revision)
return [c.id for c in self._data[revision][id]
if not c.id.startswith('__')]
def _get(self, id, default=InvalidObject, revision=None):
if revision == None:
revision = -1
- if id in self._data[revision]:
+ else:
+ revision = int(revision)
+ if id in self._data[revision] \
+ and self._data[revision][id].value != _EMPTY:
return self._data[revision][id].value
elif default == InvalidObject:
raise InvalidID(id)
raise EmptyCommit
self._data[-1]["__COMMIT__SUMMARY__"].value = summary
self._data[-1]["__COMMIT__BODY__"].value = body
- rev = len(self._data)-1
+ rev = str(len(self._data)-1)
self._data.append(copy.deepcopy(self._data[-1]))
return rev
raise InvalidRevision(index)
L = len(self._data) - 1 # -1 b/c of initial commit
if index >= -L and index <= L:
- return index % L
+ return str(index % L)
raise InvalidRevision(i)
+ def changed(self, revision):
+ """Return a tuple of lists of ids `(new, modified, removed)` from the
+ specified revision to the current situation.
+ """
+ new = []
+ modified = []
+ removed = []
+ for id,value in self._data[int(revision)].items():
+ if id.startswith('__'):
+ continue
+ if not id in self._data[-1]:
+ removed.append(id)
+ elif value.value != self._data[-1][id].value:
+ modified.append(id)
+ for id in self._data[-1]:
+ if not id in self._data[int(revision)]:
+ new.append(id)
+ return (new, modified, removed)
+
+
if TESTING == True:
class StorageTestCase (unittest.TestCase):
- """Test cases for base Storage class."""
+ """Test cases for Storage class."""
Class = Storage
super(StorageTestCase, self).__init__(*args, **kwargs)
self.dirname = None
+ # this class will be the basis of tests for several classes,
+ # so make sure we print the name of the class we're dealing with.
+ def _classname(self):
+ version = '?'
+ try:
+ if hasattr(self, 's'):
+ version = self.s.version()
+ except:
+ pass
+ return '%s:%s' % (self.Class.__name__, version)
+
+ def fail(self, msg=None):
+ """Fail immediately, with the given message."""
+ raise self.failureException(
+ '({0}) {1}'.format(self._classname(), msg))
+
+ def failIf(self, expr, msg=None):
+ "Fail the test if the expression is true."
+ if expr:
+ raise self.failureException(
+ '({0}) {1}'.format(self._classname(), msg))
+
+ def failUnless(self, expr, msg=None):
+ """Fail the test unless the expression is true."""
+ if not expr:
+ raise self.failureException(
+ '({0}) {1}'.format(self._classname(), msg))
+
def setUp(self):
"""Set up test fixtures for Storage test case."""
super(StorageTestCase, self).setUp()
self.failUnless(len(self.s.children()) == 0, self.s.children())
def test_add_identical_rooted(self):
- """
- Adding entries with the same ID should not increase the number of children.
+ """Adding entries with the same ID should not increase the number of children.
"""
for i in range(10):
self.s.add('some id', directory=False)
self.failUnless(s == ['some id'], s)
def test_add_rooted(self):
- """
- Adding entries should increase the number of children (rooted).
+ """Adding entries should increase the number of children (rooted).
"""
ids = []
for i in range(10):
self.failUnless(s == ids, '\n %s\n !=\n %s' % (s, ids))
def test_add_nonrooted(self):
- """
- Adding entries should increase the number of children (nonrooted).
+ """Adding entries should increase the number of children (nonrooted).
"""
self.s.add('parent', directory=True)
ids = []
s = self.s.children()
self.failUnless(s == ['parent'], s)
- def test_children(self):
+ def test_ancestors(self):
+ """Check ancestors lists.
"""
- Non-UUID ids should be returned as such.
+ self.s.add('parent', directory=True)
+ for i in range(10):
+ i_id = str(i)
+ self.s.add(i_id, 'parent', directory=True)
+ for j in range(10): # add some grandkids
+ j_id = str(20*(i+1)+j)
+ self.s.add(j_id, i_id, directory=(i%2 == 0))
+ ancestors = sorted(self.s.ancestors(j_id))
+ self.failUnless(ancestors == [i_id, 'parent'],
+ 'Unexpected ancestors for %s/%s, "%s"'
+ % (i_id, j_id, ancestors))
+
+ def test_children(self):
+ """Non-UUID ids should be returned as such.
"""
self.s.add('parent', directory=True)
ids = []
s = sorted(self.s.children('parent'))
self.failUnless(s == ids, '\n %s\n !=\n %s' % (s, ids))
- def test_add_invalid_directory(self):
+ def test_grandchildren(self):
+ """Grandchildren should not be returned as children.
"""
- Should not be able to add children to non-directories.
+ self.s.add('parent', directory=True)
+ ids = []
+ for i in range(5):
+ child = 'parent/%s' % str(i)
+ directory = (i % 2 == 0)
+ ids.append(child)
+ self.s.add(child, 'parent', directory=directory)
+ if directory:
+ for j in range(3):
+ grandchild = '%s/%s' % (child, str(j))
+ directory = (j % 2 == 0)
+ self.s.add(grandchild, child, directory=directory)
+ s = sorted(self.s.children('parent'))
+ self.failUnless(s == ids, '\n %s\n !=\n %s' % (s, ids))
+
+ def test_add_invalid_directory(self):
+ """Should not be able to add children to non-directories.
"""
self.s.add('parent', directory=False)
try:
self.s.children('parent'))
def test_remove_rooted(self):
- """
- Removing entries should decrease the number of children (rooted).
+ """Removing entries should decrease the number of children (rooted).
"""
ids = []
for i in range(10):
self.failUnless(s == ids, '\n %s\n !=\n %s' % (s, ids))
def test_remove_nonrooted(self):
- """
- Removing entries should decrease the number of children (nonrooted).
+ """Removing entries should decrease the number of children (nonrooted).
"""
self.s.add('parent', directory=True)
ids = []
self.failUnless(s == ['parent'], s)
def test_remove_directory_not_empty(self):
- """
- Removing a non-empty directory entry should raise exception.
+ """Removing a non-empty directory entry should raise exception.
"""
self.s.add('parent', directory=True)
ids = []
pass
def test_recursive_remove(self):
- """
- Recursive remove should empty the tree.
- """
+ """Recursive remove should empty the tree."""
self.s.add('parent', directory=True)
ids = []
for i in range(10):
val = 'unlikely value'
def test_get_default(self):
- """
- Get should return specified default if id not in Storage.
+ """Get should return specified default if id not in Storage.
"""
ret = self.s.get(self.id, default=self.val)
self.failUnless(ret == self.val,
% (vars(self.Class)['name'], ret, self.val))
def test_get_default_exception(self):
- """
- Get should raise exception if id not in Storage and no default.
+ """Get should raise exception if id not in Storage and no default.
"""
try:
ret = self.s.get(self.id)
pass
def test_get_initial_value(self):
- """
- Data value should be None before any value has been set.
+ """Data value should be default before any value has been set.
"""
self.s.add(self.id, directory=False)
- ret = self.s.get(self.id)
- self.failUnless(ret == None,
- "%s.get() returned %s not None"
- % (vars(self.Class)['name'], ret))
+ val = 'UNLIKELY DEFAULT'
+ ret = self.s.get(self.id, default=val)
+ self.failUnless(ret == val,
+ "%s.get() returned %s not %s"
+ % (vars(self.Class)['name'], ret, val))
def test_set_exception(self):
- """
- Set should raise exception if id not in Storage.
+ """Set should raise exception if id not in Storage.
"""
try:
self.s.set(self.id, self.val)
pass
def test_set(self):
- """
- Set should define the value returned by get.
+ """Set should define the value returned by get.
"""
self.s.add(self.id, directory=False)
self.s.set(self.id, self.val)
% (vars(self.Class)['name'], ret, self.val))
def test_unicode_set(self):
- """
- Set should define the value returned by get.
+ """Set should define the value returned by get.
"""
val = u'Fran\xe7ois'
self.s.add(self.id, directory=False)
val = 'unlikely value'
def test_get_set_persistence(self):
- """
- Set should define the value returned by get after reconnect.
+ """Set should define the value returned by get after reconnect.
"""
self.s.add(self.id, directory=False)
self.s.set(self.id, self.val)
"%s.get() returned %s not %s"
% (vars(self.Class)['name'], ret, self.val))
- def test_add_nonrooted_persistence(self):
+ def test_empty_get_set_persistence(self):
+ """After empty set, get may return either an empty string or default.
"""
- Adding entries should increase the number of children after reconnect.
+ self.s.add(self.id, directory=False)
+ self.s.set(self.id, '')
+ self.s.disconnect()
+ self.s.connect()
+ default = 'UNLIKELY DEFAULT'
+ ret = self.s.get(self.id, default=default)
+ self.failUnless(ret in ['', default],
+ "%s.get() returned %s not in %s"
+ % (vars(self.Class)['name'], ret, ['', default]))
+
+ def test_add_nonrooted_persistence(self):
+ """Adding entries should increase the number of children after reconnect.
"""
self.s.add('parent', directory=True)
ids = []
self.failUnless(s == ['parent'], s)
class VersionedStorageTestCase (StorageTestCase):
- """Test cases for base VersionedStorage class."""
+ """Test cases for VersionedStorage methods."""
Class = VersionedStorage
class VersionedStorage_commit_TestCase (VersionedStorageTestCase):
- """Test cases for VersionedStorage methods."""
+ """Test cases for VersionedStorage.commit and revision_ids methods."""
id = 'unlikely id'
val = 'Some value'
self.s.commit('Added initialization files')
except EmptyCommit:
pass
-
+
def test_revision_id_exception(self):
- """
- Invalid revision id should raise InvalidRevision.
+ """Invalid revision id should raise InvalidRevision.
"""
try:
rev = self.s.revision_id('highly unlikely revision id')
pass
def test_empty_commit_raises_exception(self):
- """
- Empty commit should raise exception.
+ """Empty commit should raise exception.
"""
self._setup_for_empty_commit()
try:
pass
def test_empty_commit_allowed(self):
- """
- Empty commit should _not_ raise exception if allow_empty=True.
+ """Empty commit should _not_ raise exception if allow_empty=True.
"""
self._setup_for_empty_commit()
self.s.commit(self.commit_msg, self.commit_body,
allow_empty=True)
def test_commit_revision_ids(self):
- """
- Commit / revision_id should agree on revision ids.
+ """Commit / revision_id should agree on revision ids.
"""
def val(i):
return '%s:%d' % (self.val, i+1)
% (vars(self.Class)['name'], i, rev, revs[i]))
def test_get_previous_version(self):
- """
- Get should be able to return the previous version.
+ """Get should be able to return the previous version.
"""
def val(i):
return '%s:%d' % (self.val, i+1)
% (vars(self.Class)['name'], ret, val(i), revs[i]))
def test_get_previous_children(self):
- """
- Children list should be revision dependent.
+ """Children list should be revision dependent.
"""
self.s.add('parent', directory=True)
revs = []
children = []
for i in range(10):
new_child = str(i)
- self.s.add(new_child, 'parent', directory=(i % 2 == 0))
+ self.s.add(new_child, 'parent')
self.s.set(new_child, self.val)
revs.append(self.s.commit('%s: %d' % (self.commit_msg, i),
self.commit_body))
cur_children.append(new_child)
children.append(list(cur_children))
for i in range(10):
- ret = self.s.children('parent', revision=revs[i])
+ ret = sorted(self.s.children('parent', revision=revs[i]))
self.failUnless(ret == children[i],
- "%s.get() returned %s not %s for revision %s"
+ "%s.children() returned %s not %s for revision %s"
% (vars(self.Class)['name'], ret,
children[i], revs[i]))
+ def test_avoid_previous_grandchildren(self):
+ """Previous grandchildren should not be returned as children.
+ """
+ self.s.add('parent', directory=True)
+ revs = []
+ cur_children = []
+ children = []
+ for i in range(5):
+ new_child = 'parent/%s' % str(i)
+ directory = (i % 2 == 0)
+ self.s.add(new_child, 'parent', directory=directory)
+ cur_children.append(new_child)
+ children.append(list(cur_children))
+ if directory:
+ for j in range(3):
+ new_grandchild = '%s/%s' % (new_child, str(j))
+ directory = (j % 2 == 0)
+ self.s.add(
+ new_grandchild, new_child, directory=directory)
+ if not directory:
+ self.s.set(new_grandchild, self.val)
+ else:
+ self.s.set(new_child, self.val)
+ revs.append(self.s.commit('%s: %d' % (self.commit_msg, i),
+ self.commit_body))
+ for rev,cur_children in zip(revs, children):
+ ret = sorted(self.s.children('parent', revision=rev))
+ self.failUnless(ret == cur_children,
+ "%s.children() returned %s not %s for revision %s"
+ % (vars(self.Class)['name'], ret,
+ cur_children, rev))
+
+ class VersionedStorage_changed_TestCase (VersionedStorageTestCase):
+ """Test cases for VersionedStorage.changed() method."""
+
+ def test_changed(self):
+ """Changed lists should reflect past activity"""
+ self.s.add('dir', directory=True)
+ self.s.add('modified', parent='dir')
+ self.s.set('modified', 'some value to be modified')
+ self.s.add('moved', parent='dir')
+ self.s.set('moved', 'this entry will be moved')
+ self.s.add('removed', parent='dir')
+ self.s.set('removed', 'this entry will be deleted')
+ revA = self.s.commit('Initial state')
+ self.s.add('new', parent='dir')
+ self.s.set('new', 'this entry is new')
+ self.s.set('modified', 'a new value')
+ self.s.remove('moved')
+ self.s.add('moved2', parent='dir')
+ self.s.set('moved2', 'this entry will be moved')
+ self.s.remove('removed')
+ revB = self.s.commit('Final state')
+ new,mod,rem = self.s.changed(revA)
+ self.failUnless(sorted(new) == ['moved2', 'new'],
+ 'Unexpected new: %s' % new)
+ self.failUnless(mod == ['modified'],
+ 'Unexpected modified: %s' % mod)
+ self.failUnless(sorted(rem) == ['moved', 'removed'],
+ 'Unexpected removed: %s' % rem)
+
def make_storage_testcase_subclasses(storage_class, namespace):
"""Make StorageTestCase subclasses for storage_class in namespace."""
storage_testcase_classes = [
storage_testcase_classes = [
c for c in (
ob for ob in globals().values() if isinstance(ob, type))
- if issubclass(c, StorageTestCase) \
- and c.Class == Storage]
+ if ((issubclass(c, StorageTestCase) \
+ and c.Class == Storage)
+ or
+ (issubclass(c, VersionedStorageTestCase) \
+ and c.Class == VersionedStorage))]
for base_class in storage_testcase_classes:
testcase_class_name = storage_class.__name__ + base_class.__name__
projects / be.git / blobdiff