cubicweb: comparison devtools/testlib.py

equal deleted inserted replaced

-:f4d1d5d9ccbb
+:90f2f20367bc
 __docformat__ = "restructuredtext en"
 import os
 import sys
 import re
+import urlparse
+from os.path import dirname, join
 from urllib import unquote
 from math import log
 from contextlib import contextmanager
 from warnings import warn
 import yams.schema
-from logilab.common.testlib import TestCase, InnerTest
+from logilab.common.testlib import TestCase, InnerTest, Tags
 from logilab.common.pytest import nocoverage, pause_tracing, resume_tracing
 from logilab.common.debugger import Debugger
 from logilab.common.umessage import message_from_string
 from logilab.common.decorators import cached, classproperty, clear_cache
 from logilab.common.deprecation import deprecated
 from cubicweb import cwconfig, devtools, web, server
 from cubicweb.dbapi import ProgrammingError, DBAPISession, repo_connect
 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 fake, htmlparser
+from cubicweb.devtools import BASE_URL, fake, htmlparser
 from cubicweb.utils import json
 # low-level utilities ##########################################################
 class CubicWebDebugger(Debugger):
 """
 if after is None:
 after = before
 return center - before <= line_no <= center + after
 def unprotected_entities(schema, strict=False):
 """returned a set of each non final entity type, excluding "system" entities
 (eg CWGroup, CWUser...)
 """
 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()
 MAILBOX.append(Email(recipients, msg))
 cwconfig.SMTP = MockSMTP
+class TestCaseConnectionProxy(object):
+"""thin wrapper around `cubicweb.dbapi.Connection` context-manager
+used in CubicWebTC (cf. `cubicweb.devtools.testlib.CubicWebTC.login` method)
+It just proxies to the default connection context manager but
+restores the original connection on exit.
+"""
+def __init__(self, testcase, cnx):
+self.testcase = testcase
+self.cnx = cnx
+def __getattr__(self, attrname):
+return getattr(self.cnx, attrname)
+def __enter__(self):
+return self.cnx.__enter__()
+def __exit__(self, exctype, exc, tb):
+try:
+return self.cnx.__exit__(exctype, exc, tb)
+finally:
+self.cnx.close()
+self.testcase.restore_connection()
 # base class for cubicweb tests requiring a full cw environments ###############
 class CubicWebTC(TestCase):
 """abstract class for test using an apptest environment
 * `admpassword`, password of the admin user
 """
 appid = 'data'
 configcls = devtools.ApptestConfiguration
 reset_schema = reset_vreg = False # reset schema / vreg between tests
+tags = TestCase.tags | Tags('cubicweb', 'cw_repo')
 @classproperty
 def config(cls):
-"""return the configuration object. Configuration is cached on the test
+"""return the configuration object
-class.
+Configuration is cached on the test class.
 """
 try:
 return cls.__dict__['_config']
 except KeyError:
-config = cls._config = cls.configcls(cls.appid)
+home = join(dirname(sys.modules[cls.__module__].__file__), cls.appid)
+config = cls._config = cls.configcls(cls.appid, apphome=home)
 config.mode = 'test'
 return config
 @classmethod
 def init_config(cls, config):
-"""configuration initialization hooks. You may want to override this."""
+"""configuration initialization hooks.
+You may only want to override here the configuraton logic.
+Otherwise, consider to use a different :class:`ApptestConfiguration`
+defined in the `configcls` class attribute"""
 source = config.sources()['system']
 cls.admlogin = unicode(source['db-user'])
 cls.admpassword = source['db-password']
 # uncomment the line below if you want rql queries to be logged
 #config.global_set_option('query-log-file',
 or os.environ.get('LOGNAME'))
 config.global_set_option('sender-addr', send_to)
 config.global_set_option('default-dest-addrs', send_to)
 config.global_set_option('sender-name', 'cubicweb-test')
 config.global_set_option('sender-addr', 'cubicweb-test@logilab.fr')
+# default_base_url on config class isn't enough for TestServerConfiguration
+config.global_set_option('base-url', config.default_base_url())
 # web resources
