diff -r 058bb3dc685f -r 0b59724cb3f2 devtools/httptest.py --- a/devtools/httptest.py Mon Jan 04 18:40:30 2016 +0100 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,170 +0,0 @@ -# copyright 2003-2014 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 . -"""this module contains base classes and utilities for integration with running -http server -""" -from __future__ import print_function - -__docformat__ = "restructuredtext en" - -import random -import threading -import socket - -from six.moves import range, http_client -from six.moves.urllib.parse import urlparse - - -from cubicweb.devtools.testlib import CubicWebTC -from cubicweb.devtools import ApptestConfiguration - - -def get_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` - """ - ports_scan = list(ports_scan) - random.shuffle(ports_scan) # lower the chance of race condition - for port in ports_scan: - try: - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock = s.connect(("localhost", port)) - except socket.error as err: - if err.args[0] in (111, 106): - return port - finally: - s.close() - raise RuntimeError('get_available_port([ports_range]) cannot find an available port') - - -class CubicWebServerTC(CubicWebTC): - """Class for running a Twisted-based test web server. - """ - ports_range = range(7000, 8000) - - def start_server(self): - from twisted.internet import reactor - from cubicweb.etwist.server import run - # use a semaphore to avoid starting test while the http server isn't - # fully initilialized - semaphore = threading.Semaphore(0) - def safe_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, True), kwargs={'repo': self.repo}) - self.web_thread = t - t.start() - semaphore.acquire() - if not self.web_thread.isAlive(): - # XXX race condition with actual thread death - raise RuntimeError('Could not start the web server') - #pre init utils connection - parseurl = urlparse(self.config['base-url']) - assert parseurl.port == self.config['port'], (self.config['base-url'], self.config['port']) - self._web_test_cnx = http_client.HTTPConnection(parseurl.hostname, - parseurl.port) - self._ident_cookie = None - - def stop_server(self, timeout=15): - """Stop the webserver, waiting for the thread to return""" - from twisted.internet import reactor - if self._web_test_cnx is None: - self.web_logout() - self._web_test_cnx.close() - try: - reactor.stop() - self.web_thread.join(timeout) - assert not self.web_thread.isAlive() - - finally: - reactor.__init__() - - def web_login(self, user=None, passwd=None): - """Log the current http session for the provided credential - - If no user is provided, admin connection are used. - """ - if user is None: - user = self.admlogin - passwd = self.admpassword - if passwd is None: - passwd = user - response = self.web_get("login?__login=%s&__password=%s" % - (user, passwd)) - assert response.status == http_client.SEE_OTHER, response.status - self._ident_cookie = response.getheader('Set-Cookie') - assert self._ident_cookie - return True - - def web_logout(self, user='admin', pwd=None): - """Log out current http user""" - if self._ident_cookie is not None: - response = self.web_get('logout') - self._ident_cookie = None - - def web_request(self, path='', method='GET', body=None, headers=None): - """Return an http_client.HTTPResponse object for the specified path - - Use available credential if available. - """ - if headers is None: - headers = {} - if self._ident_cookie is not None: - assert 'Cookie' not in headers - headers['Cookie'] = self._ident_cookie - self._web_test_cnx.request(method, '/' + path, headers=headers, body=body) - response = self._web_test_cnx.getresponse() - response.body = response.read() # to chain request - response.read = lambda : response.body - return response - - def web_get(self, path='', body=None, headers=None): - return self.web_request(path=path, body=body, headers=headers) - - def setUp(self): - super(CubicWebServerTC, self).setUp() - port = self.config['port'] or get_available_port(self.ports_range) - self.config.global_set_option('port', port) # force rewrite here - self.config.global_set_option('base-url', 'http://127.0.0.1:%d/' % port) - # call load_configuration again to let the config reset its datadir_url - self.config.load_configuration() - self.start_server() - - def tearDown(self): - from twisted.internet import error - try: - self.stop_server() - except error.ReactorNotRunning as err: - # Server could be launched manually - print(err) - super(CubicWebServerTC, self).tearDown()