[refactoring] introduce add_etype_button function to properly create button to add a new entity
also properly use cw_create_url to generate the URL to add the entity
# 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"importthreadingimportsocketimporthttplibfromurlparseimporturlparsefromtwisted.internetimportreactor,errorfromcubicweb.etwist.serverimportrunfromcubicweb.devtools.testlibimportCubicWebTCfromcubicweb.devtoolsimportApptestConfigurationdefget_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 .. see:: :func:`test.test_support.bind_port` """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')classCubicWebServerConfig(ApptestConfiguration):"""basic configuration class for configuring test server Class attributes: * `ports_range`: list giving range of http ports to test (range(7000, 8000) by default). The first port found as available in `ports_range` will be used to launch the test web server. """ports_range=range(7000,8000)defdefault_base_url(self):port=self['port']orget_available_port(self.ports_range)self.global_set_option('port',port)# force rewrite herereturn'http://127.0.0.1:%d/'%self['port']defpyro_enabled(self):returnFalsedefload_configuration(self):super(CubicWebServerConfig,self).load_configuration()self.global_set_option('force-html-content-type',True)classCubicWebServerTC(CubicWebTC):"""Class for running test web server. See :class:`CubicWebServerConfig`. Class attributes: * `anonymous_allowed`: flag telling if anonymous browsing should be allowed """configcls=CubicWebServerConfig# anonymous is logged by default in cubicweb test casesanonymous_allowed=Truedefstart_server(self):# 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=tt.start()semaphore.acquire()ifnotself.web_thread.isAlive():# XXX race condition with actual thread deathraiseRuntimeError('Could not start the web server')#pre init utils connectionparseurl=urlparse(self.config['base-url'])assertparseurl.port==self.config['port'],(self.config['base-url'],self.config['port'])self._web_test_cnx=httplib.HTTPConnection(parseurl.hostname,parseurl.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?__login=%s&__password=%s"%(user,passwd))assertresponse.status==httplib.SEE_OTHER,response.statusself._ident_cookie=response.getheader('Set-Cookie')assertself._ident_cookiereturnTruedefweb_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):super(CubicWebServerTC,self).setUp()self.start_server()deftearDown(self):try:self.stop_server()excepterror.ReactorNotRunning,err:# Server could be launched manuallyprinterrsuper(CubicWebServerTC,self).tearDown()@classmethoddefinit_config(cls,config):config.set_anonymous_allowed(cls.anonymous_allowed)super(CubicWebServerTC,cls).init_config(config)