-config.global_set_option('base-url', devtools.BASE_URL)
 try:
 config.global_set_option('embed-allowed', re.compile('.*'))
 except: # not in server only configuration
 pass
 server.set_debug(debugmode)
 # default test setup and teardown #########################################
 def setUp(self):
+# monkey patch send mail operation so emails are sent synchronously
+self._old_mail_commit_event = SendMailOp.commit_event
+SendMailOp.commit_event = SendMailOp.sendmails
 pause_tracing()
 previous_failure = self.__class__.__dict__.get('_repo_init_failed')
 if previous_failure is not None:
-self.skip('repository is not initialised: %r' % previous_failure)
+self.skipTest('repository is not initialised: %r' % previous_failure)
 try:
 self._init_repo()
 except Exception, ex:
 self.__class__._repo_init_failed = ex
 raise
 if not self.cnx._closed:
 self.cnx.rollback()
 for cnx in self._cnxs:
 if not cnx._closed:
 cnx.close()
+SendMailOp.commit_event = self._old_mail_commit_event
 def setup_database(self):
 """add your database setup code by overriding this method"""
 # user / session management ###############################################
 user = req.create_entity('CWUser', login=unicode(login),
 upassword=password, **kwargs)
 req.execute('SET X in_group G WHERE X eid %%(x)s, G name IN(%s)'
 % ','.join(repr(g) for g in groups),
 {'x': user.eid})
-user.clear_related_cache('in_group', 'subject')
+user.cw_clear_relation_cache('in_group', 'subject')
 if commit:
 req.cnx.commit()
 return user
 def login(self, login, **kwargs):
 """return a connection for the given login/password"""
 if login == self.admlogin:
 self.restore_connection()
-else:
+# definitly don't want autoclose when used as a context manager
-if not kwargs:
+return self.cnx
-kwargs['password'] = str(login)
+autoclose = kwargs.pop('autoclose', True)
-self.cnx = repo_connect(self.repo, unicode(login), **kwargs)
+if not kwargs:
-self.websession = DBAPISession(self.cnx)
+kwargs['password'] = str(login)
-self._cnxs.append(self.cnx)
+self.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:
 raise
 publisher.error_handler = raise_error_handler
 return publisher
 requestcls = fake.FakeRequest
-def request(self, *args, **kwargs):
+def request(self, rollbackfirst=False, **kwargs):
 """return a web ui request"""
 req = self.requestcls(self.vreg, form=kwargs)
+if rollbackfirst:
+self.websession.cnx.rollback()
 req.set_session(self.websession)
 return req
 def remote_call(self, fname, *args):
 """remote json call simulation"""
 req.cnx.commit()
 except web.Redirect:
 req.cnx.commit()
 raise
 return result
+def req_from_url(self, url):
+"""parses `url` and builds the corresponding CW-web request
+req.form will be setup using the url's query string
+"""
+req = self.request()
+if isinstance(url, unicode):
+url = url.encode(req.encoding) # req.setup_params() expects encoded strings
+querystring = urlparse.urlparse(url)[-2]
+params = urlparse.parse_qs(querystring)
+req.setup_params(params)
+return req
+def url_publish(self, url):
+"""takes `url`, uses application's app_resolver to find the
+appropriate controller, and publishes the result.
+This should pretty much correspond to what occurs in a real CW server
+except the apache-rewriter component is not called.
+"""
+req = self.req_from_url(url)
+ctrlid, rset = self.app.url_resolver.process(req, req.relative_path(False))
+return self.ctrl_publish(req, ctrlid)
 def expect_redirect(self, callback, req):
 """call the given callback with req as argument, expecting to get a
 Redirect exception
 """
 def assertAuthSuccess(self, req, origsession, nbsessions=1):
 sh = self.app.session_handler
 path, params = self.expect_redirect(lambda x: self.app.connect(x), req)
 session = req.session
