11 import logging |
11 import logging |
12 from datetime import timedelta |
12 from datetime import timedelta |
13 from os.path import (abspath, join, exists, basename, dirname, normpath, split, |
13 from os.path import (abspath, join, exists, basename, dirname, normpath, split, |
14 isfile, isabs) |
14 isfile, isabs) |
15 |
15 |
16 from cubicweb import CW_SOFTWARE_ROOT, ConfigurationError |
16 from cubicweb import CW_SOFTWARE_ROOT, ConfigurationError, schema, cwconfig |
17 from cubicweb.utils import strptime |
17 from cubicweb.utils import strptime |
18 from cubicweb.toolsutils import read_config |
|
19 from cubicweb.cwconfig import CubicWebConfiguration, merge_options |
|
20 from cubicweb.server.serverconfig import ServerConfiguration |
18 from cubicweb.server.serverconfig import ServerConfiguration |
21 from cubicweb.etwist.twconfig import TwistedConfiguration |
19 from cubicweb.etwist.twconfig import TwistedConfiguration |
|
20 |
|
21 cwconfig.CubicWebConfiguration.cls_adjust_sys_path() |
|
22 |
|
23 # db auto-population configuration ############################################# |
|
24 |
|
25 SYSTEM_ENTITIES = schema.SCHEMA_TYPES | set(( |
|
26 'CWGroup', 'CWUser', 'CWProperty', |
|
27 'State', 'Transition', 'TrInfo', |
|
28 )) |
|
29 |
|
30 SYSTEM_RELATIONS = schema.META_RTYPES | set(( |
|
31 # workflow related |
|
32 'state_of', 'transition_of', 'initial_state', 'allowed_transition', |
|
33 'destination_state', 'in_state', 'wf_info_for', 'from_state', 'to_state', |
|
34 'condition', |
|
35 # cwproperty |
|
36 'for_user', |
|
37 # schema definition |
|
38 'specializes', |
|
39 'relation_type', 'from_entity', 'to_entity', |
|
40 'constrained_by', 'cstrtype', 'widget', |
|
41 'read_permission', 'update_permission', 'delete_permission', 'add_permission', |
|
42 # permission |
|
43 'in_group', 'require_group', 'require_permission', |
|
44 # deducted from other relations |
|
45 'primary_email', |
|
46 )) |
|
47 |
|
48 # content validation configuration ############################################# |
22 |
49 |
23 # validators are used to validate (XML, DTD, whatever) view's content |
50 # validators are used to validate (XML, DTD, whatever) view's content |
24 # validators availables are : |
51 # validators availables are : |
25 # 'dtd' : validates XML + declared DTD |
52 # 'dtd' : validates XML + declared DTD |
26 # 'xml' : guarantees XML is well formed |
53 # 'xml' : guarantees XML is well formed |
27 # None : do not try to validate anything |
54 # None : do not try to validate anything |
|
55 |
|
56 # {'vid': validator} |
28 VIEW_VALIDATORS = {} |
57 VIEW_VALIDATORS = {} |
|
58 |
|
59 |
|
60 # cubicweb test configuration ################################################## |
|
61 |
29 BASE_URL = 'http://testing.fr/cubicweb/' |
62 BASE_URL = 'http://testing.fr/cubicweb/' |
|
63 |
30 DEFAULT_SOURCES = {'system': {'adapter' : 'native', |
64 DEFAULT_SOURCES = {'system': {'adapter' : 'native', |
31 'db-encoding' : 'UTF-8', #'ISO-8859-1', |
65 'db-encoding' : 'UTF-8', #'ISO-8859-1', |
32 'db-user' : u'admin', |
66 'db-user' : u'admin', |
33 'db-password' : 'gingkow', |
67 'db-password' : 'gingkow', |
34 'db-name' : 'tmpdb', |
68 'db-name' : 'tmpdb', |
116 sources = super(TestServerConfiguration, self).sources() |
150 sources = super(TestServerConfiguration, self).sources() |
117 if not sources: |
151 if not sources: |
118 sources = DEFAULT_SOURCES |
152 sources = DEFAULT_SOURCES |
119 return sources |
153 return sources |
120 |
154 |
121 def load_defaults(self): |
|
122 super(TestServerConfiguration, self).load_defaults() |
|
123 # note: don't call global set option here, OptionManager may not yet be initialized |
|
124 # add anonymous user |
|
125 self.set_option('anonymous-user', 'anon') |
|
126 self.set_option('anonymous-password', 'anon') |
|
127 # uncomment the line below if you want rql queries to be logged |
|
128 #self.set_option('query-log-file', '/tmp/test_rql_log.' + `os.getpid()`) |
|
129 self.set_option('sender-name', 'cubicweb-test') |
|
130 self.set_option('sender-addr', 'cubicweb-test@logilab.fr') |
|
131 try: |
|
132 send_to = '%s@logilab.fr' % os.getlogin() |
|
133 except OSError: |
|
134 send_to = '%s@logilab.fr' % (os.environ.get('USER') |
|
135 or os.environ.get('USERNAME') |
|
136 or os.environ.get('LOGNAME')) |
|
137 self.set_option('sender-addr', send_to) |
|
138 self.set_option('default-dest-addrs', send_to) |
|
139 self.set_option('base-url', BASE_URL) |
|
140 |
|
141 |
155 |
142 class BaseApptestConfiguration(TestServerConfiguration, TwistedConfiguration): |
156 class BaseApptestConfiguration(TestServerConfiguration, TwistedConfiguration): |
143 repo_method = 'inmemory' |
157 repo_method = 'inmemory' |
144 options = merge_options(TestServerConfiguration.options + TwistedConfiguration.options) |
158 options = cwconfig.merge_options(TestServerConfiguration.options |
|
159 + TwistedConfiguration.options) |
145 cubicweb_appobject_path = TestServerConfiguration.cubicweb_appobject_path | TwistedConfiguration.cubicweb_appobject_path |
160 cubicweb_appobject_path = TestServerConfiguration.cubicweb_appobject_path | TwistedConfiguration.cubicweb_appobject_path |
146 cube_appobject_path = TestServerConfiguration.cube_appobject_path | TwistedConfiguration.cube_appobject_path |
161 cube_appobject_path = TestServerConfiguration.cube_appobject_path | TwistedConfiguration.cube_appobject_path |
147 |
162 |
148 def available_languages(self, *args): |
163 def available_languages(self, *args): |
149 return ('en', 'fr', 'de') |
164 return ('en', 'fr', 'de') |
161 |
176 |
162 def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None): |
177 def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None): |
163 BaseApptestConfiguration.__init__(self, appid, log_threshold=log_threshold) |
178 BaseApptestConfiguration.__init__(self, appid, log_threshold=log_threshold) |
164 self.init_repository = sourcefile is None |
179 self.init_repository = sourcefile is None |
165 self.sourcefile = sourcefile |
180 self.sourcefile = sourcefile |
166 import re |
181 self.global_set_option('anonymous-user', 'anon') |
167 self.global_set_option('embed-allowed', re.compile('.*')) |
182 self.global_set_option('anonymous-password', 'anon') |
168 |
183 |
169 |
184 def load_configuration(self): |
170 class RealDatabaseConfiguration(ApptestConfiguration): |
185 super(ApptestConfiguration, self).load_configuration() |
171 init_repository = False |
186 self.global_set_option('anonymous-user', 'anon') |
172 sourcesdef = {'system': {'adapter' : 'native', |
187 self.global_set_option('anonymous-password', 'anon') |
173 'db-encoding' : 'UTF-8', #'ISO-8859-1', |
188 |
174 'db-user' : u'admin', |
189 |
175 'db-password' : 'gingkow', |
190 # test database handling ####################################################### |
176 'db-name' : 'seotest', |
191 |
177 'db-driver' : 'postgres', |
192 def init_test_database(config=None, configdir='data'): |
178 'db-host' : None, |
193 """init a test database for a specific driver""" |
179 }, |
194 from cubicweb.dbapi import in_memory_cnx |
180 'admin' : {'login': u'admin', |
195 config = config or TestServerConfiguration(configdir) |
181 'password': u'gingkow', |
196 sources = config.sources() |
182 }, |
197 driver = sources['system']['db-driver'] |
183 } |
198 if driver == 'sqlite': |
184 |
199 init_test_database_sqlite(config) |
185 def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None): |
200 elif driver == 'postgres': |
186 ApptestConfiguration.__init__(self, appid) |
201 init_test_database_postgres(config) |
187 self.init_repository = False |
202 else: |
188 |
203 raise ValueError('no initialization function for driver %r' % driver) |
189 |
204 config._cubes = None # avoid assertion error |
190 def sources(self): |
205 repo, cnx = in_memory_cnx(config, unicode(sources['admin']['login']), |
191 """ |
206 sources['admin']['password'] or 'xxx') |
192 By default, we run tests with the sqlite DB backend. |
207 if driver == 'sqlite': |
193 One may use its own configuration by just creating a |
208 install_sqlite_patch(repo.querier) |
194 'sources' file in the test directory from wich tests are |
209 return repo, cnx |
195 launched. |
210 |
196 """ |
211 |
197 self._sources = self.sourcesdef |
212 def reset_test_database(config): |
198 return self._sources |
213 """init a test database for a specific driver""" |
199 |
214 driver = config.sources()['system']['db-driver'] |
200 |
215 if driver == 'sqlite': |
201 def buildconfig(dbuser, dbpassword, dbname, adminuser, adminpassword, dbhost=None): |
216 reset_test_database_sqlite(config) |
202 """convenience function that builds a real-db configuration class""" |
217 else: |
203 sourcesdef = {'system': {'adapter' : 'native', |
218 raise ValueError('no reset function for driver %r' % driver) |
204 'db-encoding' : 'UTF-8', #'ISO-8859-1', |
219 |
205 'db-user' : dbuser, |
220 |
206 'db-password' : dbpassword, |
221 ### postgres test database handling ############################################ |
207 'db-name' : dbname, |
222 |
208 'db-driver' : 'postgres', |
223 def init_test_database_postgres(config): |
209 'db-host' : dbhost, |
224 """initialize a fresh sqlite databse used for testing purpose""" |
210 }, |
225 if config.init_repository: |
211 'admin' : {'login': adminuser, |
226 from cubicweb.server import init_repository |
212 'password': adminpassword, |
227 init_repository(config, interactive=False, drop=True) |
213 }, |
228 |
214 } |
229 |
215 return type('MyRealDBConfig', (RealDatabaseConfiguration,), |
230 ### sqlite test database handling ############################################## |
216 {'sourcesdef': sourcesdef}) |
231 |
217 |
232 def cleanup_sqlite(dbfile, removetemplate=False): |
218 def loadconfig(filename): |
233 try: |
219 """convenience function that builds a real-db configuration class |
234 os.remove(dbfile) |
220 from a file |
235 os.remove('%s-journal' % dbfile) |
221 """ |
236 except OSError: |
222 return type('MyRealDBConfig', (RealDatabaseConfiguration,), |
237 pass |
223 {'sourcesdef': read_config(filename)}) |
238 if removetemplate: |
224 |
239 try: |
225 |
240 os.remove('%s-template' % dbfile) |
226 class LivetestConfiguration(BaseApptestConfiguration): |
241 except OSError: |
227 init_repository = False |
242 pass |
228 |
243 |
229 def __init__(self, cube=None, sourcefile=None, pyro_name=None, |
244 def reset_test_database_sqlite(config): |
230 log_threshold=logging.CRITICAL): |
245 import shutil |
231 TestServerConfiguration.__init__(self, cube, log_threshold=log_threshold) |
246 dbfile = config.sources()['system']['db-name'] |
232 self.appid = pyro_name or cube |
247 cleanup_sqlite(dbfile) |
233 # don't change this, else some symlink problems may arise in some |
248 template = '%s-template' % dbfile |
234 # environment (e.g. mine (syt) ;o) |
249 if exists(template): |
235 # XXX I'm afraid this test will prevent to run test from a production |
250 shutil.copy(template, dbfile) |
236 # environment |
251 return True |
237 self._sources = None |
252 return False |
238 # instance cube test |
253 |
239 if cube is not None: |
254 def init_test_database_sqlite(config): |
240 self.apphome = self.cube_dir(cube) |
255 """initialize a fresh sqlite databse used for testing purpose""" |
241 elif 'web' in os.getcwd().split(os.sep): |
256 # remove database file if it exists |
242 # web test |
257 dbfile = config.sources()['system']['db-name'] |
243 self.apphome = join(normpath(join(dirname(__file__), '..')), 'web') |
258 if not reset_test_database_sqlite(config): |
244 else: |
259 # initialize the database |
245 # cube test |
260 import shutil |
246 self.apphome = abspath('..') |
261 from cubicweb.server import init_repository |
247 self.sourcefile = sourcefile |
262 init_repository(config, interactive=False) |
248 self.global_set_option('realm', '') |
263 dbfile = config.sources()['system']['db-name'] |
249 self.use_pyro = pyro_name is not None |
264 shutil.copy(dbfile, '%s-template' % dbfile) |
250 |
265 |
251 def pyro_enabled(self): |
|
252 if self.use_pyro: |
|
253 return True |
|
254 else: |
|
255 return False |
|
256 |
|
257 CubicWebConfiguration.cls_adjust_sys_path() |
|
258 |
266 |
259 def install_sqlite_patch(querier): |
267 def install_sqlite_patch(querier): |
260 """This patch hotfixes the following sqlite bug : |
268 """This patch hotfixes the following sqlite bug : |
261 - http://www.sqlite.org/cvstrac/tktview?tn=1327,33 |
269 - http://www.sqlite.org/cvstrac/tktview?tn=1327,33 |
262 (some dates are returned as strings rather thant date objects) |
270 (some dates are returned as strings rather thant date objects) |
291 break |
299 break |
292 return rset |
300 return rset |
293 return new_execute |
301 return new_execute |
294 querier.__class__.execute = wrap_execute(querier.__class__.execute) |
302 querier.__class__.execute = wrap_execute(querier.__class__.execute) |
295 querier.__class__._devtools_sqlite_patched = True |
303 querier.__class__._devtools_sqlite_patched = True |
296 |
|
297 def init_test_database(driver='sqlite', configdir='data', config=None, |
|
298 vreg=None): |
|
299 """init a test database for a specific driver""" |
|
300 from cubicweb.dbapi import in_memory_cnx |
|
301 if vreg and not config: |
|
302 config = vreg.config |
|
303 config = config or TestServerConfiguration(configdir) |
|
304 source = config.sources() |
|
305 if driver == 'sqlite': |
|
306 init_test_database_sqlite(config, source) |
|
307 elif driver == 'postgres': |
|
308 init_test_database_postgres(config, source) |
|
309 else: |
|
310 raise ValueError('no initialization function for driver %r' % driver) |
|
311 config._cubes = None # avoid assertion error |
|
312 repo, cnx = in_memory_cnx(vreg or config, unicode(source['admin']['login']), |
|
313 source['admin']['password'] or 'xxx') |
|
314 if driver == 'sqlite': |
|
315 install_sqlite_patch(repo.querier) |
|
316 return repo, cnx |
|
317 |
|
318 def init_test_database_postgres(config, source, vreg=None): |
|
319 """initialize a fresh sqlite databse used for testing purpose""" |
|
320 if config.init_repository: |
|
321 from cubicweb.server import init_repository |
|
322 init_repository(config, interactive=False, drop=True, vreg=vreg) |
|
323 |
|
324 def cleanup_sqlite(dbfile, removecube=False): |
|
325 try: |
|
326 os.remove(dbfile) |
|
327 os.remove('%s-journal' % dbfile) |
|
328 except OSError: |
|
329 pass |
|
330 if removecube: |
|
331 try: |
|
332 os.remove('%s-template' % dbfile) |
|
333 except OSError: |
|
334 pass |
|
335 |
|
336 def init_test_database_sqlite(config, source, vreg=None): |
|
337 """initialize a fresh sqlite databse used for testing purpose""" |
|
338 import shutil |
|
339 # remove database file if it exists (actually I know driver == 'sqlite' :) |
|
340 dbfile = source['system']['db-name'] |
|
341 cleanup_sqlite(dbfile) |
|
342 template = '%s-template' % dbfile |
|
343 if exists(template): |
|
344 shutil.copy(template, dbfile) |
|
345 else: |
|
346 # initialize the database |
|
347 from cubicweb.server import init_repository |
|
348 init_repository(config, interactive=False, vreg=vreg) |
|
349 shutil.copy(dbfile, template) |
|