# copyright 2003-2010 LOGILAB S.A. (Paris, FRANCE), all rights reserved.# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr## This file is part of CubicWeb.## CubicWeb is free software: you can redistribute it and/or modify it under the# terms of the GNU Lesser General Public License as published by the Free# Software Foundation, either version 2.1 of the License, or (at your option)# any later version.## CubicWeb is distributed in the hope that it will be useful, but WITHOUT# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more# details.## You should have received a copy of the GNU Lesser General Public License along# with CubicWeb. If not, see <http://www.gnu.org/licenses/>."""this module contains base classes and utilities for integration with runninghttp server"""from__future__importwith_statement__docformat__="restructuredtext en"importthreadingimportsocketimporthttplibfromtwisted.internetimportreactor,errorfromcubicweb.etwist.serverimportrunfromcubicweb.devtools.testlibimportCubicWebTCdefget_available_port(ports_scan):"""return the first available port from the given ports range Try to connect port by looking for refused connection (111) or transport endpoint already connected (106) errors Raise a RuntimeError if no port can be found :type ports_range: list :param ports_range: range of ports to test :rtype: int """forportinports_scan:try:s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)sock=s.connect(("localhost",port))exceptsocket.error,err:iferr.args[0]in(111,106):returnportfinally:s.close()raiseRuntimeError('get_available_port([ports_range]) cannot find an available port')classCubicWebServerTC(CubicWebTC):"""basic class for running test server :param ports_range: range of http ports to test (range(7000, 8000) by default) :type ports_range: iterable :param anonymous_logged: is anonymous user logged by default ? (True by default) :type anonymous_logged: bool :param test_url: base url used by server :param test_host: server host :param test_port: server port The first port found as available in `ports_range` will be used to launch the test server """ports_range=range(7000,8000)# anonymous is logged by default in cubicweb test casesanonymous_logged=Truetest_host='127.0.0.1'@propertydeftest_url(self):return'http://%s:%d/'%(self.test_host,self.test_port)definit_server(self):self.test_port=get_available_port(self.ports_range)self.config['port']=self.test_portself.config['base-url']=self.test_urlself.config['force-html-content-type']=Trueself.config['pyro-server']=Falsedefstart_server(self):self.config.pyro_enabled=lambda:False# use a semaphore to avoid starting test while the http server isn't# fully initilializedsemaphore=threading.Semaphore(0)defsafe_run(*args,**kwargs):try:run(*args,**kwargs)finally:semaphore.release()reactor.addSystemEventTrigger('after','startup',semaphore.release)t=threading.Thread(target=safe_run,name='cubicweb_test_web_server',args=(self.config,self.vreg,True))self.web_thread=tifnotself.anonymous_logged:self.config.global_set_option('anonymous-user',None)t.start()semaphore.acquire()ifnotself.web_thread.isAlive():# XXX race condition with actual thread deathraiseRuntimeError('Could not start the web server')#pre init utils connectionself._web_test_cnx=httplib.HTTPConnection(self.test_host,self.test_port)self._ident_cookie=Nonedefstop_server(self,timeout=15):"""Stop the webserver, waiting for the thread to return"""ifself._web_test_cnxisNone:self.web_logout()self._web_test_cnx.close()try:reactor.stop()self.web_thread.join(timeout)assertnotself.web_thread.isAlive()finally:reactor.__init__()defweb_login(self,user=None,passwd=None):"""Log the current http session for the provided credential If no user is provided, admin connection are used. """ifuserisNone:user=self.admloginpasswd=self.admpasswordifpasswdisNone:passwd=userself.login(user)response=self.web_get("?__login=%s&__password=%s"%(user,passwd))assertresponse.status==httplib.SEE_OTHER,response.statusself._ident_cookie=response.getheader('Set-Cookie')returnTruedefweb_logout(self,user='admin',pwd=None):"""Log out current http user"""ifself._ident_cookieisnotNone:response=self.web_get('logout')self._ident_cookie=Nonedefweb_get(self,path='',headers=None):"""Return an httplib.HTTPResponse object for the specified path Use available credential if available. """ifheadersisNone:headers={}ifself._ident_cookieisnotNone:assert'Cookie'notinheadersheaders['Cookie']=self._ident_cookieself._web_test_cnx.request("GET",'/'+path,headers=headers)response=self._web_test_cnx.getresponse()response.body=response.read()# to chain requestresponse.read=lambda:response.bodyreturnresponsedefsetUp(self):CubicWebTC.setUp(self)self.init_server()self.start_server()deftearDown(self):try:self.stop_server()excepterror.ReactorNotRunning,err:# Server could be launched manuallyprinterrCubicWebTC.tearDown(self)