cubicweb: comparison devtools/testlib.py

equal deleted inserted replaced

-:0939ad2edf63
+:6557833657d6
 # You should have received a copy of the GNU Lesser General Public License along
 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.
 """this module contains base classes and utilities for cubicweb tests"""
 from __future__ import print_function
-__docformat__ = "restructuredtext en"
 import sys
 import re
 from os.path import dirname, join, abspath
 from math import log
 from contextlib import contextmanager
-from warnings import warn
 from itertools import chain
 from six import text_type, string_types
 from six.moves import range
 from six.moves.urllib.parse import urlparse, parse_qs, unquote as urlunquote
 from logilab.common.decorators import cached, classproperty, clear_cache, iclassmethod
 from logilab.common.deprecation import deprecated, class_deprecated
 from logilab.common.shellutils import getlogin
 from cubicweb import (ValidationError, NoSelectableObject, AuthenticationError,
-ProgrammingError, BadConnectionId)
+BadConnectionId)
 from cubicweb import cwconfig, devtools, web, server, repoapi
 from cubicweb.utils import json
 from cubicweb.sobjects import notification
 from cubicweb.web import Redirect, application, eid_param
 from cubicweb.server.hook import SendMailOp
 from cubicweb.server.session import Session
 from cubicweb.devtools import SYSTEM_ENTITIES, SYSTEM_RELATIONS, VIEW_VALIDATORS
 from cubicweb.devtools import fake, htmlparser, DEFAULT_EMPTY_DB_ID
-from cubicweb.utils import json
 # low-level utilities ##########################################################
 class CubicWebDebugger(Debugger):
 """special debugger class providing a 'view' function which saves some
 data = self._getval(arg)
 with open('/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 after is None: after = before
 """
 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...)
 """
 protected_entities = yams.schema.BASE_TYPES
 else:
 protected_entities = yams.schema.BASE_TYPES.union(SYSTEM_ENTITIES)
 return set(schema.entities()) - protected_entities
 class JsonValidator(object):
 def parse_string(self, data):
 return json.loads(data.decode('ascii'))
 @contextmanager
 def real_error_handling(app):
 """By default, CubicWebTC `app` attribute (ie the publisher) is monkey
 patched so that unexpected error are raised rather than going through the
 # return the app
 yield app
 # restore
 app.error_handler = fake_error_handler
 # email handling, to test emails sent by an application ########################
 MAILBOX = []
 class Email(object):
 """you'll get instances of Email into MAILBOX during tests that trigger
 some notification.
 def __repr__(self):
 return '<Email to %s with subject %s>' % (','.join(self.recipients),
 self.message.get('Subject'))
 # the trick to get email into MAILBOX instead of actually sent: monkey patch
 # cwconfig.SMTP object
 class MockSMTP:
 def __init__(self, server, port):
 pass
 def close(self):
 pass
 def sendmail(self, fromaddr, recipients, msg):
 MAILBOX.append(Email(fromaddr, recipients, msg))
 cwconfig.SMTP = MockSMTP
 schema=1)
 yield mih
 cnx.commit()
 # base class for cubicweb tests requiring a full cw environments ###############
 class CubicWebTC(TestCase):
 """abstract class for test using an apptest environment
 def _close_access(self):
 while self._open_access:
 try:
 self._open_access.pop().close()
 except BadConnectionId:
-continue # already closed
+continue  # already closed
 @property
 def session(self):
 """return admin session"""
 return self._admin_session
-#XXX this doesn't need to a be classmethod anymore
+# XXX this doesn't need to a be classmethod anymore
 def _init_repo(self):
 """init the repository and connection to it.
 """
 # get or restore and working db.
 db_handler = devtools.get_test_db_handler(self.config, self.init_config)
 self.repo = db_handler.get_repo(startup=True)
 # get an admin session (without actual login)
 login = text_type(db_handler.config.default_admin_config['login'])
 self.admin_access = self.new_access(login)
 self._admin_session = self.admin_access._session
 # config management ########################################################
 @classproperty
 def config(cls):
 home = abspath(join(dirname(sys.modules[cls.__module__].__file__), cls.appid))
 config = cls._config = cls.configcls(cls.appid, apphome=home)
 config.mode = 'test'
 return config
