Cyclone-Team/gramps
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

further work to complete the reference_map implementation.

Browse Source 
svn: r5563
This commit is contained in:
Richard Taylor 2005-12-16 11:59:13 +00:00
parent 3a323a757a
commit 7dfe9ebe05
5 changed files with 244 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
ChangeLog
View File

@ -1,3 +1,10 @@
2005-12-16 Richard Taylor <rjt-gramps@thegrindstone.me.uk>
* src/GrampsBSDDB.py: reindex_reference_map added to rebuild the
reference_map when upgrading database
* test/RunAllTests.py: script to run multiple unittests
* test/GrampsDbBase_Test.py: unittest to test reference_map table
implementation.
2005-12-15 Don Allingham <don@gramps-project.org>
* src/DisplayState.py: Window management completed
* src/ViewManger.py: progress bar added back in

63
src/GrampsBSDDB.py
View File

@ -554,6 +554,62 @@ class GrampsBSDDB(GrampsDbBase):
for (ref_class_name,ref_handle) in no_longer_required_references:
self.reference_map.delete(str((handle,ref_handle),))
def reindex_reference_map(self):
"""Reindex all primary records in the database. This will be a
slow process for large databases.
At present this method does not clear the reference_map before it
reindexes. This is fine when if reindex is run to index new content or
when upgrading from a non-reference_map version of the database. But it
might be a problem if reindex is used to repair a broken index because any
references to primary objects that are no longer in the database will
remain in the reference_map index. So if you want to reindex for repair
purposes you need to clear the reference_map first.
"""
# Make a dictionary of the functions and classes that we need for
# each of the primary object tables.
primary_tables = {'Person': {'cursor_func': self.get_person_cursor,
'class_func': Person},
'Family': {'cursor_func': self.get_family_cursor,
'class_func': Family},
'Event': {'cursor_func': self.get_event_cursor,
'class_func': Event},
'Place': {'cursor_func': self.get_place_cursor,
'class_func': Place},
'Source': {'cursor_func': self.get_source_cursor,
'class_func': Source},
'MediaObject': {'cursor_func': self.get_media_cursor,
'class_func': MediaObject},
'Repository': {'cursor_func': self.get_repository_cursor,
'class_func': Repository},
}
# Now we use the functions and classes defined above to loop through each of the
# primary object tables.
for primary_table_name in primary_tables.keys():
cursor = primary_tables[primary_table_name]['cursor_func']()
data = cursor.first()
# Grap the real object class here so that the lookup does
# not happen inside the main loop.
class_func = primary_tables[primary_table_name]['class_func']
while data:
found_handle,val = data
obj = class_func()
obj.unserialize(val)
self._update_reference_map(obj,primary_table_name)
data = cursor.next()
cursor.close()
return
def abort_changes(self):
while self.undo():
@ -682,6 +738,7 @@ class GrampsBSDDB(GrampsDbBase):
transaction.add(PERSON_KEY,handle,person.serialize())
self.emit('person-delete',([str(handle)],))
self.person_map.delete(str(handle))
self._delete_primary_from_reference_map(handle)
def remove_source(self,handle,transaction):
if not self.readonly and handle and str(handle) in self.source_map:
@ -690,6 +747,7 @@ class GrampsBSDDB(GrampsDbBase):
transaction.add(SOURCE_KEY,handle,old_data)
self.emit('source-delete',([handle],))
self.source_map.delete(str(handle))
self._delete_primary_from_reference_map(handle)
def remove_repository(self,handle,transaction):
if not self.readonly and handle and str(handle) in self.repository_map:
@ -698,6 +756,7 @@ class GrampsBSDDB(GrampsDbBase):
transaction.add(REPOSITORY_KEY,handle,old_data)
self.emit('repository-delete',([handle],))
self.repository_map.delete(str(handle))
self._delete_primary_from_reference_map(handle)
def remove_family(self,handle,transaction):
if not self.readonly and handle and str(handle) in self.family_map:
@ -706,6 +765,7 @@ class GrampsBSDDB(GrampsDbBase):
transaction.add(FAMILY_KEY,handle,old_data)
self.emit('family-delete',([str(handle)],))
self.family_map.delete(str(handle))
self._delete_primary_from_reference_map(handle)
def remove_event(self,handle,transaction):
if not self.readonly and handle and str(handle) in self.event_map:
@ -714,6 +774,7 @@ class GrampsBSDDB(GrampsDbBase):
transaction.add(EVENT_KEY,handle,old_data)
self.emit('event-delete',([str(handle)],))
self.event_map.delete(str(handle))
self._delete_primary_from_reference_map(handle)
def remove_place(self,handle,transaction):
if not self.readonly and handle and str(handle) in self.place_map:
@ -722,6 +783,7 @@ class GrampsBSDDB(GrampsDbBase):
transaction.add(PLACE_KEY,handle,old_data)
self.emit('place-delete',([handle],))
self.place_map.delete(str(handle))
self._delete_primary_from_reference_map(handle)
def remove_object(self,handle,transaction):
if not self.readonly and handle and str(handle) in self.media_map:
@ -730,6 +792,7 @@ class GrampsBSDDB(GrampsDbBase):
transaction.add(MEDIA_KEY,handle,old_data)
self.emit('media-delete',([handle],))
self.media_map.delete(str(handle))
self._delete_primary_from_reference_map(handle)
def get_person_from_gramps_id(self,val):
"""finds a Person in the database from the passed gramps' ID.

2
src/GrampsDbBase.py
View File

@ -1020,6 +1020,7 @@ class GrampsDbBase(GrampsDBCallback.GrampsDBCallback):
"""
Commits the transaction to the assocated UNDO database.
"""
if self.__LOG_ALL:
log("%s: Transaction commit '%s'\n" % (self.__class__.__name__, str(msg)))
if not len(transaction) or self.readonly:
@ -1238,6 +1239,7 @@ class GrampsDbBase(GrampsDBCallback.GrampsDBCallback):
database, preserving the change in the passed transaction. This
method must be overridden in the derived class.
"""
if not self.readonly:
person = self.get_person_from_handle(handle)
self.genderStats.uncount_person (person)

