funkload.FunkLoadTestCase

53 """Unit test with browser and configuration capabilties.""" 54 # ------------------------------------------------------------ 55 # Initialisation 56 #

57 - def __init__(self, methodName='runTest', options=None):

58 """Initialise the test case. 59 60 Note that methodName is encoded in bench mode to provide additional 61 information like thread_id, concurrent virtual users...""" 62 if mmn_is_bench(methodName): 63 self.in_bench_mode = True 64 else: 65 self.in_bench_mode = False 66 self.test_name, self.cycle, self.cvus, self.thread_id = mmn_decode( 67 methodName) 68 self.meta_method_name = methodName 69 self.suite_name = self.__class__.__name__ 70 unittest.TestCase.__init__(self, methodName=self.test_name) 71 self._response = None 72 self.options = options 73 self.debug_level = getattr(options, 'debug_level', 0) 74 self._funkload_init() 75 self._dump_dir = getattr(options, 'dump_dir', None) 76 self._dumping = self._dump_dir and True or False 77 self._viewing = getattr(options, 'firefox_view', False) 78 self._accept_invalid_links = getattr(options, 'accept_invalid_links', 79 False) 80 self._simple_fetch = getattr(options, 'simple_fetch', False) 81 self._bench_label = getattr(options, 'label', None) 82 self._stop_on_fail = getattr(options, 'stop_on_fail', False) 83 self._pause = getattr(options, 'pause', False) 84 self._keyfile_path = None 85 self._certfile_path = None 86 if self._viewing and not self._dumping: 87 # viewing requires dumping contents 88 self._dumping = True 89 self._dump_dir = mkdtemp('_funkload') 90 self._loop_mode = getattr(options, 'loop_steps', False) 91 if self._loop_mode: 92 if options.loop_steps.count(':'): 93 steps = options.loop_steps.split(':') 94 self._loop_steps = range(int(steps[0]), int(steps[1])) 95 else: 96 self._loop_steps = [int(options.loop_steps)] 97 self._loop_number = options.loop_number 98 self._loop_recording = False 99 self._loop_records = []

100

101 - def _funkload_init(self):

102 """Initialize a funkload test case using a configuration file.""" 103 # look into configuration file 104 config_directory = os.getenv('FL_CONF_PATH', '.') 105 config_path = os.path.join(config_directory, 106 self.__class__.__name__ + '.conf') 107 config_path = os.path.abspath(config_path) 108 if not os.path.exists(config_path): 109 config_path = "Missing: "+ config_path 110 config = ConfigParser() 111 config.read(config_path) 112 self._config = config 113 self._config_path = config_path 114 self.default_user_agent = self.conf_get('main', 'user_agent', 115 'FunkLoad/%s' % get_version(), 116 quiet=True) 117 if self.in_bench_mode: 118 section = 'bench' 119 else: 120 section = 'ftest' 121 ok_codes = self.conf_getList(section, 'ok_codes', 122 [200, 301, 302, 303, 307], 123 quiet=True) 124 self.ok_codes = map(int, ok_codes) 125 self.sleep_time_min = self.conf_getFloat(section, 'sleep_time_min', 0) 126 self.sleep_time_max = self.conf_getFloat(section, 'sleep_time_max', 0) 127 self.log_to = self.conf_get(section, 'log_to', 'console file') 128 self.log_path = self.conf_get(section, 'log_path', 'funkload.log') 129 self.result_path = os.path.abspath( 130 self.conf_get(section, 'result_path', 'funkload.xml')) 131 132 # init loggers 133 self.logger = get_default_logger(self.log_to, self.log_path) 134 self.logger_result = get_default_logger(log_to="xml", 135 log_path=self.result_path, 136 name="FunkLoadResult") 137 #self.logd('_funkload_init config [%s], log_to [%s],' 138 # ' log_path [%s], result [%s].' % ( 139 # self._config_path, self.log_to, self.log_path, self.result_path)) 140 141 # init webunit browser (passing a fake methodName) 142 self._browser = WebTestCase(methodName='log') 143 self.clearContext()