-@classmethod # XXX could be turned into a regular method
+@classmethod  # XXX could be turned into a regular method
 def init_config(cls, config):
 """configuration initialization hooks.
 You may only want to override here the configuraton logic.
 """
 admincfg = config.default_admin_config
 cls.admlogin = text_type(admincfg['login'])
 cls.admpassword = admincfg['password']
 # uncomment the line below if you want rql queries to be logged
-#config.global_set_option('query-log-file',
+# config.global_set_option('query-log-file',
-#                         '/tmp/test_rql_log.' + `os.getpid()`)
+#                          '/tmp/test_rql_log.' + `os.getpid()`)
 config.global_set_option('log-file', None)
 # set default-dest-addrs to a dumb email address to avoid mailbox or
 # mail queue pollution
 config.global_set_option('default-dest-addrs', ['whatever'])
-send_to =  '%s@logilab.fr' % getlogin()
+send_to = '%s@logilab.fr' % getlogin()
 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
 try:
 config.global_set_option('embed-allowed', re.compile('.*'))
-except Exception: # not in server only configuration
+except Exception:  # not in server only configuration
 pass
 @property
 def vreg(self):
 return self.repo.vreg
 # global resources accessors ###############################################
 @property
 def schema(self):
 self.__class__._repo_init_failed = ex
 raise
 self.addCleanup(self._close_access)
 self.config.set_anonymous_allowed(self.anonymous_allowed)
 self.setup_database()
-MAILBOX[:] = [] # reset mailbox
+MAILBOX[:] = []  # reset mailbox
 def tearDown(self):
 # XXX hack until logilab.common.testlib is fixed
 if self._admin_session is not None:
 self.repo.close(self._admin_session.sessionid)
 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"""
 if req is None:
 return self.request().user
 else:
 return req.user
-@iclassmethod # XXX turn into a class method
+@iclassmethod  # XXX turn into a class method
 def create_user(self, req, login=None, groups=('users',), password=None,
 email=None, commit=True, **kwargs):
 """create and return a new user entity"""
 if password is None:
 password = login
 req.create_entity('EmailAddress', address=text_type(email),
 reverse_primary_email=user)
 user.cw_clear_relation_cache('in_group', 'subject')
 if commit:
 try:
-req.commit() # req is a session
+req.commit()  # req is a session
 except AttributeError:
 req.cnx.commit()
 return user
 # other utilities #########################################################
 @contextmanager
 def temporary_appobjects(self, *appobjects):
 def assertPossibleTransitions(self, entity, expected):
 transitions = entity.cw_adapt_to('IWorkflowable').possible_transitions()
 self.assertListEqual(sorted(tr.name for tr in transitions),
 sorted(expected))
 # views and actions registries inspection ##################################
 def pviews(self, req, rset):
 return sorted((a.__regid__, a.__class__)
 for a in self.vreg['views'].possible_views(req, rset=rset))
 def _test_action(self, action):
 class fake_menu(list):
 @property
 def items(self):
 return self
 class fake_box(object):
 def action_link(self, action, **kwargs):
 return (action.title, action.url())
 submenu = fake_menu()
 action.fill_menu(fake_box(), submenu)
 and not isinstance(view, class_deprecated)]
 if views:
 try:
 view = viewsvreg._select_best(views, req, rset=rset)
 if view is None:
-raise NoSelectableObject((req,), {'rset':rset}, views)
+raise NoSelectableObject((req,), {'rset': rset}, views)
 if view.linkable():
 yield view
 else:
 not_selected(self.vreg, view)
 # else the view is expected to be used as subview and should
 if view.category == 'startupview':
 yield view.__regid__
 else:
 not_selected(self.vreg, view)
 # web ui testing utilities #################################################
 @property
 @cached
 def app(self):
 """return a cubicweb publisher"""
 publisher = application.CubicWebPublisher(self.repo, self.config)
 def raise_error_handler(*args, **kwargs):
 raise
 publisher.error_handler = raise_error_handler
 return publisher
 @deprecated('[3.19] use the .remote_calling method')
 def remote_call(self, fname, *args):
 * `entity_field_dicts`, list of (entity, dictionary) where dictionary contains name:value
 for fields that are not tied to the given entity
 """
 assert field_dict or entity_field_dicts, \
 'field_dict and entity_field_dicts arguments must not be both unspecified'
 if field_dict is None:
 field_dict = {}
 form = {'__form_id': formid}
 fields = []
 for field, value in field_dict.items():
 fields.append(field)
 form[field] = value
 def _add_entity_field(entity, field, value):
 entity_fields.append(field)
 form[eid_param(field, entity.eid)] = value
 for entity, field_dict in entity_field_dicts:
 if '__maineid' not in form:
 form['__maineid'] = entity.eid
 entity_fields = []
 form.setdefault('eid', []).append(entity.eid)
 req.form will be setup using the url's query string
 """
 req = self.request(url=url)
 if isinstance(url, unicode):
-url = url.encode(req.encoding) # req.setup_params() expects encoded strings
+url = url.encode(req.encoding)  # req.setup_params() expects encoded strings
 querystring = urlparse(url)[-2]
 params = parse_qs(querystring)
 req.setup_params(params)
 return req
 req.form will be setup using the url's query string
 """
 with self.admin_access.web_request(url=url) as req:
 if isinstance(url, unicode):
