|
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) |