144 145 #self.logd('# FunkLoadTestCase._funkload_init done') 146

147 - def clearContext(self):

148 """Reset the testcase.""" 149 self._browser.clearContext() 150 self._browser.css = {} 151 self._browser.history = [] 152 self._browser.extra_headers = [] 153 if self.debug_level >= 3: 154 self._browser.debug_headers = True 155 else: 156 self._browser.debug_headers = False 157 self.step_success = True 158 self.test_status = 'Successful' 159 self.steps = 0 160 self.page_responses = 0 161 self.total_responses = 0 162 self.total_time = 0.0 163 self.total_pages = self.total_images = 0 164 self.total_links = self.total_redirects = 0 165 self.total_xmlrpc = 0 166 self.clearBasicAuth() 167 self.clearHeaders() 168 self.clearKeyAndCertificateFile() 169 self.setUserAgent(self.default_user_agent) 170 171 self.logdd('FunkLoadTestCase.clearContext done')

172 173 174 175 #------------------------------------------------------------ 176 # browser simulation 177 #

178 - def _connect(self, url, params, ok_codes, rtype, description):

179 """Handle fetching, logging, errors and history.""" 180 if params is None and rtype == 'post': 181 # enable empty post 182 params = [] 183 t_start = time.time() 184 try: 185 response = self._browser.fetch(url, params, ok_codes=ok_codes, 186 key_file=self._keyfile_path, 187 cert_file=self._certfile_path) 188 except: 189 etype, value, tback = sys.exc_info() 190 t_stop = time.time() 191 t_delta = t_stop - t_start 192 self.total_time += t_delta 193 self.step_success = False 194 self.test_status = 'Failure' 195 self.logd(' Failed in %.3fs' % t_delta) 196 if etype is HTTPError: 197 self._log_response(value.response, rtype, description, 198 t_start, t_stop, log_body=True) 199 if self._dumping: 200 self._dump_content(value.response) 201 raise self.failureException, str(value.response) 202 else: 203 self._log_response_error(url, rtype, description, t_start, 204 t_stop) 205 if etype is SocketError: 206 raise SocketError("Can't load %s." % url) 207 raise 208 t_stop = time.time() 209 # Log response 210 t_delta = t_stop - t_start 211 self.total_time += t_delta 212 if rtype in ('post', 'get'): 213 self.total_pages += 1 214 elif rtype == 'redirect': 215 self.total_redirects += 1 216 elif rtype == 'link': 217 self.total_links += 1 218 if rtype in ('post', 'get', 'redirect'): 219 # this is a valid referer for the next request 220 self.setHeader('Referer', url) 221 self._browser.history.append((rtype, url)) 222 self.logd(' Done in %.3fs' % t_delta) 223 self._log_response(response, rtype, description, t_start, t_stop) 224 if self._dumping: 225 self._dump_content(response) 226 return response

227

228 - def _browse(self, url_in, params_in=None, 229 description=None, ok_codes=None, 230 method='post', 231 follow_redirect=True, load_auto_links=True, 232 sleep=True):