-url = url.encode(req.encoding) # req.setup_params() expects encoded strings
+url = url.encode(req.encoding)  # req.setup_params() expects encoded strings
 querystring = urlparse(url)[-2]
 params = parse_qs(querystring)
 req.setup_params(params)
 yield req
 path = location
 params = {}
 else:
 cleanup = lambda p: (p[0], urlunquote(p[1]))
 params = dict(cleanup(p.split('=', 1)) for p in params.split('&') if p)
-if path.startswith(req.base_url()): # may be relative
+if path.startswith(req.base_url()):  # may be relative
 path = path[len(req.base_url()):]
 return path, params
 def expect_redirect(self, callback, req):
 """call the given callback with req as argument, expecting to get a
 def expect_redirect_handle_request(self, req, path='edit'):
 """call the publish method of the application publisher, expecting to
 get a Redirect exception
 """
-result = self.app_handle_request(req, path)
+self.app_handle_request(req, path)
-self.assertTrue(300 <= req.status_out <400, req.status_out)
+self.assertTrue(300 <= req.status_out < 400, req.status_out)
 location = req.get_response_header('location')
 return self._parse_location(req, location)
 @deprecated("[3.15] expect_redirect_handle_request is the new and better way"
 " (beware of small semantic changes)")
 def expect_redirect_publish(self, *args, **kwargs):
 return self.expect_redirect_handle_request(*args, **kwargs)
 def set_auth_mode(self, authmode, anonuser=None):
 self.set_option('auth-mode', authmode)
 self.set_option('anonymous-user', anonuser)
 if anonuser is None:
 content_type_validators = {
 # maps MIME type : validator name
 #
 # do not set html validators here, we need HTMLValidator for html
 # snippets
-#'text/html': DTDValidator,
+# 'text/html': DTDValidator,
-#'application/xhtml+xml': DTDValidator,
+# 'application/xhtml+xml': DTDValidator,
 'application/xml': htmlparser.XMLValidator,
 'text/xml': htmlparser.XMLValidator,
 'application/json': JsonValidator,
 'text/plain': None,
 'text/comma-separated-values': None,
 'image/png': None,
 }
 # maps vid : validator name (override content_type_validators)
 vid_validators = dict((vid, htmlparser.VALMAP[valkey])
 for vid, valkey in VIEW_VALIDATORS.items())
 def view(self, vid, rset=None, req=None, template='main-template',
 **kwargs):
 """This method tests the view `vid` on `rset` using `template`
 self.set_description("testing vid=%s defined in %s with (%s)" % (
 vid, view.__module__, rql))
 else:
 self.set_description("testing vid=%s defined in %s without rset" % (
 vid, view.__module__))
