1 # copyright 2003-2010 LOGILAB S.A. (Paris, FRANCE), all rights reserved. |
|
2 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr |
|
3 # |
|
4 # This file is part of CubicWeb. |
|
5 # |
|
6 # CubicWeb is free software: you can redistribute it and/or modify it under the |
|
7 # terms of the GNU Lesser General Public License as published by the Free |
|
8 # Software Foundation, either version 2.1 of the License, or (at your option) |
|
9 # any later version. |
|
10 # |
|
11 # CubicWeb is distributed in the hope that it will be useful, but WITHOUT |
|
12 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS |
|
13 # FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more |
|
14 # details. |
|
15 # |
|
16 # You should have received a copy of the GNU Lesser General Public License along |
|
17 # with CubicWeb. If not, see <http://www.gnu.org/licenses/>. |
|
18 """provide utilies for web (live) unit testing |
|
19 |
|
20 """ |
|
21 |
|
22 import os |
|
23 import socket |
|
24 import logging |
|
25 from os.path import join, dirname, normpath, abspath |
|
26 from StringIO import StringIO |
|
27 |
|
28 #from twisted.application import service, strports |
|
29 # from twisted.internet import reactor, task |
|
30 from twisted.web2 import channel |
|
31 from twisted.web2 import server |
|
32 from twisted.web2 import static |
|
33 from twisted.internet import reactor |
|
34 from twisted.internet.error import CannotListenError |
|
35 |
|
36 from logilab.common.testlib import TestCase |
|
37 |
|
38 from cubicweb.dbapi import in_memory_repo_cnx |
|
39 from cubicweb.etwist.server import CubicWebRootResource |
|
40 from cubicweb.devtools import BaseApptestConfiguration, init_test_database |
|
41 |
|
42 |
|
43 |
|
44 def get_starturl(port=7777, login=None, passwd=None): |
|
45 if login: |
|
46 return 'http://%s:%s/view?login=%s&password=%s' % (socket.gethostname(), port, login, passwd) |
|
47 else: |
|
48 return 'http://%s:%s/' % (socket.gethostname(), port) |
|
49 |
|
50 |
|
51 class LivetestResource(CubicWebRootResource): |
|
52 """redefines main resource to search for data files in several directories""" |
|
53 |
|
54 def locateChild(self, request, segments): |
|
55 """Indicate which resource to use to process down the URL's path""" |
|
56 if len(segments) and segments[0] == 'data': |
|
57 # Anything in data/ is treated as static files |
|
58 datadir = self.config.locate_resource(segments[1])[0] |
|
59 if datadir: |
|
60 return static.File(str(datadir), segments[1:]) |
|
61 # Otherwise we use this single resource |
|
62 return self, () |
|
63 |
|
64 |
|
65 |
|
66 class LivetestConfiguration(BaseApptestConfiguration): |
|
67 init_repository = False |
|
68 |
|
69 def __init__(self, cube=None, sourcefile=None, pyro_name=None, |
|
70 log_threshold=logging.CRITICAL): |
|
71 BaseApptestConfiguration.__init__(self, cube, log_threshold=log_threshold) |
|
72 self.appid = pyro_name or cube |
|
73 # don't change this, else some symlink problems may arise in some |
|
74 # environment (e.g. mine (syt) ;o) |
|
75 # XXX I'm afraid this test will prevent to run test from a production |
|
76 # environment |
|
77 self._sources = None |
|
78 # instance cube test |
|
79 if cube is not None: |
|
80 self.apphome = self.cube_dir(cube) |
|
81 elif 'web' in os.getcwd().split(os.sep): |
|
82 # web test |
|
83 self.apphome = join(normpath(join(dirname(__file__), '..')), 'web') |
|
84 else: |
|
85 # cube test |
|
86 self.apphome = abspath('..') |
|
87 self.sourcefile = sourcefile |
|
88 self.global_set_option('realm', '') |
|
89 self.use_pyro = pyro_name is not None |
|
90 |
|
91 def pyro_enabled(self): |
|
92 if self.use_pyro: |
|
93 return True |
|
94 else: |
|
95 return False |
|
96 |
|
97 |
|
98 |
|
99 def make_site(cube, options=None): |
|
100 from cubicweb.etwist import twconfig # trigger configuration registration |
|
101 config = LivetestConfiguration(cube, options.sourcefile, |
|
102 pyro_name=options.pyro_name, |
|
103 log_threshold=logging.DEBUG) |
|
104 init_test_database(config=config) |
|
105 # if '-n' in sys.argv: # debug mode |
|
106 cubicweb = LivetestResource(config, debug=True) |
|
107 toplevel = cubicweb |
|
108 website = server.Site(toplevel) |
|
109 cube_dir = config.cube_dir(cube) |
|
110 source = config.sources()['system'] |
|
111 for port in xrange(7777, 7798): |
|
112 try: |
|
113 reactor.listenTCP(port, channel.HTTPFactory(website)) |
|
114 saveconf(cube_dir, port, source['db-user'], source['db-password']) |
|
115 break |
|
116 except CannotListenError: |
|
117 print "port %s already in use, I will try another one" % port |
|
118 else: |
|
119 raise |
|
120 cubicweb.base_url = get_starturl(port=port) |
|
121 print "you can go here : %s" % cubicweb.base_url |
|
122 |
|
123 def runserver(): |
|
124 reactor.run() |
|
125 |
|
126 def saveconf(templhome, port, user, passwd): |
|
127 import pickle |
|
128 conffile = file(join(templhome, 'test', 'livetest.conf'), 'w') |
|
129 |
|
130 pickle.dump((port, user, passwd, get_starturl(port, user, passwd)), |
|
131 conffile) |
|
132 conffile.close() |
|
133 |
|
134 |
|
135 def loadconf(filename='livetest.conf'): |
|
136 import pickle |
|
137 return pickle.load(file(filename)) |
|
138 |
|
139 |
|
140 def execute_scenario(filename, **kwargs): |
|
141 """based on twill.parse.execute_file, but inserts cubicweb extensions""" |
|
142 from twill.parse import _execute_script |
|
143 stream = StringIO('extend_with cubicweb.devtools.cubicwebtwill\n' + file(filename).read()) |
|
144 kwargs['source'] = filename |
|
145 _execute_script(stream, **kwargs) |
|
146 |
|
147 |
|
148 def hijack_twill_output(new_output): |
|
149 from twill import commands as twc |
|
150 from twill import browser as twb |
|
151 twc.OUT = new_output |
|
152 twb.OUT = new_output |
|
153 |
|
154 |
|
155 class LiveTestCase(TestCase): |
|
156 |
|
157 sourcefile = None |
|
158 cube = '' |
|
159 def setUp(self): |
|
160 assert self.cube, "You must specify a cube in your testcase" |
|
161 # twill can be quite verbose ... |
|
162 self.twill_output = StringIO() |
|
163 hijack_twill_output(self.twill_output) |
|
164 # build a config, and get a connection |
|
165 self.config = LivetestConfiguration(self.cube, self.sourcefile) |
|
166 _, user, passwd, _ = loadconf() |
|
167 self.repo, self.cnx = in_memory_repo_cnx(self.config, user, password=passwd) |
|
168 self.setup_db(self.cnx) |
|
169 |
|
170 def tearDown(self): |
|
171 self.teardown_db(self.cnx) |
|
172 |
|
173 |
|
174 def setup_db(self, cnx): |
|
175 """override setup_db() to setup your environment""" |
|
176 |
|
177 def teardown_db(self, cnx): |
|
178 """override teardown_db() to clean up your environment""" |
|
179 |
|
180 def get_loggedurl(self): |
|
181 port, user, passwd, logged_url = loadconf() |
|
182 return logged_url |
|
183 |
|
184 def get_anonurl(self): |
|
185 port, _, _, _ = loadconf() |
|
186 return 'http://%s:%s/view?login=anon&password=anon' % ( |
|
187 socket.gethostname(), port) |
|
188 |
|
189 # convenience |
|
190 execute_scenario = staticmethod(execute_scenario) |
|
191 |
|
192 |
|
193 if __name__ == '__main__': |
|
194 runserver() |
|