233 """Simulate a browser handle redirects, load/cache css and images.""" 234 self._response = None 235 # Loop mode 236 if self._loop_mode: 237 if self.steps == self._loop_steps[0]: 238 self._loop_recording = True 239 self.logi('Loop mode start recording') 240 if self._loop_recording: 241 self._loop_records.append((url_in, params_in, description, 242 ok_codes, method, follow_redirect, 243 load_auto_links, False)) 244 # ok codes 245 if ok_codes is None: 246 ok_codes = self.ok_codes 247 if type(params_in) is DictType: 248 params_in = params_in.items() 249 params = [] 250 if params_in: 251 if isinstance(params_in, Data): 252 params = params_in 253 else: 254 for key, value in params_in: 255 if type(value) is DictType: 256 for val, selected in value.items(): 257 if selected: 258 params.append((key, val)) 259 elif type(value) in (ListType, TupleType): 260 for val in value: 261 params.append((key, val)) 262 else: 263 params.append((key, value)) 264 265 if method == 'get' and params: 266 url = url_in + '?' + urlencode(params) 267 else: 268 url = url_in 269 if method == 'get': 270 params = None 271 272 if method == 'get': 273 self.logd('GET: %s\n\tPage %i: %s ...' % (url, self.steps, 274 description or '')) 275 else: 276 url = url_in 277 self.logd('POST: %s %s\n\tPage %i: %s ...' % (url, str(params), 278 self.steps, 279 description or '')) 280 # Fetching 281 response = self._connect(url, params, ok_codes, method, description) 282 283 # Check redirection 284 if follow_redirect and response.code in (301, 302, 303, 307): 285 max_redirect_count = 10 286 thread_sleep() # give a chance to other threads 287 while response.code in (301, 302, 303, 307) and max_redirect_count: 288 # Figure the location - which may be relative 289 newurl = response.headers['Location'] 290 url = urljoin(url_in, newurl) 291 # save the current url as the base for future redirects 292 url_in = url 293 self.logd(' Load redirect link: %s' % url) 294 response = self._connect(url, None, ok_codes, 'redirect', None) 295 max_redirect_count -= 1 296 if not max_redirect_count: 297 self.logd(' WARNING Too many redirects give up.') 298 299 # Load auto links (css and images) 300 response.is_html = is_html(response.body) 301 if load_auto_links and response.is_html and not self._simple_fetch: 302 self.logd(' Load css and images...') 303 page = response.body 304 t_start = time.time() 305 c_start = self.total_time 306 try: 307 # pageImages is patched to call _log_response on all links 308 self._browser.pageImages(url, page, self) 309 except HTTPError, error: 310 if self._accept_invalid_links: 311 self.logd(' ' + str(error)) 312 else: 313 t_stop = time.time() 314 t_delta = t_stop - t_start 315 self.step_success = False 316 self.test_status = 'Failure' 317 self.logd(' Failed in ~ %.2fs' % t_delta) 318 # XXX The duration logged for this response is wrong 319 self._log_response(error.response, 'link', None, 320 t_start, t_stop, log_body=True) 321 raise self.failureException, str(error) 322 c_stop = self.total_time 323 self.logd(' Done in %.3fs' % (c_stop - c_start)) 324 if sleep: 325 self.sleep() 326 self._response = response 327 328 # Loop mode 329 if self._loop_mode and self.steps == self._loop_steps[-1]: 330 self._loop_recording = False 331 self.logi('Loop mode end recording.') 332 t_start = self.total_time 333 count = 0 334 for i in range(self._loop_number): 335 self.logi('Loop mode replay %i' % i) 336 for record in self._loop_records: 337 count += 1 338 self.steps += 1 339 self._browse(*record) 340 t_delta = self.total_time - t_start 341 text = ('End of loop: %d pages rendered in %.3fs, ' 342 'avg of %.3fs per page, ' 343 '%.3f SPPS without concurrency.' % (count, t_delta, 344 t_delta/count, 345 count/t_delta)) 346 self.logi(text) 347 trace(text + '\n') 348 349 return response

350

351 - def post(self, url, params=None, description=None, ok_codes=None):

352 """POST method on url with params.""" 353 self.steps += 1 354 self.page_responses = 0 355 response = self._browse(url, params, description, ok_codes, 356 method="post") 357 return response

358

359 - def get(self, url, params=None, description=None, ok_codes=None):

360 """GET method on url adding params.""" 361 self.steps += 1 362 self.page_responses = 0 363 response = self._browse(url, params, description, ok_codes, 364 method="get") 365 return response

366