-self.assertEquals(len(self.open_sessions), nbsessions, self.open_sessions)
+self.assertEqual(len(self.open_sessions), nbsessions, self.open_sessions)
-self.assertEquals(session.login, origsession.login)
+self.assertEqual(session.login, origsession.login)
-self.assertEquals(session.anonymous_session, False)
+self.assertEqual(session.anonymous_session, False)
-self.assertEquals(path, 'view')
+self.assertEqual(path, 'view')
-self.assertEquals(params, {'__message': 'welcome %s !' % req.user.login})
+self.assertEqual(params, {'__message': 'welcome %s !' % req.user.login})
 def assertAuthFailure(self, req, nbsessions=0):
 self.app.connect(req)
 self.assertIsInstance(req.session, DBAPISession)
-self.assertEquals(req.session.cnx, None)
+self.assertEqual(req.session.cnx, None)
-self.assertEquals(req.cnx, None)
+self.assertEqual(req.cnx, None)
-self.assertEquals(len(self.open_sessions), nbsessions)
+self.assertEqual(len(self.open_sessions), nbsessions)
 clear_cache(req, 'get_authorization')
 # content validation #######################################################
 # validators are used to validate (XML, DTD, whatever) view's content
 def view(self, vid, rset=None, req=None, template='main-template',
 **kwargs):
 """This method tests the view `vid` on `rset` using `template`
-If no error occured while rendering the view, the HTML is analyzed
+If no error occurred while rendering the view, the HTML is analyzed
 and parsed.
 :returns: an instance of `cubicweb.devtools.htmlparser.PageInfo`
 encapsulation the generated HTML
 """
 kwargs['rset'] = rset
 viewsreg = self.vreg['views']
 view = viewsreg.select(vid, req, **kwargs)
 # set explicit test description
 if rset is not None:
-self.set_description("testing %s, mod=%s (%s)" % (
+self.set_description("testing vid=%s defined in %s with (%s)" % (
 vid, view.__module__, rset.printable_rql()))
 else:
-self.set_description("testing %s, mod=%s (no rset)" % (
+self.set_description("testing vid=%s defined in %s without rset" % (
 vid, view.__module__))
 if template is None: # raw view testing, no template
 viewfunc = view.render
 else:
 kwargs['view'] = view
 def _test_view(self, viewfunc, view, template='main-template', kwargs={}):
 """this method does the actual call to the view
-If no error occured while rendering the view, the HTML is analyzed
+If no error occurred while rendering the view, the HTML is analyzed
 and parsed.
 :returns: an instance of `cubicweb.devtools.htmlparser.PageInfo`
 encapsulation the generated HTML
 """
 else:
 default_validator = None
 validatorclass = self.content_type_validators.get(view.content_type,
 default_validator)
 if validatorclass is None:
-return None
+return output.strip()
 validator = validatorclass()
 if isinstance(validator, htmlparser.DTDValidator):
 # XXX remove <canvas> used in progress widget, unknown in html dtd
 output = re.sub('<canvas.*?></canvas>', '', output)
 return validator.parse_string(output.strip())
 class AutoPopulateTest(CubicWebTC):
 """base class for test with auto-populating of the database"""
 __abstract__ = True
+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
 application_rql = []
 try:
 cu.execute(rql, args)
 except ValidationError, ex:
 # failed to satisfy some constraint
 print 'error in automatic db population', ex
+self.session.commit_state = None # reset uncommitable flag
 self.post_populate(cu)
 self.commit()
 def iter_individual_rsets(self, etypes=None, limit=None):
 etypes = etypes or self.to_test_etypes()
 # concrete class for automated application testing  ############################
 class AutomaticWebTest(AutoPopulateTest):
 """import this if you wan automatic tests to be ran"""
+tags = AutoPopulateTest.tags | Tags('web', 'generated')
 def setUp(self):
 AutoPopulateTest.setUp(self)
 # access to self.app for proper initialization of the authentication
 # machinery (else some views may fail)
 self.app

branch	oldstable
changeset 6665	90f2f20367bc
parent 6589	47cd31fd206b
child 6590	37b7f4df46b3
child 6720	43a38c093f6f