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     # def scommit(self):

   349     #     self.repo.commit(self.cnxid)

   350     #     self.session.set_pool()

   351 

   352     # def srollback(self):

   353     #     self.repo.rollback(self.cnxid)

   354     #     self.session.set_pool()

   355 

   356     # def sclose(self):

   357     #     self.repo.close(self.cnxid)

   358 

   359     # other utilities #########################################################

   360 

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

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

   363 

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

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

   366         # dict for replacement in RQL Request

   367         args = {}

   368         if kwargs:

   369             rql.append(':')

   370             # dict to define new entities variables

   371             entities = {}

   372             # assignement part of the request

   373             sub_rql = []

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

   375                 # entities

   376                 if hasattr(value, 'eid'):

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

   378                     entities[new_value] = value.eid

   379                     args[new_value] = value.eid

   380 

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

   382                 # final attributes

   383                 else:

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

   385                     args[key] = value

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

   387             if entities:

   388                 rql.append('WHERE')

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

   390                 sub_rql = []

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

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

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

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

   395 

   396     # vregistry inspection utilities ###########################################

   397 

   398     def pviews(self, req, rset):

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

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

   401 

   402     def pactions(self, req, rset,

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

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

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

   406                 if a.category not in skipcategories]

   407 

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

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

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

   411                 if a.category in categories]

   412 

   413     def pactionsdict(self, req, rset,

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

   415         res = {}

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

   417             if a.category not in skipcategories:

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

   419         return res

   420 

   421     def list_views_for(self, rset):

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

   423         req = rset.req

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

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

   426         viewsvreg = self.vreg['views']

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

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

   429                 continue

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

   431                 continue

   432             views = [view for view in views

   433                      if view.category != 'startupview'

   434                      and not issubclass(view, notification.NotificationView)]

   435             if views:

   436                 try:

   437                     view = viewsvreg.select_best(views, req, rset=rset)

   438                     if view.linkable():

   439                         yield view

   440                     else:

   441                         not_selected(self.vreg, view)

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

   443                     # not be tested directly

   444                 except NoSelectableObject:

   445                     continue

   446 

   447     def list_actions_for(self, rset):

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

   449         req = rset.req

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

   451             yield action

   452 

   453     def list_boxes_for(self, rset):

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

   455         req = rset.req

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

   457             yield box

   458 

   459     def list_startup_views(self):

   460         """returns the list of startup views"""

   461         req = self.request()

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

   463             if view.category == 'startupview':

   464                 yield view.id

   465             else:

   466                 not_selected(self.vreg, view)

   467 

   468     # web ui testing utilities #################################################

   469 

   470     @property

   471     @cached

   472     def app(self):

   473         """return a cubicweb publisher"""

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

   475 

   476     requestcls = fake.FakeRequest

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

   478         """return a web ui request"""

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

   480         req.set_connection(self.cnx)

   481         return req

   482 

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

   484         """remote json call simulation"""

   485         dump = simplejson.dumps

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

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

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

   489         return ctrl.publish(), req

   490 

   491     def publish(self, req):

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

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

   494         try:

   495             result = ctrl.publish()

   496             req.cnx.commit()

   497         except web.Redirect:

   498             req.cnx.commit()

   499             raise

   500         return result

   501 

   502     def expect_redirect_publish(self, req):

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

   504         Redirect exception."""

   505         try:

   506             self.publish(req)

   507         except web.Redirect, ex:

   508             try:

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

   510             except:

   511                 path, params = ex.location, ""

   512             req._url = path

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

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

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

   516         else:

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

   518 

   519     # content validation #######################################################

   520 

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

   522     # validators availables are :

   523     #  DTDValidator : validates XML + declared DTD

   524     #  SaxOnlyValidator : guarantees XML is well formed

   525     #  None : do not try to validate anything

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

   527     content_type_validators = {

   528         # maps MIME type : validator name

   529         #

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

   531         # snippets

   532         #'text/html': DTDValidator,

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

   534         'application/xml': htmlparser.SaxOnlyValidator,

   535         'text/xml': htmlparser.SaxOnlyValidator,

   536         'text/plain': None,

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

   538         'text/x-vcard': None,

   539         'text/calendar': None,

   540         'application/json': None,

   541         'image/png': None,

   542         }

   543     # maps vid : validator name (override content_type_validators)

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

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

   546 

   547 

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

   549              **kwargs):

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

   551 

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

   553         and parsed.

   554 

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

   556                   encapsulation the generated HTML

   557         """

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

   559         req.form['vid'] = vid

   560         kwargs['rset'] = rset

   561         viewsreg = self.vreg['views']

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

   563         # set explicit test description

   564         if rset is not None:

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

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

   567         else:

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

   569                 vid, view.__module__))

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

   571             viewfunc = view.render

   572         else:

   573             kwargs['view'] = view

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

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

   576                                                           **kwargs)

   577         kwargs.pop('rset')

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

   579 

   580 

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

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

   583 

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

   585         and parsed.

   586 

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

   588                   encapsulation the generated HTML

   589         """

   590         output = None

   591         try:

   592             output = viewfunc(**kwargs)

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

   594         except (SystemExit, KeyboardInterrupt):

   595             raise

   596         except:

   597             # hijack exception: generative tests stop when the exception

   598             # is not an AssertionError

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

   600             try:

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

   602             except:

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

   604             if output is not None:

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

   606                 if position:

   607                     # define filter

   608                     output = output.splitlines()

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

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

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

   612                     # the line number

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

   614                                 for idx, line in enumerate(output)

   615                                 if line_context_filter(idx+1, position))

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

   617             raise AssertionError, msg, tcbk

   618 

   619 

   620     @nocoverage

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

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

   623         try:

   624             validatorclass = self.vid_validators[view.id]

   625         except KeyError:

   626             if template is None:

   627                 default_validator = htmlparser.HTMLValidator

   628             else:

   629                 default_validator = htmlparser.DTDValidator

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

   631                                                               default_validator)

   632         if validatorclass is None:

   633             return None

   634         validator = validatorclass()

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

   636 

   637     # deprecated ###############################################################

   638 

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

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

   641         req = req or self.request()

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

   643         e.eid = None

   644         return e

   645 

   646     @nocoverage

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

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

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

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

   651         """

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

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

   654 

   655 

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

   657 

   658 from cubicweb.devtools.fill import insert_entity_queries, make_relations_queries

   701 

   702     def to_test_etypes(self):

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

   810     def setUp(self):

   811         AutoPopulateTest.setUp(self)

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

   813         # machinery (else some views may fail)

   814         self.app

   815 

   845 

   847     # XXX broken

   868