10 import os

    12 import re

    13 from urllib import unquote

    16 import simplejson

    17 

    18 import yams.schema

    19 

    20 from logilab.common.testlib import TestCase, InnerTest

    21 from logilab.common.pytest import nocoverage, pause_tracing, resume_tracing

    46 

    47 

    48 def line_context_filter(line_no, center, before=3, after=None):

    49     """return true if line are in context

    50 

    51     if after is None: after = before

    52     """

    53     if after is None:

    54         after = before

    55     return center - before <= line_no <= center + after

    56 

    57 

    58 def unprotected_entities(schema, strict=False):

    59     """returned a set of each non final entity type, excluding "system" entities

    60     (eg CWGroup, CWUser...)

    61     """

    62     if strict:

    63         protected_entities = yams.schema.BASE_TYPES

    64     else:

    65         protected_entities = yams.schema.BASE_TYPES.union(SYSTEM_ENTITIES)

    66     return set(schema.entities()) - protected_entities

    67 

    68 

    69 def get_versions(self, checkversions=False):

    70     """return the a dictionary containing cubes used by this instance

    71     as key with their version as value, including cubicweb version. This is a

    72     public method, not requiring a session id.

    73 

    74     replace Repository.get_versions by this method if you don't want versions

    75     checking

    76     """

    77     vcconf = {'cubicweb': self.config.cubicweb_version()}

    78     self.config.bootstrap_cubes()

    79     for pk in self.config.cubes():

    80         version = self.config.cube_version(pk)

    81         vcconf[pk] = version

    82     self.config._cubes = None

    83     return vcconf

    84 

    85 

    86 def refresh_repo(repo):

    87     devtools.reset_test_database(repo.config)

    88     for pool in repo.pools:

    89         pool.reconnect()

    90     repo._type_source_cache = {}

    91     repo._extid_cache = {}

    92     repo.querier._rql_cache = {}

    93     for source in repo.sources:

    94         source.reset_caches()

    95 

    96 

    97 # email handling, to test emails sent by an application ########################

    98 

    99 MAILBOX = []

   100 

   101 class Email:

   102     """you'll get instances of Email into MAILBOX during tests that trigger

   103     some notification.

   104 

   105     * `msg` is the original message object

   106 

   107     * `recipients` is a list of email address which are the recipients of this

   108       message

   109     """

   110     def __init__(self, recipients, msg):

   111         self.recipients = recipients

   112         self.msg = msg

   113 

   114     @property

   115     def message(self):

   116         return message_from_string(self.msg)

   117 

   118     @property

   119     def subject(self):

   120         return self.message.get('Subject')

   121 

   122     @property

   123     def content(self):

   124         return self.message.get_payload(decode=True)

   125 

   126     def __repr__(self):

   127         return '<Email to %s with subject %s>' % (','.join(self.recipients),

   128                                                   self.message.get('Subject'))

   129 

   130 # the trick to get email into MAILBOX instead of actually sent: monkey patch

   131 # cwconfig.SMTP object

   132 class MockSMTP:

   133     def __init__(self, server, port):

   134         pass

   135     def close(self):

   136         pass

   137     def sendmail(self, helo_addr, recipients, msg):

   138         MAILBOX.append(Email(recipients, msg))

   139 

   140 cwconfig.SMTP = MockSMTP

   141 

   142 

   143 # base class for cubicweb tests requiring a full cw environments ###############

   144 

   145 class CubicWebTC(TestCase):

   146     """abstract class for test using an apptest environment

   147 

   148     attributes:

   149     `vreg`, the vregistry

   150     `schema`, self.vreg.schema

   151     `config`, cubicweb configuration

   152     `cnx`, dbapi connection to the repository using an admin user

   153     `session`, server side session associated to `cnx`

   154     `app`, the cubicweb publisher (for web testing)

   155     `repo`, the repository object

   156 

   157     `admlogin`, login of the admin user

   158     `admpassword`, password of the admin user

   159 

   160     """

   161     appid = 'data'

   162     configcls = devtools.ApptestConfiguration

   163 

   164     @classproperty

   165     def config(cls):

   166         """return the configuration object. Configuration is cached on the test

   167         class.

   168         """

   169         try:

   170             return cls.__dict__['_config']

   171         except KeyError:

   172             config = cls._config = cls.configcls(cls.appid)

   173             config.mode = 'test'

   174             return config

   175 

   176     @classmethod

   177     def init_config(cls, config):

   178         """configuration initialization hooks. You may want to override this."""

   179         source = config.sources()['system']

   180         cls.admlogin = unicode(source['db-user'])

   181         cls.admpassword = source['db-password']

   182         # uncomment the line below if you want rql queries to be logged

   183         #config.global_set_option('query-log-file', '/tmp/test_rql_log.' + `os.getpid()`)

   184         config.global_set_option('log-file', None)

   185         # set default-dest-addrs to a dumb email address to avoid mailbox or

   186         # mail queue pollution

   187         config.global_set_option('default-dest-addrs', ['whatever'])

   188         try:

   189             send_to =  '%s@logilab.fr' % os.getlogin()

   190         except OSError:

   191             send_to =  '%s@logilab.fr' % (os.environ.get('USER')

   192                                           or os.environ.get('USERNAME')

   193                                           or os.environ.get('LOGNAME'))

   194         config.global_set_option('sender-addr', send_to)

   195         config.global_set_option('default-dest-addrs', send_to)

   196         config.global_set_option('sender-name', 'cubicweb-test')

   197         config.global_set_option('sender-addr', 'cubicweb-test@logilab.fr')

   198         # web resources

   199         config.global_set_option('base-url', devtools.BASE_URL)

   200         try:

   201             config.global_set_option('embed-allowed', re.compile('.*'))

   202         except: # not in server only configuration

   203             pass

   204 

   205     @classmethod

   206     def _init_repo(cls):

   207         """init the repository and connection to it.

   208 

   209         Repository and connection are cached on the test class. Once

   210         initialized, we simply reset connections and repository caches.

   211         """

   212         if not 'repo' in cls.__dict__:

   213             cls._build_repo()

   214         else:

   215             cls.cnx.rollback()

   216             cls._refresh_repo()

   217 

   218     @classmethod

   219     def _build_repo(cls):

   220         cls.repo, cls.cnx = devtools.init_test_database(config=cls.config)

   221         cls.init_config(cls.config)

   222         cls.vreg = cls.repo.vreg

   223         cls._orig_cnx = cls.cnx

   224         cls.config.repository = lambda x=None: cls.repo

   225         # necessary for authentication tests

   226         cls.cnx.login = cls.admlogin

   227         cls.cnx.password = cls.admpassword

   228 

   229     @classmethod

   230     def _refresh_repo(cls):

   231         refresh_repo(cls.repo)

   232 

   233     # global resources accessors ###############################################

   234 

   235     @property

   236     def schema(self):

   237         """return the application schema"""

   238         return self.vreg.schema

   239 

   240     @property

   241     def session(self):

   242         """return current server side session (using default manager account)"""

   243         return self.repo._sessions[self.cnx.sessionid]

   244 

   245     @property

   246     def adminsession(self):

   247         """return current server side session (using default manager account)"""

   248         return self.repo._sessions[self._orig_cnx.sessionid]

   249 

   250     def set_option(self, optname, value):

   251         self.config.global_set_option(optname, value)

   252 

   253     def set_debug(self, debugmode):

   254         server.set_debug(debugmode)

   255 

   256     # default test setup and teardown #########################################

   257 

   258     def setUp(self):

   259         pause_tracing()

   260         self._init_repo()

   261         resume_tracing()

   262         self.setup_database()

   263         self.commit()

   264         MAILBOX[:] = [] # reset mailbox

   265 

   266     def setup_database(self):

   267         """add your database setup code by overriding this method"""

   268 

   269     # user / session management ###############################################

   270 

   271     def user(self, req=None):

   272         """return the application schema"""

   273         if req is None:

   274             req = self.request()

   275             return self.cnx.user(req)

   276         else:

   277             return req.user

   278 

   279     def create_user(self, login, groups=('users',), password=None, req=None,

   280                     commit=True):

   281         """create and return a new user entity"""

   282         if password is None:

   283             password = login.encode('utf8')

   284         cursor = self._orig_cnx.cursor(req or self.request())

   285         rset = cursor.execute('INSERT CWUser X: X login %(login)s, X upassword %(passwd)s,'

   286                               'X in_state S WHERE S name "activated"',

   287                               {'login': unicode(login), 'passwd': password})

   288         user = rset.get_entity(0, 0)

   289         cursor.execute('SET X in_group G WHERE X eid %%(x)s, G name IN(%s)'

   290                        % ','.join(repr(g) for g in groups),

   291                        {'x': user.eid}, 'x')

   292         user.clear_related_cache('in_group', 'subject')

   293         if commit:

   294             self._orig_cnx.commit()

   295         return user

   296 

   297     def login(self, login, password=None):

   298         """return a connection for the given login/password"""

   299         if login == self.admlogin:

   300             self.restore_connection()

   301         else:

   302             self.cnx = repo_connect(self.repo, unicode(login),

   303                                     password or str(login),

   304                                     ConnectionProperties('inmemory'))

   305         if login == self.vreg.config.anonymous_user()[0]:

   306             self.cnx.anonymous_connection = True

   307         return self.cnx

   308 

   309     def restore_connection(self):

   310         if not self.cnx is self._orig_cnx:

   311             try:

   312                 self.cnx.close()

   313             except ProgrammingError:

   314                 pass # already closed

   315         self.cnx = self._orig_cnx

   316 

   317     # db api ##################################################################

   318 

   319     @nocoverage

   320     def cursor(self, req=None):

   321         return self.cnx.cursor(req or self.request())

   322 

   323     @nocoverage

   324     def execute(self, rql, args=None, eidkey=None, req=None):

   325         """executes <rql>, builds a resultset, and returns a couple (rset, req)

   326         where req is a FakeRequest

   327         """

   328         req = req or self.request(rql=rql)

   329         return self.cnx.cursor(req).execute(unicode(rql), args, eidkey)

   330 

   331     @nocoverage

   332     def commit(self):

   333         self.cnx.commit()

   334 

   335     @nocoverage

   336     def rollback(self):

   337         try:

   338             self.cnx.rollback()

   339         except ProgrammingError:

   340             pass

   341 

   342     # # server side db api #######################################################

   343 

   344     def sexecute(self, rql, args=None, eid_key=None):

   345         self.session.set_pool()

   346         return self.session.execute(rql, args, eid_key)

   347 

   348     # other utilities #########################################################

   349 

   350     def entity(self, rql, args=None, eidkey=None, req=None):

   351         return self.execute(rql, args, eidkey, req=req).get_entity(0, 0)

   352 

   353     def add_entity(self, etype, req=None, **kwargs):

   354         rql = ['INSERT %s X' % etype]

   355         # dict for replacement in RQL Request

   356         args = {}

   357         if kwargs:

   358             rql.append(':')

   359             # dict to define new entities variables

   360             entities = {}

   361             # assignement part of the request

   362             sub_rql = []

   363             for key, value in kwargs.iteritems():

   364                 # entities

   365                 if hasattr(value, 'eid'):

   366                     new_value = "%s__" % key.upper()

   367                     entities[new_value] = value.eid

   368                     args[new_value] = value.eid

   369 

   370                     sub_rql.append("X %s %s" % (key, new_value))

   371                 # final attributes

   372                 else:

   373                     sub_rql.append('X %s %%(%s)s' % (key, key))

   374                     args[key] = value

   375             rql.append(', '.join(sub_rql))

   376             if entities:

   377                 rql.append('WHERE')

   378                 # WHERE part of the request (to link entity to they eid)

   379                 sub_rql = []

   380                 for key, value in entities.iteritems():

   381                     sub_rql.append("%s eid %%(%s)s" % (key, key))

   382                 rql.append(', '.join(sub_rql))

   383         return self.execute(' '.join(rql), args, req=req).get_entity(0, 0)

   384 

   385     # vregistry inspection utilities ###########################################

   386 

   387     def pviews(self, req, rset):

   388         return sorted((a.id, a.__class__)

   389                       for a in self.vreg['views'].possible_views(req, rset=rset))

   390 

   391     def pactions(self, req, rset,

   392                  skipcategories=('addrelated', 'siteactions', 'useractions')):

   393         return [(a.id, a.__class__)

   394                 for a in self.vreg['actions'].possible_vobjects(req, rset=rset)

   395                 if a.category not in skipcategories]

   396 

   397     def pactions_by_cats(self, req, rset, categories=('addrelated',)):

   398         return [(a.id, a.__class__)

   399                 for a in self.vreg['actions'].possible_vobjects(req, rset=rset)

   400                 if a.category in categories]

   401 

   402     def pactionsdict(self, req, rset,

   403                      skipcategories=('addrelated', 'siteactions', 'useractions')):

   404         res = {}

   405         for a in self.vreg['actions'].possible_vobjects(req, rset=rset):

   406             if a.category not in skipcategories:

   407                 res.setdefault(a.category, []).append(a.__class__)

   408         return res

   409 

   410     def list_views_for(self, rset):

   411         """returns the list of views that can be applied on `rset`"""

   412         req = rset.req

   413         only_once_vids = ('primary', 'secondary', 'text')

   414         req.data['ex'] = ValueError("whatever")

   415         viewsvreg = self.vreg['views']

   416         for vid, views in viewsvreg.items():

   417             if vid[0] == '_':

   418                 continue

   419             if rset.rowcount > 1 and vid in only_once_vids:

   420                 continue

   421             views = [view for view in views

   422                      if view.category != 'startupview'

   423                      and not issubclass(view, NotificationView)]

   424             if views:

   425                 try:

   426                     view = viewsvreg._select_best(views, req, rset=rset)

   427                     if view.linkable():

   428                         yield view

   429                     else:

   430                         not_selected(self.vreg, view)

   431                     # else the view is expected to be used as subview and should

   432                     # not be tested directly

   433                 except NoSelectableObject:

   434                     continue

   435 

   436     def list_actions_for(self, rset):

   437         """returns the list of actions that can be applied on `rset`"""

   438         req = rset.req

   439         for action in self.vreg['actions'].possible_objects(req, rset=rset):

   440             yield action

   441 

   442     def list_boxes_for(self, rset):

   443         """returns the list of boxes that can be applied on `rset`"""

   444         req = rset.req

   445         for box in self.vreg['boxes'].possible_objects(req, rset=rset):

   446             yield box

   447 

   448     def list_startup_views(self):

   449         """returns the list of startup views"""

   450         req = self.request()

   451         for view in self.vreg['views'].possible_views(req, None):

   452             if view.category == 'startupview':

   453                 yield view.id

   454             else:

   455                 not_selected(self.vreg, view)

   456 

   457     # web ui testing utilities #################################################

   458 

   459     @property

   460     @cached

   461     def app(self):

   462         """return a cubicweb publisher"""

   463         return application.CubicWebPublisher(self.config, vreg=self.vreg)

   464 

   465     requestcls = fake.FakeRequest

   466     def request(self, *args, **kwargs):

   467         """return a web ui request"""

   468         req = self.requestcls(self.vreg, form=kwargs)

   469         req.set_connection(self.cnx)

   470         return req

   471 

   472     def remote_call(self, fname, *args):

   473         """remote json call simulation"""

   474         dump = simplejson.dumps

   475         args = [dump(arg) for arg in args]

   476         req = self.request(fname=fname, pageid='123', arg=args)

   477         ctrl = self.vreg['controllers'].select('json', req)

   478         return ctrl.publish(), req

   479 

   480     def publish(self, req):

   481         """call the publish method of the edit controller"""

   482         ctrl = self.vreg['controllers'].select('edit', req)

   483         try:

   484             result = ctrl.publish()

   485             req.cnx.commit()

   486         except web.Redirect:

   487             req.cnx.commit()

   488             raise

   489         return result

   490 

   491     def expect_redirect_publish(self, req):

   492         """call the publish method of the edit controller, expecting to get a

   493         Redirect exception."""

   494         try:

   495             self.publish(req)

   496         except web.Redirect, ex:

   497             try:

   498                 path, params = ex.location.split('?', 1)

   499             except:

   500                 path, params = ex.location, ""

   501             req._url = path

   502             cleanup = lambda p: (p[0], unquote(p[1]))

   503             params = dict(cleanup(p.split('=', 1)) for p in params.split('&') if p)

   504             return req.relative_path(False), params # path.rsplit('/', 1)[-1], params

   505         else:

   506             self.fail('expected a Redirect exception')

   507 

   508     # content validation #######################################################

   509 

   510     # validators are used to validate (XML, DTD, whatever) view's content

   511     # validators availables are :

   512     #  DTDValidator : validates XML + declared DTD

   513     #  SaxOnlyValidator : guarantees XML is well formed

   514     #  None : do not try to validate anything

   515     # validators used must be imported from from.devtools.htmlparser

   516     content_type_validators = {

   517         # maps MIME type : validator name

   518         #

   519         # do not set html validators here, we need HTMLValidator for html

   520         # snippets

   521         #'text/html': DTDValidator,

   522         #'application/xhtml+xml': DTDValidator,

   523         'application/xml': htmlparser.SaxOnlyValidator,

   524         'text/xml': htmlparser.SaxOnlyValidator,

   525         'text/plain': None,

   526         'text/comma-separated-values': None,

   527         'text/x-vcard': None,

   528         'text/calendar': None,

   529         'application/json': None,

   530         'image/png': None,

   531         }

   532     # maps vid : validator name (override content_type_validators)

   533     vid_validators = dict((vid, htmlparser.VALMAP[valkey])

   534                           for vid, valkey in VIEW_VALIDATORS.iteritems())

   535 

   536 

   537     def view(self, vid, rset=None, req=None, template='main-template',

   538              **kwargs):

   539         """This method tests the view `vid` on `rset` using `template`

   540 

   541         If no error occured while rendering the view, the HTML is analyzed

   542         and parsed.

   543 

   544         :returns: an instance of `cubicweb.devtools.htmlparser.PageInfo`

   545                   encapsulation the generated HTML

   546         """

   547         req = req or rset and rset.req or self.request()

   548         req.form['vid'] = vid

   549         kwargs['rset'] = rset

   550         viewsreg = self.vreg['views']

   551         view = viewsreg.select(vid, req, **kwargs)

   552         # set explicit test description

   553         if rset is not None:

   554             self.set_description("testing %s, mod=%s (%s)" % (

   555                 vid, view.__module__, rset.printable_rql()))

   556         else:

   557             self.set_description("testing %s, mod=%s (no rset)" % (

   558                 vid, view.__module__))

   559         if template is None: # raw view testing, no template

   560             viewfunc = view.render

   561         else:

   562             kwargs['view'] = view

   563             templateview = viewsreg.select(template, req, **kwargs)

   564             viewfunc = lambda **k: viewsreg.main_template(req, template,

   565                                                           **kwargs)

   566         kwargs.pop('rset')

   567         return self._test_view(viewfunc, view, template, kwargs)

   568 

   569 

   570     def _test_view(self, viewfunc, view, template='main-template', kwargs={}):

   571         """this method does the actual call to the view

   572 

   573         If no error occured while rendering the view, the HTML is analyzed

   574         and parsed.

   575 

   576         :returns: an instance of `cubicweb.devtools.htmlparser.PageInfo`

   577                   encapsulation the generated HTML

   578         """

   579         output = None

   580         try:

   581             output = viewfunc(**kwargs)

   582             return self._check_html(output, view, template)

   583         except (SystemExit, KeyboardInterrupt):

   584             raise

   585         except:

   586             # hijack exception: generative tests stop when the exception

   587             # is not an AssertionError

   588             klass, exc, tcbk = sys.exc_info()

   589             try:

   590                 msg = '[%s in %s] %s' % (klass, view.id, exc)

   591             except:

   592                 msg = '[%s in %s] undisplayable exception' % (klass, view.id)

   593             if output is not None:

   594                 position = getattr(exc, "position", (0,))[0]

   595                 if position:

   596                     # define filter

   597                     output = output.splitlines()

   598                     width = int(log(len(output), 10)) + 1

   599                     line_template = " %" + ("%i" % width) + "i: %s"

   600                     # XXX no need to iterate the whole file except to get

   601                     # the line number

   602                     output = '\n'.join(line_template % (idx + 1, line)

   603                                 for idx, line in enumerate(output)

   604                                 if line_context_filter(idx+1, position))

   605                     msg += '\nfor output:\n%s' % output

   606             raise AssertionError, msg, tcbk

   607 

   608 

   609     @nocoverage

   610     def _check_html(self, output, view, template='main-template'):

   611         """raises an exception if the HTML is invalid"""

   612         try:

   613             validatorclass = self.vid_validators[view.id]

   614         except KeyError:

   615             if template is None:

   616                 default_validator = htmlparser.HTMLValidator

   617             else:

   618                 default_validator = htmlparser.DTDValidator

   619             validatorclass = self.content_type_validators.get(view.content_type,

   620                                                               default_validator)

   621         if validatorclass is None:

   622             return None

   623         validator = validatorclass()

   624         return validator.parse_string(output.strip())

   625 

   626     # deprecated ###############################################################

   627 

   628     @deprecated('use self.vreg["etypes"].etype_class(etype)(self.request())')

   629     def etype_instance(self, etype, req=None):

   630         req = req or self.request()

   631         e = self.vreg['etypes'].etype_class(etype)(req)

   632         e.eid = None

   633         return e

   634 

   635     @nocoverage

   636     @deprecated('use req = self.request(); rset = req.execute()')

   637     def rset_and_req(self, rql, optional_args=None, args=None, eidkey=None):

   638         """executes <rql>, builds a resultset, and returns a

   639         couple (rset, req) where req is a FakeRequest

   640         """

   641         return (self.execute(rql, args, eidkey),

   642                 self.request(rql=rql, **optional_args or {}))

   643 

   644 

   645 # auto-populating test classes and utilities ###################################

   646 

   647 from cubicweb.devtools.fill import insert_entity_queries, make_relations_queries

   690 

   691     def to_test_etypes(self):

   692         return unprotected_entities(self.schema, strict=True)

   799     def setUp(self):

   800         AutoPopulateTest.setUp(self)

   801         # access to self.app for proper initialization of the authentication

   802         # machinery (else some views may fail)

   803         self.app

   804 

   834 

   836     # XXX broken

   857