118
test/GrampsDbBase_Test.py Normal file
View File

@ -0,0 +1,118 @@
import unittest
import logging
import os
import tempfile
import shutil
import sys
sys.path.append('../src')
import GrampsBSDDB
import RelLib
logger = logging.getLogger('Gramps.GrampsDbBase_Test')
class ReferenceMapTest (unittest.TestCase):
def setUp(self):
self._tmpdir = tempfile.mkdtemp()
self._filename = os.path.join(self._tmpdir,'test.grdb')
self._db = GrampsBSDDB.GrampsBSDDB()
self._db.load(self._filename,
None, # callback
"w")
def tearDown(self):
shutil.rmtree(self._tmpdir)
def _add_person_and_source(self):
# Add a Source
tran = self._db.transaction_begin()
source = RelLib.Source()
self._db.add_source(source,tran)
self._db.commit_source(source,tran)
self._db.transaction_commit(tran, "Add Source")
src_ref = RelLib.SourceRef()
src_ref.set_base_handle(source.get_handle())
# Add Person with reference to the Source
tran = self._db.transaction_begin()
person = RelLib.Person()
person.add_source_reference(src_ref)
self._db.add_person(person,tran)
self._db.commit_person(person,tran)
self._db.transaction_commit(tran, "Add Person")
return (person,source)
def test_simple_lookup(self):
"""insert a record and a reference and check that
a lookup for the reference returns the original
record."""
person,source = self._add_person_and_source()
references = [ ref for ref in self._db.find_backlink_handles(source.get_handle()) ]
assert len(references) == 1
assert references[0] == ('Person',person.get_handle())
def test_delete_primary(self):
"""check that deleting a primary will remove the backreferences
from the reference_map"""
person,source = self._add_person_and_source()
assert self._db.get_person_from_handle(person.get_handle()) is not None
tran = self._db.transaction_begin()
self._db.remove_person(person.get_handle(),tran)
self._db.transaction_commit(tran, "Del Person")
assert self._db.get_person_from_handle(person.get_handle()) == None
references = [ ref for ref in self._db.find_backlink_handles(source.get_handle()) ]
assert len(references) == 0, "len(references) == %s " % str(len(references))
def test_reindex_reference_map(self):
"""Test that the reindex function works."""
# unhook the reference_map update function so that we
# can insert some records without the reference_map being updated.
update_method = self._db._update_reference_map
self._db._update_reference_map = lambda x,y: 1
# Insert a person/source pair.
person,source = self._add_person_and_source()
# Check that the reference map does not contain the reference.
references = [ ref for ref in self._db.find_backlink_handles(source.get_handle()) ]
assert len(references) == 0, "len(references) == %s " % str(len(references))
# Reinstate the reference_map method and reindex the database
self._db._update_reference_map = update_method
self._db.reindex_reference_map()
# Check that the reference now appears in the reference_map
references = [ ref for ref in self._db.find_backlink_handles(source.get_handle()) ]
assert len(references) == 1, "len(references) == %s " % str(len(references))
def testSuite():
return unittest.makeSuite(ReferenceMapTest,'test')
if __name__ == '__main__':
unittest.TextTestRunner().run(testSuite())

54
test/RunAllTests.py Normal file
View File

@ -0,0 +1,54 @@
"""Copyright QinetiQ Ltd (2005) - See LICENSE.TXT for licensing information.
"""
import logging
import os
import unittest
from optparse import OptionParser
def make_parser():
usage = "usage: %prog [options]"
parser = OptionParser(usage)
parser.add_option("-v", "--verbosity", type="int", dest="verbose_level", default=0,
help="Level of verboseness")
return parser
def getTestSuites():
test_modules = [ i for i in os.listdir('.') if i[-8:] == "_Test.py" ]
test_suites = []
for module in test_modules:
mod = __import__(module[:-3])
test_suites.append(mod.testSuite())
return test_suites
def allTheTests():
return unittest.TestSuite(getTestSuites())
if __name__ == '__main__':
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(logging.Formatter('%(name)-12s: %(levelname)-8s %(message)s'))
logger = logging.getLogger('Gramps')
logger.addHandler(console)
(options,args) = make_parser().parse_args()
if options.verbose_level == 1:
logger.setLevel(logging.INFO)
elif options.verbose_level >= 2:
logger.setLevel(logging.DEBUG)
os.environ['GRAMPS_SIGNAL'] = "1"
elif options.verbose_level >= 3:
logger.setLevel(logging.NOTSET)
else:
logger.setLevel(logging.ERROR)
unittest.TextTestRunner(verbosity=options.verbose_level).run(allTheTests())