1 """this module contains base classes for web tests

     2 

     3 :organization: Logilab

     4 :copyright: 2001-2008 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

     5 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr

     6 """

     7 __docformat__ = "restructuredtext en"

     8 

     9 import sys

    10 from math import log

    11 

    12 from logilab.common.debugger import Debugger

    13 from logilab.common.testlib import InnerTest

    14 from logilab.common.pytest import nocoverage

    15 

    16 from rql import parse

    17 

    18 from cubicweb.devtools import VIEW_VALIDATORS

    19 from cubicweb.devtools.apptest import EnvBasedTC

    20 from cubicweb.devtools._apptest import unprotected_entities, SYSTEM_RELATIONS

    21 from cubicweb.devtools.htmlparser import DTDValidator, SaxOnlyValidator, HTMLValidator

    22 from cubicweb.devtools.fill import insert_entity_queries, make_relations_queries

    23 

    24 from cubicweb.sobjects.notification import NotificationView

    25 

    26 from cubicweb.vregistry import NoSelectableObject

    27 from cubicweb.web.action import Action

    28 from cubicweb.web.views.basetemplates import TheMainTemplate

    29 

    30 

    31 ## TODO ###############

    32 # creation tests: make sure an entity was actually created

    33 # Existing Test Environment

    34 

    35 class CubicWebDebugger(Debugger):

    36 

    37     def do_view(self, arg):

    38         import webbrowser

    39         data = self._getval(arg)

    40         file('/tmp/toto.html', 'w').write(data)

    41         webbrowser.open('file:///tmp/toto.html')

    42 

    43 def how_many_dict(schema, cursor, how_many, skip):

    44     """compute how many entities by type we need to be able to satisfy relations

    45     cardinality

    46     """

    47     # compute how many entities by type we need to be able to satisfy relation constraint

    48     relmap = {}

    49     for rschema in schema.relations():

    50         if rschema.meta or rschema.is_final(): # skip meta relations

    51             continue

    52         for subj, obj in rschema.iter_rdefs():

    53             card = rschema.rproperty(subj, obj, 'cardinality')

    54             if card[0] in '1?' and len(rschema.subjects(obj)) == 1:

    55                 relmap.setdefault((rschema, subj), []).append(str(obj))

    56             if card[1] in '1?' and len(rschema.objects(subj)) == 1:

    57                 relmap.setdefault((rschema, obj), []).append(str(subj))

    58     unprotected = unprotected_entities(schema)

    59     for etype in skip:

    60         unprotected.add(etype)

    61     howmanydict = {}

    62     for etype in unprotected_entities(schema, strict=True):

    63         howmanydict[str(etype)] = cursor.execute('Any COUNT(X) WHERE X is %s' % etype)[0][0]

    64         if etype in unprotected:

    65             howmanydict[str(etype)] += how_many

    66     for (rschema, etype), targets in relmap.iteritems():

    67         # XXX should 1. check no cycle 2. propagate changes

    68         relfactor = sum(howmanydict[e] for e in targets)

    69         howmanydict[str(etype)] = max(relfactor, howmanydict[etype])

    70     return howmanydict

    71 

    72 

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

    74     """return true if line are in context

    75     if after is None: after = before"""

    76     if after is None:

    77         after = before

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

    79 

    80 ## base webtest class #########################################################

    81 class WebTest(EnvBasedTC):

    82     """base class for web tests"""

    83     __abstract__ = True

    84 

    85     pdbclass = CubicWebDebugger

    86     # this is a hook to be able to define a list of rql queries

    87     # that are application dependent and cannot be guessed automatically

    88     application_rql = []

    89 

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

    91     # validators availables are :

    92     #  DTDValidator : validates XML + declared DTD

    93     #  SaxOnlyValidator : guarantees XML is well formed

    94     #  None : do not try to validate anything

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

    96     validators = {

    97         # maps vid : validator name

    98         'hcal' : SaxOnlyValidator,

    99         'rss' : SaxOnlyValidator,

   100         'rssitem' : None,

   101         'xml' : SaxOnlyValidator,

   102         'xmlitem' : None,

   103         'xbel' : SaxOnlyValidator,

   104         'xbelitem' : None,

   105         'vcard' : None,

   106         'fulltext': None,

   107         'fullthreadtext': None,

   108         'fullthreadtext_descending': None,

   109         'text' : None,

   110         'treeitemview': None,

   111         'textincontext' : None,

   112         'textoutofcontext' : None,

   113         'combobox' : None,

   114         'csvexport' : None,

   115         'ecsvexport' : None,

   116         }

   117     valmap = {None: None, 'dtd': DTDValidator, 'xml': SaxOnlyValidator}

   118     no_auto_populate = ()

   119     ignored_relations = ()

   120     

   121     def __init__(self, *args, **kwargs):

   122         EnvBasedTC.__init__(self, *args, **kwargs)

   123         for view, valkey in VIEW_VALIDATORS.iteritems():

   124             self.validators[view] = self.valmap[valkey]

   125         

   126     def custom_populate(self, how_many, cursor):

   127         pass

   128         

   129     def post_populate(self, cursor):

   130         pass

   131     

   132     @nocoverage

   133     def auto_populate(self, how_many):

   134         """this method populates the database with `how_many` entities

   135         of each possible type. It also inserts random relations between them

   136         """

   137         cu = self.cursor()

   138         self.custom_populate(how_many, cu)

   139         vreg = self.vreg

   140         howmanydict = how_many_dict(self.schema, cu, how_many, self.no_auto_populate)

   141         for etype in unprotected_entities(self.schema):

   142             if etype in self.no_auto_populate:

   143                 continue

   144             nb = howmanydict.get(etype, how_many)

   145             for rql, args in insert_entity_queries(etype, self.schema, vreg, nb):

   146                 cu.execute(rql, args)

   147         edict = {}

   148         for etype in unprotected_entities(self.schema, strict=True):

   149             rset = cu.execute('%s X' % etype)

   150             edict[str(etype)] = set(row[0] for row in rset.rows)

   151         existingrels = {}

   152         ignored_relations = SYSTEM_RELATIONS + self.ignored_relations

   153         for rschema in self.schema.relations():

   154             if rschema.is_final() or rschema in ignored_relations:

   155                 continue

   156             rset = cu.execute('DISTINCT Any X,Y WHERE X %s Y' % rschema)

   157             existingrels.setdefault(rschema.type, set()).update((x,y) for x, y in rset)

   158         q = make_relations_queries(self.schema, edict, cu, ignored_relations,

   159                                    existingrels=existingrels)

   160         for rql, args in q:

   161             cu.execute(rql, args)

   162         self.post_populate(cu)

   163         self.commit()

   164 

   165     @nocoverage

   166     def _check_html(self, output, vid, template='main'):

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

   168         if template is None:

   169             default_validator = HTMLValidator

   170         else:

   171             default_validator = DTDValidator

   172         validatorclass = self.validators.get(vid, default_validator)

   173         if validatorclass is None:

   174             return None

   175         validator = validatorclass()

   176         output = output.strip()

   177         return validator.parse_string(output)

   178 

   179 

   180     def view(self, vid, rset, req=None, template='main', htmlcheck=True, **kwargs):

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

   182 

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

   184         and parsed.

   185 

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

   187                   encapsulation the generated HTML

   188         """

   189         req = req or rset.req

   190         # print "testing ", vid,

   191         # if rset:

   192         #     print rset, len(rset), id(rset)

   193         # else:

   194         #     print 

   195         req.form['vid'] = vid

   196         view = self.vreg.select_view(vid, req, rset, **kwargs)

   197         if view.content_type not in ('application/xml', 'application/xhtml+xml', 'text/html'):

   198             htmlcheck = False

   199         # set explicit test description

   200         if rset is not None:

   201             self.set_description("testing %s, mod=%s (%s)" % (vid, view.__module__, rset.printable_rql()))

   202         else:

   203             self.set_description("testing %s, mod=%s (no rset)" % (vid, view.__module__))

   204         viewfunc = lambda **k: self.vreg.main_template(req, template, **kwargs)

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

   206             viewfunc = view.dispatch

   207         elif template == 'main':

   208             _select_view_and_rset = TheMainTemplate._select_view_and_rset

   209             # patch TheMainTemplate.process_rql to avoid recomputing resultset

   210             TheMainTemplate._select_view_and_rset = lambda *a, **k: (view, rset)

   211         try:

   212             return self._test_view(viewfunc, vid, htmlcheck, template, **kwargs)

   213         finally:

   214             if template == 'main':

   215                 TheMainTemplate._select_view_and_rset = _select_view_and_rset

   216 

   217 

   218     def _test_view(self, viewfunc, vid, htmlcheck=True, template='main', **kwargs):

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

   220 

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

   222         and parsed.

   223 

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

   225                   encapsulation the generated HTML

   226         """

   227         output = None

   228         try:

   229             output = viewfunc(**kwargs)

   230             if htmlcheck:

   231                 return self._check_html(output, vid, template)

   232             else:

   233                 return output

   234         except (SystemExit, KeyboardInterrupt):

   235             raise

   236         except:

   237             # hijack exception: generative tests stop when the exception

   238             # is not an AssertionError

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

   240             try:

   241                 msg = '[%s in %s] %s' % (klass, vid, exc)

   242             except:

   243                 msg = '[%s in %s] undisplayable exception' % (klass, vid)

   244             if output is not None:

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

   246                 if position:

   247                     # define filter

   248                     

   249                     

   250                     output = output.splitlines()

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

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

   253 

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

   255                     # the line number

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

   257                                 for idx, line in enumerate(output)

   258                                 if line_context_filter(idx+1, position))

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

   260             raise AssertionError, msg, tcbk

   261 

   262         

   263     def iter_automatic_rsets(self):

   264         """generates basic resultsets for each entity type"""

   265         etypes = unprotected_entities(self.schema, strict=True)

   266         for etype in etypes:

   267             yield self.execute('Any X WHERE X is %s' % etype)

   268 

   269         etype1 = etypes.pop()

   270         etype2 = etypes.pop()

   271         # test a mixed query (DISTINCT/GROUP to avoid getting duplicate

   272         # X which make muledit view failing for instance (html validation fails

   273         # because of some duplicate "id" attributes)

   274         yield self.execute('DISTINCT Any X, MAX(Y) GROUPBY X WHERE X is %s, Y is %s' % (etype1, etype2))

   275         # test some application-specific queries if defined

   276         for rql in self.application_rql:

   277             yield self.execute(rql)

   278 

   279                 

   280     def list_views_for(self, rset):

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

   282         req = rset.req

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

   284         skipped = ('restriction', 'cell')

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

   286         for vid, views in self.vreg.registry('views').items():

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

   288                 continue

   289             try:

   290                 view = self.vreg.select(views, req, rset)

   291                 if view.id in skipped:

   292                     continue

   293                 if view.category == 'startupview':

   294                     continue

   295                 if rset.rowcount > 1 and view.id in only_once_vids:

   296                     continue

   297                 if not isinstance(view, NotificationView):

   298                     yield view

   299             except NoSelectableObject:

   300                 continue

   301 

   302     def list_actions_for(self, rset):

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

   304         req = rset.req

   305         for action in self.vreg.possible_objects('actions', req, rset):

   306             yield action

   307 

   308         

   309     def list_boxes_for(self, rset):

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

   311         req = rset.req

   312         for box in self.vreg.possible_objects('boxes', req, rset):

   313             yield box

   314             

   315         

   316     def list_startup_views(self):

   317         """returns the list of startup views"""

   318         req = self.request()

   319         for view in self.vreg.possible_views(req, None):

   320             if view.category != 'startupview':

   321                 continue

   322             yield view.id

   323 

   324     def _test_everything_for(self, rset):

   325         """this method tries to find everything that can be tested

   326         for `rset` and yields a callable test (as needed in generative tests)

   327         """

   328         rqlst = parse(rset.rql)

   329         propdefs = self.vreg['propertydefs']

   330         # make all components visible

   331         for k, v in propdefs.items():

   332             if k.endswith('visible') and not v['default']:

   333                 propdefs[k]['default'] = True

   334         for view in self.list_views_for(rset):

   335             backup_rset = rset._prepare_copy(rset.rows, rset.description)

   336             yield InnerTest(self._testname(rset, view.id, 'view'),

   337                             self.view, view.id, rset,

   338                             rset.req.reset_headers(), 'main', not view.binary)

   339             # We have to do this because some views modify the

   340             # resultset's syntax tree

   341             rset = backup_rset

   342         for action in self.list_actions_for(rset):

   343             # XXX this seems a bit dummy

   344             #yield InnerTest(self._testname(rset, action.id, 'action'),

   345             #                self.failUnless,

   346             #                isinstance(action, Action))

   347             yield InnerTest(self._testname(rset, action.id, 'action'), action.url)

   348         for box in self.list_boxes_for(rset):

   349             yield InnerTest(self._testname(rset, box.id, 'box'), box.dispatch)

   350 

   351 

   352 

   353     @staticmethod

   354     def _testname(rset, objid, objtype):

   355         return '%s_%s_%s' % ('_'.join(rset.column_types(0)), objid, objtype)

   356             

   357 

   358 class AutomaticWebTest(WebTest):

   359     """import this if you wan automatic tests to be ran"""

   360     ## one each

   361     def test_one_each_config(self):

   362         self.auto_populate(1)

   363         for rset in self.iter_automatic_rsets():

   364             for testargs in self._test_everything_for(rset):

   365                 yield testargs

   366 

   367     ## ten each

   368     def test_ten_each_config(self):

   369         self.auto_populate(10)

   370         for rset in self.iter_automatic_rsets():

   371             for testargs in self._test_everything_for(rset):

   372                 yield testargs

   373                 

   374     ## startup views

   375     def test_startup_views(self):

   376         for vid in self.list_startup_views():

   377             req = self.request()

   378             yield self.view, vid, None, req

   379 

   380 

   381 class RealDBTest(WebTest):

   382 

   383     def iter_individual_rsets(self, etypes=None, limit=None):

   384         etypes = etypes or unprotected_entities(self.schema, strict=True)

   385         for etype in etypes:

   386             rset = self.execute('Any X WHERE X is %s' % etype)

   387             for row in xrange(len(rset)):

   388                 if limit and row > limit:

   389                     break

   390                 rset2 = rset.limit(limit=1, offset=row)

   391                 yield rset2

   392 

   393