1 """Test tools for cubicweb

     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 os

    10 import logging

    11 from os.path import (abspath, join, exists, basename, dirname, normpath, split,

    12                      isfile, isabs)

    13 

    14 from mx.DateTime import strptime, DateTimeDelta

    15 

    16 from cubicweb import CW_SOFTWARE_ROOT, ConfigurationError

    17 from cubicweb.toolsutils import read_config

    18 from cubicweb.cwconfig import CubicWebConfiguration, merge_options

    19 from cubicweb.server.serverconfig import ServerConfiguration

    20 from cubicweb.etwist.twconfig import TwistedConfiguration

    21 

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

    23 # validators availables are :

    24 #  'dtd' : validates XML + declared DTD

    25 #  'xml' : guarantees XML is well formed

    26 #  None : do not try to validate anything

    27 VIEW_VALIDATORS = {}

    28 BASE_URL = 'http://testing.fr/cubicweb/'

    29 DEFAULT_SOURCES = {'system': {'adapter' : 'native',

    30                               'db-encoding' : 'UTF-8', #'ISO-8859-1',

    31                               'db-user' : u'admin',

    32                               'db-password' : 'gingkow',

    33                               'db-name' : 'tmpdb',

    34                               'db-driver' : 'sqlite',

    35                               'db-host' : None,

    36                               },

    37                    'admin' : {'login': u'admin',

    38                               'password': u'gingkow',

    39                               },

    40                    }

    41 

    42 class TestServerConfiguration(ServerConfiguration):

    43     mode = 'test'

    44     set_language = False

    45     read_application_schema = False

    46     bootstrap_schema = False

    47     init_repository = True

    48     options = merge_options(ServerConfiguration.options + (

    49         ('anonymous-user',

    50          {'type' : 'string',

    51           'default': None,

    52           'help': 'login of the CubicWeb user account to use for anonymous user (if you want to allow anonymous)',

    53           'group': 'main', 'inputlevel': 1,

    54           }),

    55         ('anonymous-password',

    56          {'type' : 'string',

    57           'default': None,

    58           'help': 'password of the CubicWeb user account matching login',

    59           'group': 'main', 'inputlevel': 1,

    60           }),

    61         ))

    62                             

    63     if not os.environ.get('APYCOT_ROOT'):

    64         REGISTRY_DIR = normpath(join(CW_SOFTWARE_ROOT, '../cubes'))

    65     

    66     def __init__(self, appid, log_threshold=logging.CRITICAL+10):

    67         ServerConfiguration.__init__(self, appid)

    68         self.global_set_option('log-file', None)

    69         self.init_log(log_threshold, force=True)

    70         # need this, usually triggered by cubicweb-ctl

    71         self.load_cwctl_plugins()

    72 

    73     anonymous_user = TwistedConfiguration.anonymous_user.im_func

    74         

    75     @property

    76     def apphome(self):

    77         if exists(self.appid):

    78             return abspath(self.appid)

    79         # application cube test

    80         return abspath('..')

    81     appdatahome = apphome

    82     

    83     def main_config_file(self):

    84         """return application's control configuration file"""

    85         return join(self.apphome, '%s.conf' % self.name)

    86 

    87     def instance_md5_version(self):

    88         return ''

    89 

    90     def bootstrap_cubes(self):

    91         try:

    92             super(TestServerConfiguration, self).bootstrap_cubes()

    93         except IOError:

    94             # no cubes

    95             self.init_cubes( () )

    96 

    97     sourcefile = None

    98     def sources_file(self):

    99         """define in subclasses self.sourcefile if necessary"""

   100         if self.sourcefile:

   101             print 'Reading sources from', self.sourcefile

   102             sourcefile = self.sourcefile

   103             if not isabs(sourcefile):

   104                 sourcefile = join(self.apphome, sourcefile)

   105         else:

   106             sourcefile = super(TestServerConfiguration, self).sources_file()

   107         return sourcefile

   108 

   109     def sources(self):

   110         """By default, we run tests with the sqlite DB backend.  One may use its

   111         own configuration by just creating a 'sources' file in the test

   112         directory from wich tests are launched or by specifying an alternative

   113         sources file using self.sourcefile.

   114         """

   115         sources = super(TestServerConfiguration, self).sources()

   116         if not sources:

   117             sources = DEFAULT_SOURCES

   118         return sources

   119     

   120     def load_defaults(self):

   121         super(TestServerConfiguration, self).load_defaults()

   122         # note: don't call global set option here, OptionManager may not yet be initialized

   123         # add anonymous user

   124         self.set_option('anonymous-user', 'anon')

   125         self.set_option('anonymous-password', 'anon')

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

   127         #self.set_option('query-log-file', '/tmp/test_rql_log.' + `os.getpid()`)

   128         self.set_option('sender-name', 'cubicweb-test')

   129         self.set_option('sender-addr', 'cubicweb-test@logilab.fr')

   130         try:

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

   132         except OSError:

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

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

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

   136         self.set_option('sender-addr', send_to)

   137         self.set_option('default-dest-addrs', send_to)

   138         self.set_option('base-url', BASE_URL)

   139 

   140 

   141 class BaseApptestConfiguration(TestServerConfiguration, TwistedConfiguration):

   142     repo_method = 'inmemory'

   143     options = merge_options(TestServerConfiguration.options + TwistedConfiguration.options)

   144     cubicweb_vobject_path = TestServerConfiguration.cubicweb_vobject_path | TwistedConfiguration.cubicweb_vobject_path

   145     cube_vobject_path = TestServerConfiguration.cube_vobject_path | TwistedConfiguration.cube_vobject_path

   146 

   147     def available_languages(self, *args):

   148         return ('en', 'fr', 'de')

   149     

   150     def ext_resources_file(self):

   151         """return application's external resources file"""

   152         return join(self.apphome, 'data', 'external_resources')

   153     

   154     def pyro_enabled(self):

   155         # but export PYRO_MULTITHREAD=0 or you get problems with sqlite and threads

   156         return True

   157 

   158 

   159 class ApptestConfiguration(BaseApptestConfiguration):

   160     

   161     def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None):

   162         BaseApptestConfiguration.__init__(self, appid, log_threshold=log_threshold)

   163         self.init_repository = sourcefile is None

   164         self.sourcefile = sourcefile

   165         import re

   166         self.global_set_option('embed-allowed', re.compile('.*'))

   167         

   168 

   169 class RealDatabaseConfiguration(ApptestConfiguration):

   170     init_repository = False

   171     sourcesdef =  {'system': {'adapter' : 'native',

   172                               'db-encoding' : 'UTF-8', #'ISO-8859-1',

   173                               'db-user' : u'admin',

   174                               'db-password' : 'gingkow',

   175                               'db-name' : 'seotest',

   176                               'db-driver' : 'postgres',

   177                               'db-host' : None,

   178                               },

   179                    'admin' : {'login': u'admin',

   180                               'password': u'gingkow',

   181                               },

   182                    }

   183     

   184     def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None):

   185         ApptestConfiguration.__init__(self, appid)

   186         self.init_repository = False

   187 

   188 

   189     def sources(self):

   190         """

   191         By default, we run tests with the sqlite DB backend.

   192         One may use its own configuration by just creating a

   193         'sources' file in the test directory from wich tests are

   194         launched. 

   195         """

   196         self._sources = self.sourcesdef

   197         return self._sources

   198 

   199 

   200 def buildconfig(dbuser, dbpassword, dbname, adminuser, adminpassword, dbhost=None):

   201     """convenience function that builds a real-db configuration class"""

   202     sourcesdef =  {'system': {'adapter' : 'native',

   203                               'db-encoding' : 'UTF-8', #'ISO-8859-1',

   204                               'db-user' : dbuser,

   205                               'db-password' : dbpassword,

   206                               'db-name' : dbname,

   207                               'db-driver' : 'postgres',

   208                               'db-host' : dbhost,

   209                               },

   210                    'admin' : {'login': adminuser,

   211                               'password': adminpassword,

   212                               },

   213                    }

   214     return type('MyRealDBConfig', (RealDatabaseConfiguration,),

   215                 {'sourcesdef': sourcesdef})

   216 

   217 def loadconfig(filename):

   218     """convenience function that builds a real-db configuration class

   219     from a file

   220     """

   221     return type('MyRealDBConfig', (RealDatabaseConfiguration,),

   222                 {'sourcesdef': read_config(filename)})

   223     

   224 

   225 class LivetestConfiguration(BaseApptestConfiguration):

   226     init_repository = False

   227     

   228     def __init__(self, cube=None, sourcefile=None, pyro_name=None,

   229                  log_threshold=logging.CRITICAL):

   230         TestServerConfiguration.__init__(self, cube, log_threshold=log_threshold)

   231         self.appid = pyro_name or cube

   232         # don't change this, else some symlink problems may arise in some

   233         # environment (e.g. mine (syt) ;o)

   234         # XXX I'm afraid this test will prevent to run test from a production

   235         # environment

   236         self._sources = None

   237         # application cube test

   238         if cube is not None:

   239             self.apphome = self.cube_dir(cube)

   240         elif 'web' in os.getcwd().split(os.sep):

   241             # web test

   242             self.apphome = join(normpath(join(dirname(__file__), '..')), 'web')

   243         else:

   244             # application cube test

   245             self.apphome = abspath('..')

   246         self.sourcefile = sourcefile

   247         self.global_set_option('realm', '')

   248         self.use_pyro = pyro_name is not None

   249 

   250     def pyro_enabled(self):

   251         if self.use_pyro:

   252             return True

   253         else:

   254             return False

   255 

   256 CubicWebConfiguration.cls_adjust_sys_path()

   257                                                     

   258 def install_sqlite_path(querier):

   259     """This patch hotfixes the following sqlite bug :

   260      - http://www.sqlite.org/cvstrac/tktview?tn=1327,33

   261     (some dates are returned as strings rather thant date objects)

   262     """

   263     def wrap_execute(base_execute):

   264         def new_execute(*args, **kwargs):

   265             rset = base_execute(*args, **kwargs)

   266             if rset.description:

   267                 found_date = False

   268                 for row, rowdesc in zip(rset, rset.description):

   269                     for cellindex, (value, vtype) in enumerate(zip(row, rowdesc)):

   270                         if vtype in ('Date', 'Datetime') and type(value) is unicode:

   271                             found_date = True

   272                             try:

   273                                 row[cellindex] = strptime(value, '%Y-%m-%d %H:%M:%S')

   274                             except:

   275                                 row[cellindex] = strptime(value, '%Y-%m-%d')

   276                         if vtype == 'Time' and type(value) is unicode:

   277                             found_date = True

   278                             try:

   279                                 row[cellindex] = strptime(value, '%H:%M:%S')

   280                             except:

   281                                 # DateTime used as Time?

   282                                 row[cellindex] = strptime(value, '%Y-%m-%d %H:%M:%S')

   283                         if vtype == 'Interval' and type(value) is int:

   284                             found_date = True

   285                             row[cellindex] = DateTimeDelta(0, 0, 0, value)

   286                     if not found_date:

   287                         break

   288             return rset

   289         return new_execute

   290     querier.__class__.execute = wrap_execute(querier.__class__.execute)

   291 

   292 

   293 def init_test_database(driver='sqlite', configdir='data', config=None,

   294                        vreg=None):

   295     """init a test database for a specific driver"""

   296     from cubicweb.dbapi import in_memory_cnx

   297     if vreg and not config:

   298         config = vreg.config

   299     config = config or TestServerConfiguration(configdir)

   300     source = config.sources()

   301     if driver == 'sqlite':

   302         init_test_database_sqlite(config, source)

   303     elif driver == 'postgres':

   304         init_test_database_postgres(config, source)

   305     else:

   306         raise ValueError('no initialization function for driver %r' % driver)

   307     config._cubes = None # avoid assertion error

   308     repo, cnx = in_memory_cnx(vreg or config, unicode(source['admin']['login']),

   309                               source['admin']['password'] or 'xxx')

   310     if driver == 'sqlite':

   311         install_sqlite_path(repo.querier)

   312     return repo, cnx

   313 

   314 def init_test_database_postgres(config, source, vreg=None):

   315     """initialize a fresh sqlite databse used for testing purpose"""

   316     if config.init_repository:

   317         from cubicweb.server import init_repository

   318         init_repository(config, interactive=False, drop=True, vreg=vreg)

   319 

   320 def cleanup_sqlite(dbfile, removecube=False):

   321     try:

   322         os.remove(dbfile)

   323         os.remove('%s-journal' % dbfile)

   324     except OSError:

   325         pass

   326     if removecube:

   327         try:

   328             os.remove('%s-cube' % dbfile)

   329         except OSError:

   330             pass

   331     

   332 def init_test_database_sqlite(config, source, vreg=None):

   333     """initialize a fresh sqlite databse used for testing purpose"""

   334     import shutil

   335     # remove database file if it exists (actually I know driver == 'sqlite' :)

   336     dbfile = source['system']['db-name']

   337     cleanup_sqlite(dbfile)

   338     cube = '%s-cube' % dbfile

   339     if exists(cube):

   340         shutil.copy(cube, dbfile)

   341     else:

   342         # initialize the database

   343         from cubicweb.server import init_repository

   344         init_repository(config, interactive=False, vreg=vreg)

   345         shutil.copy(dbfile, cube)