1 """Test tools for cubicweb |
1 """Test tools for cubicweb |
2 |
2 |
3 :organization: Logilab |
3 :organization: Logilab |
4 :copyright: 2001-2008 LOGILAB S.A. (Paris, FRANCE), all rights reserved. |
4 :copyright: 2001-2009 LOGILAB S.A. (Paris, FRANCE), all rights reserved. |
5 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr |
5 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr |
6 """ |
6 """ |
7 __docformat__ = "restructuredtext en" |
7 __docformat__ = "restructuredtext en" |
8 |
8 |
9 import os |
9 import os |
10 import logging |
10 import logging |
|
11 from datetime import timedelta |
11 from os.path import (abspath, join, exists, basename, dirname, normpath, split, |
12 from os.path import (abspath, join, exists, basename, dirname, normpath, split, |
12 isfile, isabs) |
13 isfile, isabs) |
13 |
14 |
14 from mx.DateTime import strptime, DateTimeDelta |
|
15 |
|
16 from cubicweb import CW_SOFTWARE_ROOT, ConfigurationError |
15 from cubicweb import CW_SOFTWARE_ROOT, ConfigurationError |
|
16 from cubicweb.utils import strptime |
17 from cubicweb.toolsutils import read_config |
17 from cubicweb.toolsutils import read_config |
18 from cubicweb.cwconfig import CubicWebConfiguration, merge_options |
18 from cubicweb.cwconfig import CubicWebConfiguration, merge_options |
19 from cubicweb.server.serverconfig import ServerConfiguration |
19 from cubicweb.server.serverconfig import ServerConfiguration |
20 from cubicweb.etwist.twconfig import TwistedConfiguration |
20 from cubicweb.etwist.twconfig import TwistedConfiguration |
21 |
21 |
57 'default': None, |
57 'default': None, |
58 'help': 'password of the CubicWeb user account matching login', |
58 'help': 'password of the CubicWeb user account matching login', |
59 'group': 'main', 'inputlevel': 1, |
59 'group': 'main', 'inputlevel': 1, |
60 }), |
60 }), |
61 )) |
61 )) |
62 |
62 |
63 if not os.environ.get('APYCOT_ROOT'): |
63 if not os.environ.get('APYCOT_ROOT'): |
64 REGISTRY_DIR = normpath(join(CW_SOFTWARE_ROOT, '../cubes')) |
64 REGISTRY_DIR = normpath(join(CW_SOFTWARE_ROOT, '../cubes')) |
65 |
65 |
66 def __init__(self, appid, log_threshold=logging.CRITICAL+10): |
66 def __init__(self, appid, log_threshold=logging.CRITICAL+10): |
67 ServerConfiguration.__init__(self, appid) |
67 ServerConfiguration.__init__(self, appid) |
68 self.global_set_option('log-file', None) |
68 self.global_set_option('log-file', None) |
69 self.init_log(log_threshold, force=True) |
69 self.init_log(log_threshold, force=True) |
70 # need this, usually triggered by cubicweb-ctl |
70 # need this, usually triggered by cubicweb-ctl |
71 self.load_cwctl_plugins() |
71 self.load_cwctl_plugins() |
72 |
72 |
73 anonymous_user = TwistedConfiguration.anonymous_user.im_func |
73 anonymous_user = TwistedConfiguration.anonymous_user.im_func |
74 |
74 |
75 @property |
75 @property |
76 def apphome(self): |
76 def apphome(self): |
77 if exists(self.appid): |
77 if exists(self.appid): |
78 return abspath(self.appid) |
78 return abspath(self.appid) |
79 # application cube test |
79 # application cube test |
80 return abspath('..') |
80 return abspath('..') |
81 appdatahome = apphome |
81 appdatahome = apphome |
82 |
82 |
83 def main_config_file(self): |
83 def main_config_file(self): |
84 """return application's control configuration file""" |
84 """return application's control configuration file""" |
85 return join(self.apphome, '%s.conf' % self.name) |
85 return join(self.apphome, '%s.conf' % self.name) |
86 |
86 |
87 def instance_md5_version(self): |
87 def instance_md5_version(self): |
114 """ |
114 """ |
115 sources = super(TestServerConfiguration, self).sources() |
115 sources = super(TestServerConfiguration, self).sources() |
116 if not sources: |
116 if not sources: |
117 sources = DEFAULT_SOURCES |
117 sources = DEFAULT_SOURCES |
118 return sources |
118 return sources |
119 |
119 |
120 def load_defaults(self): |
120 def load_defaults(self): |
121 super(TestServerConfiguration, self).load_defaults() |
121 super(TestServerConfiguration, self).load_defaults() |
122 # note: don't call global set option here, OptionManager may not yet be initialized |
122 # note: don't call global set option here, OptionManager may not yet be initialized |
123 # add anonymous user |
123 # add anonymous user |
124 self.set_option('anonymous-user', 'anon') |
124 self.set_option('anonymous-user', 'anon') |
144 cubicweb_vobject_path = TestServerConfiguration.cubicweb_vobject_path | TwistedConfiguration.cubicweb_vobject_path |
144 cubicweb_vobject_path = TestServerConfiguration.cubicweb_vobject_path | TwistedConfiguration.cubicweb_vobject_path |
145 cube_vobject_path = TestServerConfiguration.cube_vobject_path | TwistedConfiguration.cube_vobject_path |
145 cube_vobject_path = TestServerConfiguration.cube_vobject_path | TwistedConfiguration.cube_vobject_path |
146 |
146 |
147 def available_languages(self, *args): |
147 def available_languages(self, *args): |
148 return ('en', 'fr', 'de') |
148 return ('en', 'fr', 'de') |
149 |
149 |
150 def ext_resources_file(self): |
150 def ext_resources_file(self): |
151 """return application's external resources file""" |
151 """return application's external resources file""" |
152 return join(self.apphome, 'data', 'external_resources') |
152 return join(self.apphome, 'data', 'external_resources') |
153 |
153 |
154 def pyro_enabled(self): |
154 def pyro_enabled(self): |
155 # but export PYRO_MULTITHREAD=0 or you get problems with sqlite and threads |
155 # but export PYRO_MULTITHREAD=0 or you get problems with sqlite and threads |
156 return True |
156 return True |
157 |
157 |
158 |
158 |
159 class ApptestConfiguration(BaseApptestConfiguration): |
159 class ApptestConfiguration(BaseApptestConfiguration): |
160 |
160 |
161 def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None): |
161 def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None): |
162 BaseApptestConfiguration.__init__(self, appid, log_threshold=log_threshold) |
162 BaseApptestConfiguration.__init__(self, appid, log_threshold=log_threshold) |
163 self.init_repository = sourcefile is None |
163 self.init_repository = sourcefile is None |
164 self.sourcefile = sourcefile |
164 self.sourcefile = sourcefile |
165 import re |
165 import re |
166 self.global_set_option('embed-allowed', re.compile('.*')) |
166 self.global_set_option('embed-allowed', re.compile('.*')) |
167 |
167 |
168 |
168 |
169 class RealDatabaseConfiguration(ApptestConfiguration): |
169 class RealDatabaseConfiguration(ApptestConfiguration): |
170 init_repository = False |
170 init_repository = False |
171 sourcesdef = {'system': {'adapter' : 'native', |
171 sourcesdef = {'system': {'adapter' : 'native', |
172 'db-encoding' : 'UTF-8', #'ISO-8859-1', |
172 'db-encoding' : 'UTF-8', #'ISO-8859-1', |
178 }, |
178 }, |
179 'admin' : {'login': u'admin', |
179 'admin' : {'login': u'admin', |
180 'password': u'gingkow', |
180 'password': u'gingkow', |
181 }, |
181 }, |
182 } |
182 } |
183 |
183 |
184 def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None): |
184 def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None): |
185 ApptestConfiguration.__init__(self, appid) |
185 ApptestConfiguration.__init__(self, appid) |
186 self.init_repository = False |
186 self.init_repository = False |
187 |
187 |
188 |
188 |
189 def sources(self): |
189 def sources(self): |
190 """ |
190 """ |
191 By default, we run tests with the sqlite DB backend. |
191 By default, we run tests with the sqlite DB backend. |
192 One may use its own configuration by just creating a |
192 One may use its own configuration by just creating a |
193 'sources' file in the test directory from wich tests are |
193 'sources' file in the test directory from wich tests are |
194 launched. |
194 launched. |
195 """ |
195 """ |
196 self._sources = self.sourcesdef |
196 self._sources = self.sourcesdef |
197 return self._sources |
197 return self._sources |
198 |
198 |
199 |
199 |
218 """convenience function that builds a real-db configuration class |
218 """convenience function that builds a real-db configuration class |
219 from a file |
219 from a file |
220 """ |
220 """ |
221 return type('MyRealDBConfig', (RealDatabaseConfiguration,), |
221 return type('MyRealDBConfig', (RealDatabaseConfiguration,), |
222 {'sourcesdef': read_config(filename)}) |
222 {'sourcesdef': read_config(filename)}) |
223 |
223 |
224 |
224 |
225 class LivetestConfiguration(BaseApptestConfiguration): |
225 class LivetestConfiguration(BaseApptestConfiguration): |
226 init_repository = False |
226 init_repository = False |
227 |
227 |
228 def __init__(self, cube=None, sourcefile=None, pyro_name=None, |
228 def __init__(self, cube=None, sourcefile=None, pyro_name=None, |
229 log_threshold=logging.CRITICAL): |
229 log_threshold=logging.CRITICAL): |
230 TestServerConfiguration.__init__(self, cube, log_threshold=log_threshold) |
230 TestServerConfiguration.__init__(self, cube, log_threshold=log_threshold) |
231 self.appid = pyro_name or cube |
231 self.appid = pyro_name or cube |
232 # don't change this, else some symlink problems may arise in some |
232 # don't change this, else some symlink problems may arise in some |
252 return True |
252 return True |
253 else: |
253 else: |
254 return False |
254 return False |
255 |
255 |
256 CubicWebConfiguration.cls_adjust_sys_path() |
256 CubicWebConfiguration.cls_adjust_sys_path() |
257 |
257 |
258 def install_sqlite_path(querier): |
258 def install_sqlite_path(querier): |
259 """This patch hotfixes the following sqlite bug : |
259 """This patch hotfixes the following sqlite bug : |
260 - http://www.sqlite.org/cvstrac/tktview?tn=1327,33 |
260 - http://www.sqlite.org/cvstrac/tktview?tn=1327,33 |
261 (some dates are returned as strings rather thant date objects) |
261 (some dates are returned as strings rather thant date objects) |
262 """ |
262 """ |
269 found_date = False |
269 found_date = False |
270 for row, rowdesc in zip(rset, rset.description): |
270 for row, rowdesc in zip(rset, rset.description): |
271 for cellindex, (value, vtype) in enumerate(zip(row, rowdesc)): |
271 for cellindex, (value, vtype) in enumerate(zip(row, rowdesc)): |
272 if vtype in ('Date', 'Datetime') and type(value) is unicode: |
272 if vtype in ('Date', 'Datetime') and type(value) is unicode: |
273 found_date = True |
273 found_date = True |
|
274 value = value.rsplit('.', 1)[0] |
274 try: |
275 try: |
275 row[cellindex] = strptime(value, '%Y-%m-%d %H:%M:%S') |
276 row[cellindex] = strptime(value, '%Y-%m-%d %H:%M:%S') |
276 except: |
277 except: |
277 row[cellindex] = strptime(value, '%Y-%m-%d') |
278 row[cellindex] = strptime(value, '%Y-%m-%d') |
278 if vtype == 'Time' and type(value) is unicode: |
279 if vtype == 'Time' and type(value) is unicode: |
282 except: |
283 except: |
283 # DateTime used as Time? |
284 # DateTime used as Time? |
284 row[cellindex] = strptime(value, '%Y-%m-%d %H:%M:%S') |
285 row[cellindex] = strptime(value, '%Y-%m-%d %H:%M:%S') |
285 if vtype == 'Interval' and type(value) is int: |
286 if vtype == 'Interval' and type(value) is int: |
286 found_date = True |
287 found_date = True |
287 row[cellindex] = DateTimeDelta(0, 0, 0, value) |
288 row[cellindex] = timedelta(0, value, 0) # XXX value is in number of seconds? |
288 if not found_date: |
289 if not found_date: |
289 break |
290 break |
290 return rset |
291 return rset |
291 return new_execute |
292 return new_execute |
292 querier.__class__.execute = wrap_execute(querier.__class__.execute) |
293 querier.__class__.execute = wrap_execute(querier.__class__.execute) |
328 if removecube: |
329 if removecube: |
329 try: |
330 try: |
330 os.remove('%s-cube' % dbfile) |
331 os.remove('%s-cube' % dbfile) |
331 except OSError: |
332 except OSError: |
332 pass |
333 pass |
333 |
334 |
334 def init_test_database_sqlite(config, source, vreg=None): |
335 def init_test_database_sqlite(config, source, vreg=None): |
335 """initialize a fresh sqlite databse used for testing purpose""" |
336 """initialize a fresh sqlite databse used for testing purpose""" |
336 import shutil |
337 import shutil |
337 # remove database file if it exists (actually I know driver == 'sqlite' :) |
338 # remove database file if it exists (actually I know driver == 'sqlite' :) |
338 dbfile = source['system']['db-name'] |
339 dbfile = source['system']['db-name'] |