367 - def exists(self, url, params=None, description="Checking existence"):

368 """Try a GET on URL return True if the page exists or False.""" 369 resp = self.get(url, params, description=description, 370 ok_codes=[200, 301, 302, 303, 307, 404, 503]) 371 if resp.code not in [200, 301, 302, 303, 307]: 372 self.logd('Page %s not found.' % url) 373 return False 374 self.logd('Page %s exists.' % url) 375 return True

376

377 - def xmlrpc(self, url_in, method_name, params=None, description=None):

378 """Call an xml rpc method_name on url with params.""" 379 self.steps += 1 380 self.page_responses = 0 381 self.logd('XMLRPC: %s::%s\n\tCall %i: %s ...' % (url_in, method_name, 382 self.steps, 383 description or '')) 384 response = None 385 t_start = time.time() 386 if self._authinfo is not None: 387 url = url_in.replace('//', '//'+self._authinfo) 388 else: 389 url = url_in 390 try: 391 server = ServerProxy(url) 392 method = getattr(server, method_name) 393 if params is not None: 394 response = method(*params) 395 else: 396 response = method() 397 except: 398 etype, value, tback = sys.exc_info() 399 t_stop = time.time() 400 t_delta = t_stop - t_start 401 self.total_time += t_delta 402 self.step_success = False 403 self.test_status = 'Error' 404 self.logd(' Failed in %.3fs' % t_delta) 405 self._log_xmlrpc_response(url_in, method_name, description, 406 response, t_start, t_stop, -1) 407 if etype is SocketError: 408 raise SocketError("Can't access %s." % url) 409 raise 410 t_stop = time.time() 411 t_delta = t_stop - t_start 412 self.total_time += t_delta 413 self.total_xmlrpc += 1 414 self.logd(' Done in %.3fs' % t_delta) 415 self._log_xmlrpc_response(url_in, method_name, description, response, 416 t_start, t_stop, 200) 417 self.sleep() 418 return response

419

420 - def xmlrpc_call(self, url_in, method_name, params=None, description=None):

421 """BBB of xmlrpc, this method will be removed for 1.6.0.""" 422 warn('Since 1.4.0 the method "xmlrpc_call" is renamed into "xmlrpc".', 423 DeprecationWarning, stacklevel=2) 424 return self.xmlrpc(url_in, method_name, params, description)

425

426 - def waitUntilAvailable(self, url, time_out=20, sleep_time=2):

427 """Wait until url is available. 428 429 Try a get on url every sleep_time until server is reached or 430 time is out.""" 431 time_start = time.time() 432 while(True): 433 try: 434 self._browser.fetch(url, None, 435 ok_codes=[200, 301, 302, 303, 307], 436 key_file=self._keyfile_path, 437 cert_file=self._certfile_path) 438 except SocketError: 439 if time.time() - time_start > time_out: 440 self.fail('Time out service %s not available after %ss' % 441 (url, time_out)) 442 else: 443 return 444 time.sleep(sleep_time)

445

446 - def setBasicAuth(self, login, password):

447 """Set http basic authentication.""" 448 self._browser.setBasicAuth(login, password) 449 self._authinfo = '%s:%s@' % (login, password)

450

451 - def clearBasicAuth(self):

452 """Remove basic authentication.""" 453 self._browser.clearBasicAuth() 454 self._authinfo = None

455

456 - def addHeader(self, key, value):

457 """Add an http header.""" 458 self._browser.extra_headers.append((key, value))

459

460 - def setHeader(self, key, value):

461 """Add or override an http header. 462 463 If value is None, the key is removed.""" 464 headers = self._browser.extra_headers 465 for i, (k, v) in enumerate(headers): 466 if k == key: 467 if value is not None: 468 headers[i] = (key, value) 469 else: 470 del headers[i] 471 break 472 else: 473 if value is not None: 474 headers.append((key, value))

475

476 - def delHeader(self, key):

