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 """this module contains base classes and utilities for integration with running

    19 http server

    20 """

    21 from __future__ import with_statement

    22 

    23 __docformat__ = "restructuredtext en"

    24 

    25 import threading

    26 import socket

    27 import httplib

    28 

    29 from twisted.internet import reactor, error

    30 

    31 from cubicweb.etwist.server import run

    32 from cubicweb.devtools.testlib import CubicWebTC

    33 

    34 

    35 def get_available_port(ports_scan):

    36     """return the first available port from the given ports range

    37 

    38     Try to connect port by looking for refused connection (111) or transport

    39     endpoint already connected (106) errors

    40 

    41     Raise a RuntimeError if no port can be found

    42 

    43     :type ports_range: list

    44     :param ports_range: range of ports to test

    45     :rtype: int

    46     """

    47     for port in ports_scan:

    48         try:

    49             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    50             sock = s.connect(("localhost", port))

    51         except socket.error, err:

    52             if err.args[0] in (111, 106):

    53                 return port

    54         finally:

    55             s.close()

    56     raise RuntimeError('get_available_port([ports_range]) cannot find an available port')

    57 

    58 class CubicWebServerTC(CubicWebTC):

    59     """basic class for running test server

    60 

    61     :param ports_range: range of http ports to test (range(7000, 8000) by default)

    62     :type ports_range: iterable

    63     :param anonymous_logged: is anonymous user logged by default ? (True by default)

    64     :type anonymous_logged: bool

    65     :param test_url: base url used by server

    66     :param test_host: server host

    67     :param test_port: server port

    68 

    69     The first port found as available in `ports_range` will be used to launch

    70     the test server

    71     """

    72     ports_range = range(7000, 8000)

    73     # anonymous is logged by default in cubicweb test cases

    74     anonymous_logged = True

    75     test_host='127.0.0.1'

    76 

    77 

    78 

    79     @property

    80     def test_url(self):

    81         return 'http://%s:%d/' % (self.test_host, self.test_port)

    82 

    83     def init_server(self):

    84         self.test_port = get_available_port(self.ports_range)

    85         self.config['port'] = self.test_port

    86         self.config['base-url'] = self.test_url

    87         self.config['force-html-content-type'] = True

    88         self.config['pyro-server'] = False

    89 

    90     def start_server(self):

    91         self.config.pyro_enabled = lambda : False

    92         # use a semaphore to avoid starting test while the http server isn't

    93         # fully initilialized

    94         semaphore = threading.Semaphore(0)

    95         def safe_run(*args, **kwargs):

    96             try:

    97                 run(*args, **kwargs)

    98             finally:

    99                 semaphore.release()

   100 

   101         reactor.addSystemEventTrigger('after', 'startup', semaphore.release)

   102         t = threading.Thread(target=safe_run, name='cubicweb_test_web_server',

   103                              args=(self.config, self.vreg, True))

   104         self.web_thread = t

   105         if not self.anonymous_logged:

   106                 self.config.global_set_option('anonymous-user', None)

   107         t.start()

   108         semaphore.acquire()

   109         if not self.web_thread.isAlive():

   110             # XXX race condition with actual thread death

   111             raise RuntimeError('Could not start the web server')

   112         #pre init utils connection

   113         self._web_test_cnx = httplib.HTTPConnection(self.test_host, self.test_port)

   114         self._ident_cookie = None

   115 

   116     def stop_server(self, timeout=15):

   117         """Stop the webserver, waiting for the thread to return"""

   118         if self._web_test_cnx is None:

   119             self.web_logout()

   120             self._web_test_cnx.close()

   121         try:

   122             reactor.stop()

   123             self.web_thread.join(timeout)

   124             assert not self.web_thread.isAlive()

   125 

   126         finally:

   127             reactor.__init__()

   128 

   129     def web_login(self, user=None, passwd=None):

   130         """Log the current http session for the provided credential

   131 

   132         If no user is provided, admin connection are used.

   133         """

   134         if user is None:

   135             user  = self.admlogin

   136             passwd = self.admpassword

   137         if passwd is None:

   138             passwd = user

   139         self.login(user)

   140         response = self.web_get("?__login=%s&__password=%s" %

   141                                 (user, passwd))

   142         assert response.status == httplib.SEE_OTHER, response.status

   143         self._ident_cookie = response.getheader('Set-Cookie')

   144         return True

   145 

   146     def web_logout(self, user='admin', pwd=None):

   147         """Log out current http user"""

   148         if self._ident_cookie is not None:

   149             response = self.web_get('logout')

   150         self._ident_cookie = None

   151 

   152     def web_get(self, path='', headers=None):

   153         """Return an httplib.HTTPResponse object for the specified path

   154 

   155         Use available credential if available.

   156         """

   157         if headers is None:

   158             headers = {}

   159         if self._ident_cookie is not None:

   160             assert 'Cookie' not in headers

   161             headers['Cookie'] = self._ident_cookie

   162         self._web_test_cnx.request("GET", '/' + path, headers=headers)

   163         response = self._web_test_cnx.getresponse()

   164         response.body = response.read() # to chain request

   165         response.read = lambda : response.body

   166         return response

   167 

   168     def setUp(self):

   169         CubicWebTC.setUp(self)

   170         self.init_server()

   171         self.start_server()

   172 

   173     def tearDown(self):

   174         try:

   175             self.stop_server()

   176         except error.ReactorNotRunning, err:

   177             # Server could be launched manually

   178             print err

   179         CubicWebTC.tearDown(self)

   180