equal
deleted
inserted
replaced
16 # You should have received a copy of the GNU Lesser General Public License along |
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/>. |
17 # with CubicWeb. If not, see <http://www.gnu.org/licenses/>. |
18 """this module contains base classes and utilities for integration with running |
18 """this module contains base classes and utilities for integration with running |
19 http server |
19 http server |
20 """ |
20 """ |
21 from __future__ import print_function |
|
22 |
21 |
|
22 import http.client |
23 import random |
23 import random |
24 import threading |
24 import threading |
25 import socket |
25 import socket |
26 |
26 from urllib.parse import urlparse |
27 from six.moves import range, http_client |
|
28 from six.moves.urllib.parse import urlparse |
|
29 |
|
30 |
27 |
31 from cubicweb.devtools.testlib import CubicWebTC |
28 from cubicweb.devtools.testlib import CubicWebTC |
32 |
29 |
33 |
30 |
34 def get_available_port(ports_scan): |
31 def get_available_port(ports_scan): |
81 passwd = self.admpassword |
78 passwd = self.admpassword |
82 if passwd is None: |
79 if passwd is None: |
83 passwd = user |
80 passwd = user |
84 response = self.web_get("login?__login=%s&__password=%s" % |
81 response = self.web_get("login?__login=%s&__password=%s" % |
85 (user, passwd)) |
82 (user, passwd)) |
86 assert response.status == http_client.SEE_OTHER, response.status |
83 assert response.status == http.client.SEE_OTHER, response.status |
87 self._ident_cookie = response.getheader('Set-Cookie') |
84 self._ident_cookie = response.getheader('Set-Cookie') |
88 assert self._ident_cookie |
85 assert self._ident_cookie |
89 return True |
86 return True |
90 |
87 |
91 def web_logout(self, user='admin', pwd=None): |
88 def web_logout(self, user='admin', pwd=None): |
93 if self._ident_cookie is not None: |
90 if self._ident_cookie is not None: |
94 self.web_get('logout') |
91 self.web_get('logout') |
95 self._ident_cookie = None |
92 self._ident_cookie = None |
96 |
93 |
97 def web_request(self, path='', method='GET', body=None, headers=None): |
94 def web_request(self, path='', method='GET', body=None, headers=None): |
98 """Return an http_client.HTTPResponse object for the specified path |
95 """Return an http.client.HTTPResponse object for the specified path |
99 |
96 |
100 Use available credential if available. |
97 Use available credential if available. |
101 """ |
98 """ |
102 if headers is None: |
99 if headers is None: |
103 headers = {} |
100 headers = {} |
129 |
126 |
130 class CubicWebServerTC(_CubicWebServerTC): |
127 class CubicWebServerTC(_CubicWebServerTC): |
131 def start_server(self): |
128 def start_server(self): |
132 from cubicweb.wsgi.handler import CubicWebWSGIApplication |
129 from cubicweb.wsgi.handler import CubicWebWSGIApplication |
133 from wsgiref import simple_server |
130 from wsgiref import simple_server |
134 from six.moves import queue |
131 import queue |
135 |
132 |
136 config = self.config |
133 config = self.config |
137 port = config['port'] or 8080 |
134 port = config['port'] or 8080 |
138 interface = config['interface'] |
135 interface = config['interface'] |
139 handler_cls = simple_server.WSGIRequestHandler |
136 handler_cls = simple_server.WSGIRequestHandler |
162 if not flag: |
159 if not flag: |
163 t.join() |
160 t.join() |
164 self.fail(start_flag.get()) |
161 self.fail(start_flag.get()) |
165 parseurl = urlparse(self.config['base-url']) |
162 parseurl = urlparse(self.config['base-url']) |
166 assert parseurl.port == self.config['port'], (self.config['base-url'], self.config['port']) |
163 assert parseurl.port == self.config['port'], (self.config['base-url'], self.config['port']) |
167 self._web_test_cnx = http_client.HTTPConnection(parseurl.hostname, |
164 self._web_test_cnx = http.client.HTTPConnection(parseurl.hostname, |
168 parseurl.port) |
165 parseurl.port) |
169 self._ident_cookie = None |
166 self._ident_cookie = None |
170 |
167 |
171 def stop_server(self, timeout=15): |
168 def stop_server(self, timeout=15): |
172 if self._web_test_cnx is None: |
169 if self._web_test_cnx is None: |