477 """Remove an http header key.""" 478 self.setHeader(key, None)

479

480 - def clearHeaders(self):

481 """Remove all http headers set by addHeader or setUserAgent. 482 483 Note that the Referer is also removed.""" 484 self._browser.extra_headers = []

485

486 - def debugHeaders(self, debug_headers=True):

487 """Print request headers.""" 488 self._browser.debug_headers = debug_headers

489

490 - def setUserAgent(self, agent):

491 """Set User-Agent http header for the next requests. 492 493 If agent is None, the user agent header is removed.""" 494 self.setHeader('User-Agent', agent)

495

496 - def sleep(self):

497 """Sleeps a random amount of time. 498 499 Between the predefined sleep_time_min and sleep_time_max values. 500 """ 501 if self._pause: 502 raw_input("Press ENTER to continue ") 503 return 504 s_min = self.sleep_time_min 505 s_max = self.sleep_time_max 506 if s_max != s_min: 507 s_val = s_min + abs(s_max - s_min) * random() 508 else: 509 s_val = s_min 510 # we should always sleep something 511 thread_sleep(s_val)

512

513 - def setKeyAndCertificateFile(self, keyfile_path, certfile_path):

514 """Set the paths to a key file and a certificate file that will be 515 used by a https (ssl/tls) connection when calling the post or get 516 methods. 517 518 keyfile_path : path to a PEM formatted file that contains your 519 private key. 520 certfile_path : path to a PEM formatted certificate chain file. 521 """ 522 self._keyfile_path = keyfile_path 523 self._certfile_path = certfile_path

524

525 - def clearKeyAndCertificateFile(self):

526 """Clear any key file or certificate file paths set by calls to 527 setKeyAndCertificateFile. 528 """ 529 self._keyfile_path = None 530 self._certfile_path = None

531 532 533 #------------------------------------------------------------ 534 # Assertion helpers 535 #

536 - def getLastUrl(self):

537 """Return the last accessed url taking into account redirection.""" 538 response = self._response 539 if response is not None: 540 return response.url 541 return ''

542

543 - def getBody(self):

544 """Return the last response content.""" 545 response = self._response 546 if response is not None: 547 return response.body 548 return ''

549

550 - def listHref(self, url_pattern=None, content_pattern=None):

551 """Return a list of href anchor url present in the last html response. 552 553 Filtering href with url pattern or link text pattern.""" 554 response = self._response 555 ret = [] 556 if response is not None: 557 a_links = response.getDOM().getByName('a') 558 if a_links: 559 for link in a_links: 560 try: 561 ret.append((link.getContentString(), link.href)) 562 except AttributeError: 563 pass 564 if url_pattern is not None: 565 pat = re.compile(url_pattern) 566 ret = [link for link in ret 567 if pat.search(link[1]) is not None] 568 if content_pattern is not None: 569 pat = re.compile(content_pattern) 570 ret = [link for link in ret 571 if link[0] and (pat.search(link[0]) is not None)] 572 return [link[1] for link in ret]

573

574 - def getLastBaseUrl(self):

575 """Return the base href url.""" 576 response = self._response 577 if response is not None: 578 base = response.getDOM().getByName('base') 579 if base: 580 return base[0].href 581 return ''

582 583 584 #------------------------------------------------------------ 585 # configuration file utils 586 #

587 - def conf_get(self, section, key, default=_marker, quiet=False):

588 """Return an entry from the options or configuration file.""" 589 # check for a command line options 590 opt_key = '%s_%s' % (section, key) 591 opt_val = getattr(self.options, opt_key, None) 592 if opt_val: 593 #print('[%s] %s = %s from options.' % (section, key, opt_val)) 594 return opt_val 595 # check for the configuration file if opt val is None 596 # or nul 597 try: 598 val = self._config.get(section, key) 599 except (NoSectionError, NoOptionError): 600 if not quiet: 601 self.logi('[%s] %s not found' % (section, key)) 602 if default is _marker: 603 raise 604 val = default 605 #print('[%s] %s = %s from config.' % (section, key, val)) 606 return val

