webtest.py (13464B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\cherrypy\test\webtest.py 2 import os 3 import pprint 4 import re 5 import socket 6 import sys 7 import time 8 import traceback 9 import types 10 from unittest import * 11 from unittest import _TextTestResult 12 from cherrypy._cpcompat import basestring, HTTPConnection, HTTPSConnection, unicodestr 13 14 def interface(host): 15 if host == '0.0.0.0': 16 return '127.0.0.1' 17 if host == '::': 18 return '::1' 19 return host 20 21 22 class TerseTestResult(_TextTestResult): 23 24 def printErrors(self): 25 if self.errors or self.failures: 26 if self.dots or self.showAll: 27 self.stream.writeln() 28 self.printErrorList('ERROR', self.errors) 29 self.printErrorList('FAIL', self.failures) 30 31 32 class TerseTestRunner(TextTestRunner): 33 34 def _makeResult(self): 35 return TerseTestResult(self.stream, self.descriptions, self.verbosity) 36 37 def run(self, test): 38 result = self._makeResult() 39 test(result) 40 result.printErrors() 41 if not result.wasSuccessful(): 42 self.stream.write('FAILED (') 43 failed, errored = list(map(len, (result.failures, result.errors))) 44 if failed: 45 self.stream.write('failures=%d' % failed) 46 if errored: 47 if failed: 48 self.stream.write(', ') 49 self.stream.write('errors=%d' % errored) 50 self.stream.writeln(')') 51 return result 52 53 54 class ReloadingTestLoader(TestLoader): 55 56 def loadTestsFromName(self, name, module = None): 57 parts = name.split('.') 58 unused_parts = [] 59 if module is None: 60 if not parts: 61 raise ValueError('incomplete test name: %s' % name) 62 else: 63 parts_copy = parts[:] 64 while parts_copy: 65 target = '.'.join(parts_copy) 66 if target in sys.modules: 67 module = reload(sys.modules[target]) 68 parts = unused_parts 69 break 70 else: 71 try: 72 module = __import__(target) 73 parts = unused_parts 74 break 75 except ImportError: 76 unused_parts.insert(0, parts_copy[-1]) 77 del parts_copy[-1] 78 if not parts_copy: 79 raise 80 81 parts = parts[1:] 82 obj = module 83 for part in parts: 84 obj = getattr(obj, part) 85 86 if type(obj) == types.ModuleType: 87 return self.loadTestsFromModule(obj) 88 if isinstance(obj, (type, types.ClassType)) and issubclass(obj, TestCase): 89 return self.loadTestsFromTestCase(obj) 90 if type(obj) == types.UnboundMethodType: 91 return obj.im_class(obj.__name__) 92 if hasattr(obj, '__call__'): 93 test = obj() 94 if not isinstance(test, TestCase) and not isinstance(test, TestSuite): 95 raise ValueError('calling %s returned %s, not a test' % (obj, test)) 96 return test 97 raise ValueError('do not know how to make test from: %s' % obj) 98 99 100 try: 101 if sys.platform[:4] == 'java': 102 103 def getchar(): 104 return sys.stdin.read(1) 105 106 107 else: 108 import msvcrt 109 110 def getchar(): 111 return msvcrt.getch() 112 113 114 except ImportError: 115 import tty, termios 116 117 def getchar(): 118 fd = sys.stdin.fileno() 119 old_settings = termios.tcgetattr(fd) 120 try: 121 tty.setraw(sys.stdin.fileno()) 122 ch = sys.stdin.read(1) 123 finally: 124 termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) 125 126 return ch 127 128 129 class WebCase(TestCase): 130 HOST = '127.0.0.1' 131 PORT = 8000 132 HTTP_CONN = HTTPConnection 133 PROTOCOL = 'HTTP/1.1' 134 scheme = 'http' 135 url = None 136 status = None 137 headers = None 138 body = None 139 encoding = 'utf-8' 140 time = None 141 142 def get_conn(self, auto_open = False): 143 if self.scheme == 'https': 144 cls = HTTPSConnection 145 else: 146 cls = HTTPConnection 147 conn = cls(self.interface(), self.PORT) 148 conn.auto_open = auto_open 149 conn.connect() 150 return conn 151 152 def set_persistent(self, on = True, auto_open = False): 153 try: 154 self.HTTP_CONN.close() 155 except (TypeError, AttributeError): 156 pass 157 158 if on: 159 self.HTTP_CONN = self.get_conn(auto_open=auto_open) 160 elif self.scheme == 'https': 161 self.HTTP_CONN = HTTPSConnection 162 else: 163 self.HTTP_CONN = HTTPConnection 164 165 def _get_persistent(self): 166 return hasattr(self.HTTP_CONN, '__class__') 167 168 def _set_persistent(self, on): 169 self.set_persistent(on) 170 171 persistent = property(_get_persistent, _set_persistent) 172 173 def interface(self): 174 return interface(self.HOST) 175 176 def getPage(self, url, headers = None, method = 'GET', body = None, protocol = None): 177 ServerError.on = False 178 if isinstance(url, unicodestr): 179 url = url.encode('utf-8') 180 if isinstance(body, unicodestr): 181 body = body.encode('utf-8') 182 self.url = url 183 self.time = None 184 start = time.time() 185 result = openURL(url, headers, method, body, self.HOST, self.PORT, self.HTTP_CONN, protocol or self.PROTOCOL) 186 self.time = time.time() - start 187 self.status, self.headers, self.body = result 188 self.cookies = [ ('Cookie', v) for k, v in self.headers if k.lower() == 'set-cookie' ] 189 if ServerError.on: 190 raise ServerError() 191 return result 192 193 interactive = True 194 console_height = 30 195 196 def _handlewebError(self, msg): 197 print '' 198 print ' ERROR: %s' % msg 199 if not self.interactive: 200 raise self.failureException(msg) 201 p = ' Show: [B]ody [H]eaders [S]tatus [U]RL; [I]gnore, [R]aise, or sys.e[X]it >> ' 202 sys.stdout.write(p) 203 sys.stdout.flush() 204 while True: 205 i = getchar().upper() 206 if i not in 'BHSUIRX': 207 continue 208 print i.upper() 209 if i == 'B': 210 for x, line in enumerate(self.body.splitlines()): 211 if (x + 1) % self.console_height == 0: 212 sys.stdout.write('<-- More -->\r') 213 m = getchar().lower() 214 sys.stdout.write(' \r') 215 if m == 'q': 216 break 217 print line 218 219 elif i == 'H': 220 pprint.pprint(self.headers) 221 elif i == 'S': 222 print self.status 223 elif i == 'U': 224 print self.url 225 else: 226 if i == 'I': 227 return 228 if i == 'R': 229 raise self.failureException(msg) 230 elif i == 'X': 231 self.exit() 232 sys.stdout.write(p) 233 sys.stdout.flush() 234 235 def exit(self): 236 sys.exit() 237 238 def assertStatus(self, status, msg = None): 239 if isinstance(status, basestring): 240 if not self.status == status: 241 if msg is None: 242 msg = 'Status (%r) != %r' % (self.status, status) 243 self._handlewebError(msg) 244 elif isinstance(status, int): 245 code = int(self.status[:3]) 246 if code != status: 247 if msg is None: 248 msg = 'Status (%r) != %r' % (self.status, status) 249 self._handlewebError(msg) 250 else: 251 match = False 252 for s in status: 253 if isinstance(s, basestring): 254 if self.status == s: 255 match = True 256 break 257 elif int(self.status[:3]) == s: 258 match = True 259 break 260 261 if not match: 262 if msg is None: 263 msg = 'Status (%r) not in %r' % (self.status, status) 264 self._handlewebError(msg) 265 266 def assertHeader(self, key, value = None, msg = None): 267 lowkey = key.lower() 268 for k, v in self.headers: 269 if k.lower() == lowkey: 270 if value is None or str(value) == v: 271 return v 272 273 if msg is None: 274 if value is None: 275 msg = '%r not in headers' % key 276 else: 277 msg = '%r:%r not in headers' % (key, value) 278 self._handlewebError(msg) 279 280 def assertHeaderItemValue(self, key, value, msg = None): 281 actual_value = self.assertHeader(key, msg=msg) 282 header_values = map(str.strip, actual_value.split(',')) 283 if value in header_values: 284 return value 285 if msg is None: 286 msg = '%r not in %r' % (value, header_values) 287 self._handlewebError(msg) 288 289 def assertNoHeader(self, key, msg = None): 290 lowkey = key.lower() 291 matches = [ k for k, v in self.headers if k.lower() == lowkey ] 292 if matches: 293 if msg is None: 294 msg = '%r in headers' % key 295 self._handlewebError(msg) 296 297 def assertBody(self, value, msg = None): 298 if value != self.body: 299 if msg is None: 300 msg = 'expected body:\n%r\n\nactual body:\n%r' % (value, self.body) 301 self._handlewebError(msg) 302 303 def assertInBody(self, value, msg = None): 304 if value not in self.body: 305 if msg is None: 306 msg = '%r not in body: %s' % (value, self.body) 307 self._handlewebError(msg) 308 309 def assertNotInBody(self, value, msg = None): 310 if value in self.body: 311 if msg is None: 312 msg = '%r found in body' % value 313 self._handlewebError(msg) 314 315 def assertMatchesBody(self, pattern, msg = None, flags = 0): 316 if re.search(pattern, self.body, flags) is None: 317 if msg is None: 318 msg = 'No match for %r in body' % pattern 319 self._handlewebError(msg) 320 321 322 methods_with_bodies = ('POST', 'PUT') 323 324 def cleanHeaders(headers, method, body, host, port): 325 if headers is None: 326 headers = [] 327 found = False 328 for k, v in headers: 329 if k.lower() == 'host': 330 found = True 331 break 332 333 if not found: 334 if port == 80: 335 headers.append(('Host', host)) 336 else: 337 headers.append(('Host', '%s:%s' % (host, port))) 338 if method in methods_with_bodies: 339 found = False 340 for k, v in headers: 341 if k.lower() == 'content-type': 342 found = True 343 break 344 345 if not found: 346 headers.append(('Content-Type', 'application/x-www-form-urlencoded')) 347 headers.append(('Content-Length', str(len(body or '')))) 348 return headers 349 350 351 def shb(response): 352 h = [] 353 key, value = (None, None) 354 for line in response.msg.headers: 355 if line: 356 if line[0] in ' \t': 357 value += line.strip() 358 else: 359 if key and value: 360 h.append((key, value)) 361 key, value = line.split(':', 1) 362 key = key.strip() 363 value = value.strip() 364 365 if key and value: 366 h.append((key, value)) 367 return ('%s %s' % (response.status, response.reason), h, response.read()) 368 369 370 def openURL(url, headers = None, method = 'GET', body = None, host = '127.0.0.1', port = 8000, http_conn = HTTPConnection, protocol = 'HTTP/1.1'): 371 headers = cleanHeaders(headers, method, body, host, port) 372 for trial in range(10): 373 try: 374 if hasattr(http_conn, 'host'): 375 conn = http_conn 376 else: 377 conn = http_conn(interface(host), port) 378 conn._http_vsn_str = protocol 379 conn._http_vsn = int(''.join([ x for x in protocol if x.isdigit() ])) 380 if sys.version_info < (2, 4): 381 382 def putheader(self, header, value): 383 if header == 'Accept-Encoding' and value == 'identity': 384 return 385 self.__class__.putheader(self, header, value) 386 387 import new 388 conn.putheader = new.instancemethod(putheader, conn, conn.__class__) 389 conn.putrequest(method.upper(), url, skip_host=True) 390 else: 391 conn.putrequest(method.upper(), url, skip_host=True, skip_accept_encoding=True) 392 for key, value in headers: 393 conn.putheader(key, value) 394 395 conn.endheaders() 396 if body is not None: 397 conn.send(body) 398 response = conn.getresponse() 399 s, h, b = shb(response) 400 if not hasattr(http_conn, 'host'): 401 conn.close() 402 return (s, h, b) 403 except socket.error: 404 time.sleep(0.5) 405 406 raise 407 408 409 ignored_exceptions = [] 410 ignore_all = False 411 412 class ServerError(Exception): 413 on = False 414 415 416 def server_error(exc = None): 417 if exc is None: 418 exc = sys.exc_info() 419 if ignore_all or exc[0] in ignored_exceptions: 420 return False 421 else: 422 ServerError.on = True 423 print '' 424 print ''.join(traceback.format_exception(*exc)) 425 return True