cubicweb: comparison devtools/testlib.py

equal deleted inserted replaced

-:376314ebf273
+:76e0dba5f8f3
 from cubicweb.sobjects import notification
 from cubicweb.web import Redirect, application
 from cubicweb.server.session import security_enabled
 from cubicweb.server.hook import SendMailOp
 from cubicweb.devtools import SYSTEM_ENTITIES, SYSTEM_RELATIONS, VIEW_VALIDATORS
-from cubicweb.devtools import BASE_URL, fake, htmlparser
+from cubicweb.devtools import BASE_URL, fake, htmlparser, DEFAULT_EMPTY_DB_ID
 from cubicweb.utils import json
 # low-level utilities ##########################################################
 class CubicWebDebugger(Debugger):
 html into a temporary file and open a web browser to examinate it.
 """
 def do_view(self, arg):
 import webbrowser
 data = self._getval(arg)
-file('/tmp/toto.html', 'w').write(data)
+with file('/tmp/toto.html', 'w') as toto:
+toto.write(data)
 webbrowser.open('file:///tmp/toto.html')
 def line_context_filter(line_no, center, before=3, after=None):
 """return true if line are in context
 if strict:
 protected_entities = yams.schema.BASE_TYPES
 else:
 protected_entities = yams.schema.BASE_TYPES.union(SYSTEM_ENTITIES)
 return set(schema.entities()) - protected_entities
-def refresh_repo(repo, resetschema=False, resetvreg=False):
-for pool in repo.pools:
-pool.close(True)
-repo.system_source.shutdown()
-devtools.reset_test_database(repo.config)
-for pool in repo.pools:
-pool.reconnect()
-repo._type_source_cache = {}
-repo._extid_cache = {}
-repo.querier._rql_cache = {}
-for source in repo.sources:
-source.reset_caches()
-if resetschema:
-repo.set_schema(repo.config.load_schema(), resetvreg=resetvreg)
 # email handling, to test emails sent by an application ########################
 MAILBOX = []
 """
 appid = 'data'
 configcls = devtools.ApptestConfiguration
 reset_schema = reset_vreg = False # reset schema / vreg between tests
 tags = TestCase.tags | Tags('cubicweb', 'cw_repo')
+test_db_id = DEFAULT_EMPTY_DB_ID
+_cnxs = set() # establised connection
+_cnx  = None  # current connection
+# Too much complicated stuff. the class doesn't need to bear the repo anymore
+@classmethod
+def set_cnx(cls, cnx):
+cls._cnxs.add(cnx)
+cls._cnx = cnx
+@property
+def cnx(self):
+return self.__class__._cnx
 @classproperty
 def config(cls):
 """return the configuration object
 Configuration is cached on the test class.
 """
 try:
+assert not cls is CubicWebTC, "Don't use CubicWebTC directly to prevent database caching issue"
 return cls.__dict__['_config']
 except KeyError:
 home = abspath(join(dirname(sys.modules[cls.__module__].__file__), cls.appid))
 config = cls._config = cls.configcls(cls.appid, apphome=home)
 config.mode = 'test'
 try:
 config.global_set_option('embed-allowed', re.compile('.*'))
 except: # not in server only configuration
 pass
+#XXX this doesn't need to a be classmethod anymore
 @classmethod
 def _init_repo(cls):
 """init the repository and connection to it.
+"""
-Repository and connection are cached on the test class. Once
+# setup configuration for test
-initialized, we simply reset connections and repository caches.
-"""
-if not 'repo' in cls.__dict__:
-cls._build_repo()
-else:
-try:
-cls.cnx.rollback()
-except ProgrammingError:
-pass
-cls._refresh_repo()
-@classmethod
-def _build_repo(cls):
-cls.repo, cls.cnx = devtools.init_test_database(config=cls.config)
 cls.init_config(cls.config)
-cls.repo.hm.call_hooks('server_startup', repo=cls.repo)
+# get or restore and working db.
+db_handler = devtools.get_test_db_handler(cls.config)
+db_handler.build_db_cache(cls.test_db_id, cls.pre_setup_database)
+cls.repo, cnx = db_handler.get_repo_and_cnx(cls.test_db_id)
+# no direct assignation to cls.cnx anymore.
+# cnx is now an instance property that use a class protected attributes.
+cls.set_cnx(cnx)
 cls.vreg = cls.repo.vreg
-cls.websession = DBAPISession(cls.cnx, cls.admlogin)
+cls.websession = DBAPISession(cnx, cls.admlogin)
-cls._orig_cnx = (cls.cnx, cls.websession)
+cls._orig_cnx = (cnx, cls.websession)
 cls.config.repository = lambda x=None: cls.repo