607

608 - def conf_getInt(self, section, key, default=_marker, quiet=False):

609 """Return an integer from the configuration file.""" 610 return int(self.conf_get(section, key, default, quiet))

611

612 - def conf_getFloat(self, section, key, default=_marker, quiet=False):

613 """Return a float from the configuration file.""" 614 return float(self.conf_get(section, key, default, quiet))

615

616 - def conf_getList(self, section, key, default=_marker, quiet=False, 617 separator=None):

618 """Return a list from the configuration file.""" 619 value = self.conf_get(section, key, default, quiet) 620 if value is default: 621 return value 622 if separator is None: 623 separator = ':' 624 if value.count(separator): 625 return value.split(separator) 626 return [value]

627 628 629 630 #------------------------------------------------------------ 631 # Extend unittest.TestCase to provide bench cycle hook 632 #

633 - def setUpCycle(self):

634 """Called on bench mode before a cycle start.""" 635 pass

636

637 - def tearDownCycle(self):

638 """Called after a cycle in bench mode.""" 639 pass

640 641 642 643 #------------------------------------------------------------ 644 # logging 645 #

646 - def logd(self, message):

647 """Debug log.""" 648 self.logger.debug(self.meta_method_name +': ' +message)

649

650 - def logdd(self, message):

651 """Verbose Debug log.""" 652 if self.debug_level >= 2: 653 self.logger.debug(self.meta_method_name +': ' +message)

654

655 - def logi(self, message):

656 """Info log.""" 657 if hasattr(self, 'logger'): 658 self.logger.info(self.meta_method_name+': '+message) 659 else: 660 print self.meta_method_name+': '+message

661

662 - def _logr(self, message, force=False):

663 """Log a result.""" 664 if force or not self.in_bench_mode or recording(): 665 self.logger_result.info(message)

666

667 - def _open_result_log(self, **kw):

668 """Open the result log.""" 669 xml = ['<funkload version="%s" time="%s">' % ( 670 get_version(), datetime.now().isoformat())] 671 for key, value in kw.items(): 672 xml.append('<config key="%s" value=%s />' % ( 673 key, quoteattr(str(value)))) 674 self._logr('\n'.join(xml), force=True)

675

676 - def _close_result_log(self):

677 """Close the result log.""" 678 self._logr('</funkload>', force=True)

679

680 - def _log_response_error(self, url, rtype, description, time_start, 681 time_stop):

682 """Log a response that raise an unexpected exception.""" 683 self.total_responses += 1 684 self.page_responses += 1 685 info = {} 686 info['cycle'] = self.cycle 687 info['cvus'] = self.cvus 688 info['thread_id'] = self.thread_id 689 info['suite_name'] = self.suite_name 690 info['test_name'] = self.test_name 691 info['step'] = self.steps 692 info['number'] = self.page_responses 693 info['type'] = rtype 694 info['url'] = quoteattr(url) 695 info['code'] = -1 696 info['description'] = description and quoteattr(description) or '""' 697 info['time_start'] = time_start 698 info['duration'] = time_stop - time_start 699 info['result'] = 'Error' 700 info['traceback'] = quoteattr(' '.join( 701 traceback.format_exception(*sys.exc_info()))) 702 message = '''<response cycle="%(cycle).3i" cvus="%(cvus).3i" thread="%(thread_id).3i" suite="%(suite_name)s" name="%(test_name)s" step="%(step).3i" number="%(number).3i" type="%(type)s" result="%(result)s" url=%(url)s code="%(code)s" description=%(description)s time="%(time_start)s" duration="%(duration)s" traceback=%(traceback)s />''' % info 703 self._logr(message)

704

705 - def _log_response(self, response, rtype, description, time_start, 706 time_stop, log_body=False):

