diff --git a/test/GrampsDb/Cursor_Test.py b/gramps/gen/db/test/cursor_test.py similarity index 76% rename from test/GrampsDb/Cursor_Test.py rename to gramps/gen/db/test/cursor_test.py index 935293a3e..290eaae15 100644 --- a/test/GrampsDb/Cursor_Test.py +++ b/gramps/gen/db/test/cursor_test.py @@ -18,32 +18,19 @@ # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # -# test/GrampsDb/Cursor_Test.py # $Id$ import unittest -import logging import os import tempfile import shutil -import time -import traceback -import sys from bsddb import dbshelve, db -sys.path.append('../../gramps') - try: set() except NameError: from sets import Set as set -import gramps.gen.const - -logger = logging.getLogger('Gramps.GrampsDbBase_Test') - -from GrampsDbTestBase import GrampsDbBaseTest - class Data(object): def __init__(self, handle,surname, name): @@ -64,7 +51,21 @@ class CursorTest(unittest.TestCase): self.env.set_cachesize(0,0x2000000) self.env.set_lk_max_locks(25000) self.env.set_lk_max_objects(25000) - self.env.set_flags(db.DB_LOG_AUTOREMOVE,1) # clean up unused logs + + # clean up unused logs + autoremove_flag = None + autoremove_method = None + for flag in ["DB_LOG_AUTO_REMOVE", "DB_LOG_AUTOREMOVE"]: + if hasattr(db, flag): + autoremove_flag = getattr(db, flag) + break + for method in ["log_set_config", "set_flags"]: + if hasattr(self.env, method): + autoremove_method = getattr(self.env, method) + break + if autoremove_method and autoremove_flag: + autoremove_method(autoremove_flag, 1) + # The DB_PRIVATE flag must go if we ever move to multi-user setup env_flags = db.DB_CREATE|db.DB_RECOVER|db.DB_PRIVATE|\ db.DB_INIT_MPOOL|db.DB_INIT_LOCK|\ @@ -105,12 +106,12 @@ class CursorTest(unittest.TestCase): data = Data(str(1),'surname1','name1') the_txn = self.env.txn_begin() - self.person_map.put(data.handle,data,txn=the_txn) + self.person_map.put(data.handle, data, txn=the_txn) the_txn.commit() v = self.person_map.get(data.handle) - assert v.handle == data.handle + self.assertEqual(v.handle, data.handle) def test_insert_with_curor_closed(self): """test_insert_with_curor_closed""" @@ -123,15 +124,16 @@ class CursorTest(unittest.TestCase): cursor.close() cursor_txn.commit() - data = Data(str(2),'surname2','name2') + data = Data(str(2), 'surname2', 'name2') the_txn = self.env.txn_begin() - self.person_map.put(data.handle,data,txn=the_txn) + self.person_map.put(data.handle, data, txn=the_txn) the_txn.commit() v = self.person_map.get(data.handle) - assert v.handle == data.handle + self.assertEqual(v.handle, data.handle) + @unittest.skip("Insert expected to fail with open cursor") def test_insert_with_curor_open(self): """test_insert_with_curor_open""" @@ -140,9 +142,9 @@ class CursorTest(unittest.TestCase): cursor.first() cursor.next() - data = Data(str(2),'surname2','name2') + data = Data(str(2),'surname2', 'name2') the_txn = self.env.txn_begin() - self.person_map.put(data.handle,data,txn=the_txn) + self.person_map.put(data.handle, data, txn=the_txn) the_txn.commit() cursor.close() @@ -150,9 +152,10 @@ class CursorTest(unittest.TestCase): v = self.person_map.get(data.handle) - assert v.handle == data.handle + self.assertEqual(v.handle, data.handle) - def xtest_insert_with_curor_open_and_db_open(self): + @unittest.skip("Insert expected to fail with open cursor") + def test_insert_with_curor_open_and_db_open(self): """test_insert_with_curor_open_and_db_open""" (person2,surnames2) = self._open_tables() @@ -162,9 +165,9 @@ class CursorTest(unittest.TestCase): cursor.first() cursor.next() - data = Data(str(2),'surname2','name2') + data = Data(str(2),'surname2', 'name2') the_txn = self.env.txn_begin() - self.person_map.put(data.handle,data,txn=the_txn) + self.person_map.put(data.handle, data, txn=the_txn) the_txn.commit() cursor.close() @@ -172,11 +175,11 @@ class CursorTest(unittest.TestCase): v = self.person_map.get(data.handle) - assert v.handle == data.handle + self.assertEqual(v.handle, data.handle) def testSuite(): - suite = unittest.makeSuite(CursorTest,'test') + suite = unittest.makeSuite(CursorTest, 'test') return suite diff --git a/test/GrampsDb/GrampsDbTestBase.py b/gramps/gen/db/test/grampsdbtestbase.py similarity index 64% rename from test/GrampsDb/GrampsDbTestBase.py rename to gramps/gen/db/test/grampsdbtestbase.py index 9390132c8..704f66100 100644 --- a/test/GrampsDb/GrampsDbTestBase.py +++ b/gramps/gen/db/test/grampsdbtestbase.py @@ -18,31 +18,21 @@ # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # -# test/GrampsDb/GrampsDbTestBase.py # $Id$ import unittest -import logging -import os import tempfile import shutil -import time -import traceback -import sys - -sys.path.append('../gramps') try: set() except NameError: from sets import Set as set - -from gramps.gen.db import DbBsddb -from gramps.cli.clidbman import CLIDbManager -import gramps.gen.const -import gramps.gen.lib -logger = logging.getLogger('Gramps.GrampsDbTestBase') +from gramps.gen.db import DbBsddb, DbTxn +from gramps.cli.clidbman import CLIDbManager +from gramps.gen.lib import (Source, RepoRef, Citation, Repository, Person, + Family, Event, Place, MediaObject) class GrampsDbBaseTest(unittest.TestCase): """Base class for unittest that need to be able to create @@ -52,7 +42,6 @@ class GrampsDbBaseTest(unittest.TestCase): def dummy_callback(dummy): pass self._tmpdir = tempfile.mkdtemp() - #self._filename = os.path.join(self._tmpdir,'test.grdb') self._db = DbBsddb() dbman = CLIDbManager(None) @@ -107,79 +96,77 @@ class GrampsDbBaseTest(unittest.TestCase): def _add_source(self,repos=None): # Add a Source - - tran = self._db.transaction_begin() - source = gen.lib.Source() - if repos is not None: - repo_ref = gen.lib.RepoRef() - repo_ref.set_reference_handle(repos.get_handle()) - source.add_repo_reference(repo_ref) - self._db.add_source(source,tran) - self._db.commit_source(source,tran) - self._db.transaction_commit(tran, "Add Source") - return source + with DbTxn("Add Source and Citation", self._db) as tran: + source = Source() + if repos is not None: + repo_ref = RepoRef() + repo_ref.set_reference_handle(repos.get_handle()) + source.add_repo_reference(repo_ref) + self._db.add_source(source, tran) + self._db.commit_source(source, tran) + citation = Citation() + citation.set_reference_handle(source.get_handle()) + self._db.add_citation(citation, tran) + self._db.commit_citation(citation, tran) + + return citation def _add_repository(self): # Add a Repository - tran = self._db.transaction_begin() - repos = gen.lib.Repository() - self._db.add_repository(repos,tran) - self._db.commit_repository(repos,tran) - self._db.transaction_commit(tran, "Add Repository") + with DbTxn("Add Repository", self._db) as tran: + repos = Repository() + self._db.add_repository(repos, tran) + self._db.commit_repository(repos, tran) return repos - def _add_object_with_source(self,sources, object_class,add_method,commit_method): + def _add_object_with_source(self, citations, object_class, add_method, + commit_method): object = object_class() - for source in sources: - src_ref = gen.lib.SourceRef() - src_ref.set_reference_handle(source.get_handle()) - object.add_source_reference(src_ref) - - - tran = self._db.transaction_begin() - add_method(object,tran) - commit_method(object,tran) - self._db.transaction_commit(tran, "Add Object") + with DbTxn("Add Object", self._db) as tran: + for citation in citations: + object.add_citation(citation.get_handle()) + add_method(object, tran) + commit_method(object, tran) return object - def _add_person_with_sources(self,sources): + def _add_person_with_sources(self, citations): - return self._add_object_with_source(sources, - gen.lib.Person, + return self._add_object_with_source(citations, + Person, self._db.add_person, self._db.commit_person) - def _add_family_with_sources(self,sources): + def _add_family_with_sources(self, citations): - return self._add_object_with_source(sources, - gen.lib.Family, + return self._add_object_with_source(citations, + Family, self._db.add_family, self._db.commit_family) - def _add_event_with_sources(self,sources): + def _add_event_with_sources(self, citations): - return self._add_object_with_source(sources, - gen.lib.Event, + return self._add_object_with_source(citations, + Event, self._db.add_event, self._db.commit_event) - def _add_place_with_sources(self,sources): + def _add_place_with_sources(self, citations): - return self._add_object_with_source(sources, - gen.lib.Place, + return self._add_object_with_source(citations, + Place, self._db.add_place, self._db.commit_place) - def _add_media_object_with_sources(self,sources): + def _add_media_object_with_sources(self, citations): - return self._add_object_with_source(sources, - gen.lib.MediaObject, + return self._add_object_with_source(citations, + MediaObject, self._db.add_object, self._db.commit_media_object) diff --git a/test/GrampsDb/GrampsDbBase_Test.py b/gramps/gen/db/test/reference_map_test.py similarity index 57% rename from test/GrampsDb/GrampsDbBase_Test.py rename to gramps/gen/db/test/reference_map_test.py index 2b951e2ef..2c99595f6 100644 --- a/test/GrampsDb/GrampsDbBase_Test.py +++ b/gramps/gen/db/test/reference_map_test.py @@ -18,32 +18,25 @@ # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # -# test/GrampsDb/GrampsDbBase_Test.py # $Id$ import unittest import logging -import os -import tempfile -import shutil import time -import traceback -import sys - -sys.path.append(os.path.join(os.path.dirname(__file__), '../../gramps')) try: set() except NameError: from sets import Set as set -import gramps.gen.lib +from .. import DbTxn +from ...lib import Person, Event, Source, Citation logger = logging.getLogger('Gramps.GrampsDbBase_Test') -from GrampsDbTestBase import GrampsDbBaseTest +from grampsdbtestbase import GrampsDbBaseTest -class ReferenceMapTest (GrampsDbBaseTest): +class ReferenceMapTest(GrampsDbBaseTest): """Test methods on the GrampsDbBase class that are related to the reference_map index implementation.""" @@ -52,74 +45,83 @@ class ReferenceMapTest (GrampsDbBaseTest): a lookup for the reference returns the original record.""" - source = self._add_source() - person = self._add_person_with_sources([source]) + citation = self._add_source() + person = self._add_person_with_sources([citation]) - references = list(self._db.find_backlink_handles(source.get_handle())) + references = list(self._db.find_backlink_handles(citation.get_handle())) - assert len(references) == 1 - assert references[0] == (gen.lib.Person.__name__,person.get_handle()) + self.assertEqual(len(references), 1) + self.assertEqual(references[0], (Person.__name__, person.get_handle())) def test_backlink_for_repository(self): - """check that the source / repos backlink lookup works.""" + """check that the citation /source / repos backlink lookup works.""" repos = self._add_repository() - source = self._add_source(repos=repos) + citation = self._add_source(repos=repos) references = list(self._db.find_backlink_handles(repos.get_handle())) - assert len(references) == 1 - assert references[0] == (gen.lib.Source.__name__,source.get_handle()) + self.assertEqual(len(references), 1) + self.assertEqual(references[0][0], Source.__name__) + + references = list(self._db.find_backlink_handles(references[0][1])) + + self.assertEqual(len(references), 1) + self.assertEqual(references[0], + (Citation.__name__, citation.get_handle())) def test_class_limited_lookup(self): """check that class limited lookups work.""" - source = self._add_source() - person = self._add_person_with_sources([source]) + citation = self._add_source() + person = self._add_person_with_sources([citation]) - self._add_family_with_sources([source]) - self._add_event_with_sources([source]) - self._add_place_with_sources([source]) - self._add_media_object_with_sources([source]) + self._add_family_with_sources([citation]) + self._add_event_with_sources([citation]) + self._add_place_with_sources([citation]) + self._add_media_object_with_sources([citation]) # make sure that we have the correct number of references (one for each object) - references = list(self._db.find_backlink_handles(source.get_handle())) + references = list(self._db.find_backlink_handles(citation.get_handle())) - assert len(references) == 5, "len(references) == %s " % str(len(references)) + self.assertEqual(len(references), 5, + "len(references) == %s " % str(len(references))) # should just return the person reference - references = [ ref for ref in self._db.find_backlink_handles(source.get_handle(),(gen.lib.Person.__name__,)) ] - assert len(references) == 1, "len(references) == %s " % str(len(references)) - assert references[0][0] == gen.lib.Person.__name__, "references = %s" % repr(references) + references = [ref for ref in self._db.find_backlink_handles(citation.get_handle(), (Person.__name__,))] + self.assertEqual(len(references), 1, + "len(references) == %s " % str(len(references))) + self.assertEqual(references[0][0], Person.__name__, + "references = %s" % repr(references)) # should just return the person and event reference - references = list(self._db.find_backlink_handles(source.get_handle(), - (gen.lib.Person.__name__, gen.lib.Event.__name__))) - assert len(references) == 2, "len(references) == %s " % str(len(references)) - assert references[0][0] == gen.lib.Person.__name__, "references = %s" % repr(references) - assert references[1][0] == gen.lib.Event.__name__, "references = %s" % repr(references) - - + references = list(self._db.find_backlink_handles(citation.get_handle(), + (Person.__name__, Event.__name__))) + self.assertEqual(len(references), 2, + "len(references) == %s " % str(len(references))) + self.assertEqual(references[0][0], Person.__name__, + "references = %s" % repr(references)) + self.assertEqual(references[1][0], Event.__name__, + "references = %s" % repr(references)) def test_delete_primary(self): """check that deleting a primary will remove the backreferences from the reference_map""" - source = self._add_source() - person = self._add_person_with_sources([source]) + citation = self._add_source() + person = self._add_person_with_sources([citation]) - assert self._db.get_person_from_handle(person.get_handle()) is not None + self.assertIsNotNone(self._db.get_person_from_handle(person.get_handle())) - tran = self._db.transaction_begin() - self._db.remove_person(person.get_handle(),tran) - self._db.transaction_commit(tran, "Del Person") + with DbTxn("Del Person", self._db) as tran: + self._db.remove_person(person.get_handle(),tran) - assert self._db.get_person_from_handle(person.get_handle()) is None + self.assertIsNone(self._db.get_person_from_handle(person.get_handle())) - references = list(self._db.find_backlink_handles(source.get_handle())) - - assert len(references) == 0, "len(references) == %s " % str(len(references)) + references = list(self._db.find_backlink_handles(citation.get_handle())) + self.assertEqual(len(references), 0, + "len(references) == %s " % str(len(references))) def test_reindex_reference_map(self): """Test that the reindex function works.""" @@ -130,28 +132,28 @@ class ReferenceMapTest (GrampsDbBaseTest): # 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 + self._db.update_reference_map = lambda x,y,z: 1 # Insert a person/source pair. - source = self._add_source() - person = self._add_person_with_sources([source]) + citation = self._add_source() + person = self._add_person_with_sources([citation]) # Check that the reference map does not contain the reference. - references = list(self._db.find_backlink_handles(source.get_handle())) + references = list(self._db.find_backlink_handles(citation.get_handle())) - assert len(references) == 0, "len(references) == %s " % str(len(references)) + self.assertEqual(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.update_reference_map = update_method self._db.reindex_reference_map(cb) # Check that the reference now appears in the reference_map - references = list(self._db.find_backlink_handles(source.get_handle())) + references = list(self._db.find_backlink_handles(citation.get_handle())) - assert len(references) == 1, "len(references) == %s " % str(len(references)) + self.assertEqual(len(references), 1, + "len(references) == %s " % str(len(references))) - - def perf_simple_search_speed(self): num_sources = 100 @@ -211,8 +213,9 @@ class ReferenceMapTest (GrampsDbBaseTest): logger.info("with refs %s\n", str(with_reference_map)) logger.info("without refs %s\n", str(without_reference_map)) - assert with_reference_map < (without_reference_map / 10), "Reference_map should an order of magnitude faster." - + self.assertLess(with_reference_map, without_reference_map / 10, + "Reference_map should an order of magnitude faster.") + def testSuite(): suite = unittest.makeSuite(ReferenceMapTest,'test') return suite