-@classmethod
+def _close_cnx(self):
-def _refresh_repo(cls):
+for cnx in list(self._cnxs):
-refresh_repo(cls.repo, cls.reset_schema, cls.reset_vreg)
+if not cnx._closed:
+cnx.rollback()
+cnx.close()
+self._cnxs.remove(cnx)
 # global resources accessors ###############################################
 @property
 def schema(self):
 # default test setup and teardown #########################################
 def setUp(self):
 # monkey patch send mail operation so emails are sent synchronously
-self._old_mail_postcommit_event = SendMailOp.postcommit_event
+self._patch_SendMailOp()
-SendMailOp.postcommit_event = SendMailOp.sendmails
 pause_tracing()
 previous_failure = self.__class__.__dict__.get('_repo_init_failed')
 if previous_failure is not None:
 self.skipTest('repository is not initialised: %r' % previous_failure)
 try:
 self._init_repo()
+self.addCleanup(self._close_cnx)
 except Exception, ex:
 self.__class__._repo_init_failed = ex
 raise
 resume_tracing()
-self._cnxs = []
 self.setup_database()
 self.commit()
 MAILBOX[:] = [] # reset mailbox
 def tearDown(self):
-if not self.cnx._closed:
+# XXX hack until logilab.common.testlib is fixed
-self.cnx.rollback()
+while self._cleanups:
-for cnx in self._cnxs:
+cleanup, args, kwargs = self._cleanups.pop(-1)
-if not cnx._closed:
+cleanup(*args, **kwargs)
-cnx.close()
-SendMailOp.postcommit_event = self._old_mail_postcommit_event
+def _patch_SendMailOp(self):
+# monkey patch send mail operation so emails are sent synchronously
+_old_mail_postcommit_event = SendMailOp.postcommit_event
+SendMailOp.postcommit_event = SendMailOp.sendmails
+def reverse_SendMailOp_monkey_patch():
+SendMailOp.postcommit_event = _old_mail_postcommit_event
+self.addCleanup(reverse_SendMailOp_monkey_patch)
 def setup_database(self):
 """add your database setup code by overriding this method"""
+@classmethod
+def pre_setup_database(cls, session, config):
+"""add your pre database setup code by overriding this method
+Do not forget to set the cls.test_db_id value to enable caching of the
+result.
+"""
 # user / session management ###############################################
 def user(self, req=None):
 """return the application schema"""
 # definitly don't want autoclose when used as a context manager
 return self.cnx
 autoclose = kwargs.pop('autoclose', True)
 if not kwargs:
 kwargs['password'] = str(login)
-self.cnx = repo_connect(self.repo, unicode(login), **kwargs)
+self.set_cnx(repo_connect(self.repo, unicode(login), **kwargs))
 self.websession = DBAPISession(self.cnx)
-self._cnxs.append(self.cnx)
 if login == self.vreg.config.anonymous_user()[0]:
 self.cnx.anonymous_connection = True
 if autoclose:
 return TestCaseConnectionProxy(self, self.cnx)
 return self.cnx
 def restore_connection(self):
 if not self.cnx is self._orig_cnx[0]:
 if not self.cnx._closed:
 self.cnx.close()
-try:
+cnx, self.websession = self._orig_cnx
-self._cnxs.remove(self.cnx)
+self.set_cnx(cnx)
-except ValueError:
-pass
-self.cnx, self.websession = self._orig_cnx
 # db api ##################################################################
 @nocoverage
 def cursor(self, req=None):
 class AutoPopulateTest(CubicWebTC):
 """base class for test with auto-populating of the database"""
 __abstract__ = True
+test_db_id = 'autopopulate'
 tags = CubicWebTC.tags | Tags('autopopulated')
 pdbclass = CubicWebDebugger
 # this is a hook to be able to define a list of rql queries
 # that are application dependent and cannot be guessed automatically
 """import this if you wan automatic tests to be ran"""
 tags = AutoPopulateTest.tags | Tags('web', 'generated')
 def setUp(self):
-AutoPopulateTest.setUp(self)
+assert not self.__class__ is AutomaticWebTest, 'Please subclass AutomaticWebTest to pprevent database caching issue'
+super(AutomaticWebTest, self).setUp()
 # access to self.app for proper initialization of the authentication
 # machinery (else some views may fail)
 self.app
 ## one each

branch	stable
changeset 7088	76e0dba5f8f3
parent 7075	4751d77394b1
parent 7078	bad26a22fe29
child 7163	d6d905d0344f