707 """Log a response.""" 708 self.total_responses += 1 709 self.page_responses += 1 710 info = {} 711 info['cycle'] = self.cycle 712 info['cvus'] = self.cvus 713 info['thread_id'] = self.thread_id 714 info['suite_name'] = self.suite_name 715 info['test_name'] = self.test_name 716 info['step'] = self.steps 717 info['number'] = self.page_responses 718 info['type'] = rtype 719 info['url'] = quoteattr(response.url) 720 info['code'] = response.code 721 info['description'] = description and quoteattr(description) or '""' 722 info['time_start'] = time_start 723 info['duration'] = time_stop - time_start 724 info['result'] = self.step_success and 'Successful' or 'Failure' 725 response_start = '''<response cycle="%(cycle).3i" cvus="%(cvus).3i" thread="%(thread_id).3i" suite="%(suite_name)s" name="%(test_name)s" step="%(step).3i" number="%(number).3i" type="%(type)s" result="%(result)s" url=%(url)s code="%(code)s" description=%(description)s time="%(time_start)s" duration="%(duration)s"''' % info 726 727 if not log_body: 728 message = response_start + ' />' 729 else: 730 response_start = response_start + '>\n <headers>' 731 header_xml = [] 732 for key, value in response.headers.items(): 733 header_xml.append(' <header name="%s" value=%s />' % ( 734 key, quoteattr(value))) 735 headers = '\n'.join(header_xml) + '\n </headers>' 736 message = '\n'.join([ 737 response_start, 738 headers, 739 ' <body><![CDATA[\n%s\n]]>\n </body>' % response.body, 740 '</response>']) 741 self._logr(message)

742

743 - def _log_xmlrpc_response(self, url, method, description, response, 744 time_start, time_stop, code):

745 """Log a response.""" 746 self.total_responses += 1 747 self.page_responses += 1 748 info = {} 749 info['cycle'] = self.cycle 750 info['cvus'] = self.cvus 751 info['thread_id'] = self.thread_id 752 info['suite_name'] = self.suite_name 753 info['test_name'] = self.test_name 754 info['step'] = self.steps 755 info['number'] = self.page_responses 756 info['type'] = 'xmlrpc' 757 info['url'] = quoteattr(url + '#' + method) 758 info['code'] = code 759 info['description'] = description and quoteattr(description) or '""' 760 info['time_start'] = time_start 761 info['duration'] = time_stop - time_start 762 info['result'] = self.step_success and 'Successful' or 'Failure' 763 message = '''<response cycle="%(cycle).3i" cvus="%(cvus).3i" thread="%(thread_id).3i" suite="%(suite_name)s" name="%(test_name)s" step="%(step).3i" number="%(number).3i" type="%(type)s" result="%(result)s" url=%(url)s code="%(code)s" description=%(description)s time="%(time_start)s" duration="%(duration)s" />"''' % info 764 self._logr(message)

765

766 - def _log_result(self, time_start, time_stop):

767 """Log the test result.""" 768 info = {} 769 info['cycle'] = self.cycle 770 info['cvus'] = self.cvus 771 info['thread_id'] = self.thread_id 772 info['suite_name'] = self.suite_name 773 info['test_name'] = self.test_name 774 info['steps'] = self.steps 775 info['time_start'] = time_start 776 info['duration'] = time_stop - time_start 777 info['connection_duration'] = self.total_time 778 info['requests'] = self.total_responses 779 info['pages'] = self.total_pages 780 info['xmlrpc'] = self.total_xmlrpc 781 info['redirects'] = self.total_redirects 782 info['images'] = self.total_images 783 info['links'] = self.total_links 784 info['result'] = self.test_status 785 if self.test_status != 'Successful': 786 info['traceback'] = 'traceback=' + quoteattr(' '.join( 787 traceback.format_exception(*sys.exc_info()))) + ' ' 788 else: 789 info['traceback'] = '' 790 text = '''<testResult cycle="%(cycle).3i" cvus="%(cvus).3i" thread="%(thread_id).3i" suite="%(suite_name)s" name="%(test_name)s" time="%(time_start)s" result="%(result)s" steps="%(steps)s" duration="%(duration)s" connection_duration="%(connection_duration)s" requests="%(requests)s" pages="%(pages)s" xmlrpc="%(xmlrpc)s" redirects="%(redirects)s" images="%(images)s" links="%(links)s" %(traceback)s/>''' % info 791 self._logr(text)

