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             return port

    52         except socket.error, err:

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

    54                 return port

    55         finally:

    56             s.close()

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

    58 

    59 class CubicWebServerTC(CubicWebTC):

    60     """basic class for running test server

    61 

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

    63     :type ports_range: iterable

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

    65     :type anonymous_logged: bool

    66     :param test_url: base url used by server

    67     :param test_host: server host

    68     :param test_port: server port

    69 

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

    71     the test server

    72     """

    73     ports_range = range(7000, 8000)

    74     # anonymous is logged by default in cubicweb test cases

    75     anonymous_logged = True

    76     test_host='127.0.0.1'

    77 

    78 

    79 

    80     @property

    81     def test_url(self):

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

    83 

    84     def init_server(self):

    85         self.test_port = get_available_port(self.ports_range)

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

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

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

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

    90 

    91     def start_server(self):

    92         self.config.pyro_enabled = lambda : False

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

    94         # fully initilialized

    95         semaphore = threading.Semaphore(0)

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

    97             try:

    98                 run(*args, **kwargs)

    99             finally:

   100                 semaphore.release()

   101 

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

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

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

   105         self.web_thread = t

   106         if not self.anonymous_logged:

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

   108         t.start()

   109         semaphore.acquire()

   110         if not self.web_thread.isAlive():

   111             # XXX race condition with actual thread death

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

   113         #pre init utils connection

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

   115         self._ident_cookie = None

   116 

   117     def stop_server(self, timeout=15):

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

   119         if self._web_test_cnx is None:

   120             self.web_logout()

   121             self._web_test_cnx.close()

   122         try:

   123             reactor.stop()

   124             self.web_thread.join(timeout)

   125             assert not self.web_thread.isAlive()

   126 

   127         finally:

   128             reactor.__init__()

   129 

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

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

   132 

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

   134         """

   135         if user is None:

   136             user  = self.admlogin

   137             passwd = self.admpassword

   138         if passwd is None:

   139             passwd = user

   140         self.login(user)

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

   142                                 (user, passwd))

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

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

   145         return True

   146 

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

   148         """Log out current http user"""

   149         if self._ident_cookie is not None:

   150             response = self.web_get('logout')

   151         self._ident_cookie = None

   152 

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

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

   155 

   156         Use available credential if available.

   157         """

   158         if headers is None:

   159             headers = {}

   160         if self._ident_cookie is not None:

   161             assert 'Cookie' not in headers

   162             headers['Cookie'] = self._ident_cookie

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

   164         response = self._web_test_cnx.getresponse()

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

   166         response.read = lambda : response.body

   167         return response

   168 

   169     def setUp(self):

   170         CubicWebTC.setUp(self)

   171         self.init_server()

   172         self.start_server()

   173 

   174     def tearDown(self):

   175         try:

   176             self.stop_server()

   177         except error.ReactorNotRunning, err:

   178             # Server could be launched manually

   179             print err

   180         CubicWebTC.tearDown(self)

   181