-if template is None: # raw view testing, no template
+if template is None:  # raw view testing, no template
 viewfunc = view.render
 else:
 kwargs['view'] = view
 viewfunc = lambda **k: viewsreg.main_template(req, template,
 rset=rset, **kwargs)
 return self._test_view(viewfunc, view, template, kwargs)
 def _test_view(self, viewfunc, view, template='main-template', kwargs={}):
 """this method does the actual call to the view
 If no error occurred while rendering the view, the HTML is analyzed
 if isinstance(output, text_type):
 # XXX
 output = output.encode('utf-8')
 validator = self.get_validator(view, output=output)
 if validator is None:
-return output # return raw output if no validator is defined
+return output  # return raw output if no validator is defined
 if isinstance(validator, htmlparser.DTDValidator):
 # XXX remove <canvas> used in progress widget, unknown in html dtd
 output = re.sub('<canvas.*?></canvas>', '', output)
-return self.assertWellFormed(validator, output.strip(), context= view.__regid__)
+return self.assertWellFormed(validator, output.strip(), context=view.__regid__)
 def assertWellFormed(self, validator, content, context=None):
 try:
 return validator.parse_string(content)
 except Exception:
 from cubicweb.devtools.fill import insert_entity_queries, make_relations_queries
 # XXX cleanup unprotected_entities & all mess
 def how_many_dict(schema, cnx, how_many, skip):
 """given a schema, compute how many entities by type we need to be able to
 satisfy relations cardinality.
 The `how_many` argument tells how many entities of which type we want at
 relmap.setdefault((rschema, subj), []).append(str(obj))
 if card[1] in '1+' and card[0] in '1?':
 # reverse subj and obj in the above explanation
 relmap.setdefault((rschema, obj), []).append(str(subj))
 unprotected = unprotected_entities(schema)
-for etype in skip: # XXX (syt) duh? explain or kill
+for etype in skip:  # XXX (syt) duh? explain or kill
 unprotected.add(etype)
 howmanydict = {}
 # step 1, compute a base number of each entity types: number of already
 # existing entities of this type + `how_many`
 for etype in unprotected_entities(schema, strict=True):
 def custom_populate(self, how_many, cnx):
 pass
 def post_populate(self, cnx):
 pass
 @nocoverage
 def auto_populate(self, how_many):
 """this method populates the database with `how_many` entities
 of each possible type. It also inserts random relations between them
 try:
 cnx.execute(rql, args)
 except ValidationError as ex:
 # failed to satisfy some constraint
 print('error in automatic db population', ex)
-cnx.commit_state = None # reset uncommitable flag
+cnx.commit_state = None  # reset uncommitable flag
 self.post_populate(cnx)
 def iter_individual_rsets(self, etypes=None, limit=None):
 etypes = etypes or self.to_test_etypes()
 with self.admin_access.web_request() as req:
 except KeyError:
 etype2 = etype1
 # test a mixed query (DISTINCT/GROUP to avoid getting duplicate
 # X which make muledit view failing for instance (html validation fails
 # because of some duplicate "id" attributes)
-yield req.execute('DISTINCT Any X, MAX(Y) GROUPBY X WHERE X is %s, Y is %s' % (etype1, etype2))
+yield req.execute('DISTINCT Any X, MAX(Y) GROUPBY X WHERE X is %s, Y is %s' %
+(etype1, etype2))
 # test some application-specific queries if defined
 for rql in self.application_rql:
 yield req.execute(rql)
 def _test_everything_for(self, rset):
 rset.req.reset_headers(), 'main-template')
 # We have to do this because some views modify the
 # resultset's syntax tree
 rset = backup_rset
 for action in self.list_actions_for(rset):
-yield InnerTest(self._testname(rset, action.__regid__, 'action'), self._test_action, action)
+yield InnerTest(self._testname(rset, action.__regid__, 'action'),
+self._test_action, action)
 for box in self.list_boxes_for(rset):
 w = [].append
 yield InnerTest(self._testname(rset, box.__regid__, 'box'), box.render, w)
 @staticmethod
 # access to self.app for proper initialization of the authentication
 # machinery (else some views may fail)
 self.app
-## one each
 def test_one_each_config(self):
 self.auto_populate(1)
 for rset in self.iter_automatic_rsets(limit=1):
 for testargs in self._test_everything_for(rset):
 yield testargs
-## ten each
 def test_ten_each_config(self):
 self.auto_populate(10)
 for rset in self.iter_automatic_rsets(limit=10):
 for testargs in self._test_everything_for(rset):
 yield testargs
-## startup views
 def test_startup_views(self):
 for vid in self.list_startup_views():
 with self.admin_access.web_request() as req:
 yield self.view, vid, None, req

changeset 10974	6557833657d6
parent 10937	eb05348b0e2d
child 10997	da712d3f0601