792

793 - def _dump_content(self, response):

794 """Dump the html content in a file. 795 796 Use firefox to render the content if we are in rt viewing mode.""" 797 dump_dir = self._dump_dir 798 if dump_dir is None: 799 return 800 if getattr(response, 'code', 301) in [301, 302, 303, 307]: 801 return 802 if not response.body: 803 return 804 if not os.access(dump_dir, os.W_OK): 805 os.mkdir(dump_dir, 0775) 806 content_type = response.headers.get('content-type') 807 if content_type == 'text/xml': 808 ext = '.xml' 809 else: 810 ext = os.path.splitext(response.url)[1] 811 if not ext.startswith('.') or len(ext) > 4: 812 ext = '.html' 813 file_path = os.path.abspath( 814 os.path.join(dump_dir, '%3.3i%s' % (self.steps, ext))) 815 f = open(file_path, 'w') 816 f.write(response.body) 817 f.close() 818 if self._viewing: 819 cmd = 'firefox -remote "openfile(file://%s,new-tab)"' % file_path 820 ret = os.system(cmd) 821 if ret != 0: 822 self.logi('Failed to remote control firefox: %s' % cmd) 823 self._viewing = False

824 825 826 #------------------------------------------------------------ 827 # Overriding unittest.TestCase 828 #

829 - def __call__(self, result=None):

830 """Run the test method. 831 832 Override to log test result.""" 833 t_start = time.time() 834 if result is None: 835 result = self.defaultTestResult() 836 result.startTest(self) 837 if sys.version_info >= (2, 5): 838 testMethod = getattr(self, self._testMethodName) 839 else: 840 testMethod = getattr(self, self._TestCase__testMethodName) 841 try: 842 ok = False 843 try: 844 self.logd('Starting -----------------------------------\n\t%s' 845 % self.conf_get(self.meta_method_name, 'description', 846 '')) 847 self.setUp() 848 except KeyboardInterrupt: 849 raise 850 except: 851 if sys.version_info >= (2, 5): 852 result.addError(self, self._exc_info()) 853 else: 854 result.addError(self, self._TestCase__exc_info()) 855 self.test_status = 'Error' 856 self._log_result(t_start, time.time()) 857 return 858 try: 859 testMethod() 860 ok = True 861 except self.failureException: 862 if sys.version_info >= (2, 5): 863 result.addFailure(self, self._exc_info()) 864 else: 865 result.addFailure(self, self._TestCase__exc_info()) 866 self.test_status = 'Failure' 867 except KeyboardInterrupt: 868 raise 869 except: 870 if sys.version_info >= (2, 5): 871 result.addFailure(self, self._exc_info()) 872 else: 873 result.addError(self, self._TestCase__exc_info()) 874 self.test_status = 'Error' 875 try: 876 self.tearDown() 877 except KeyboardInterrupt: 878 raise 879 except: 880 if sys.version_info >= (2, 5): 881 result.addFailure(self, self._exc_info()) 882 else: 883 result.addError(self, self._TestCase__exc_info()) 884 self.test_status = 'Error' 885 ok = False 886 if ok: 887 result.addSuccess(self) 888 finally: 889 self._log_result(t_start, time.time()) 890 if not ok and self._stop_on_fail: 891 result.stop() 892 result.stopTest(self)

Source Code for Module funkload.FunkLoadTestCase