def createThreads(self, cycle, number_of_threads):
        """Creates number_of_threads threads and returns as a list.

        NOTE: This method is not thread safe. Thread safety must be
        handled by the caller."""
        threads = []
        i = 0
        for i in range(number_of_threads):
            thread_id = self.createThreadId()
            thread_signaller = ThreadSignaller()
            thread = LoopTestRunner(self.module_name, self.class_name,
                                    self.method_name, self.options,
                                    cycle, number_of_threads,
                                    thread_id, thread_signaller,
                                    self.sleep_time)
            trace(".")
            try:
                thread.start()
            except ThreadError:
                trace("\nERROR: Can not create more than %i threads, try a "
                      "smaller stack size using: 'ulimit -s 2048' "
                      "for example\n" % (i + 1))
                raise
            thread_data = ThreadData(thread, thread_id, thread_signaller)
            threads.append(thread_data)
            thread_sleep(self.startup_delay)
        trace(' done.\n')
        return threads

 def do_link(self, attributes):
     """Process link tag."""
     newattributes = [('rel', 'stylesheet'), ('type', 'text/css')]
     for name, value in attributes:
         if name == 'href':
             # construct full url
             url = urlparse.urljoin(self.base, value)
             # make sure it's syntactically valid
             if not valid_url.match(url):
                 continue
             # TODO: figure the re-write path
             # newattributes.append((name, path))
             if not self.session.css.has_key(url):
                 self.ftestcase.logdd('    link: %s ...' % url)
                 t_start = time.time()
                 self.session.css[url] = self.session.fetch(url)
                 t_stop = time.time()
                 self.ftestcase.logdd('     Done in %.3fs' %
                                      (t_stop - t_start))
                 self.session.history.append(('link', url))
                 self.ftestcase.total_time += (t_stop - t_start)
                 self.ftestcase.total_links += 1
                 self.ftestcase._log_response(self.session.css[url],
                                              'link', None, t_start, t_stop)
                 thread_sleep()      # give a chance to other threads
         else:
             newattributes.append((name, value))
     # Write the link tag to file (with revised paths)
     self.unknown_starttag('link', newattributes)

 def run(self):
     """Run a test in loop."""
     while (self.thread_signaller.running()):
         test_result = unittest.TestResult()
         self.test.clearContext()
         self.test(test_result)
         if test_result.wasSuccessful():
             if recording():
                 add_cycle_result('success')
                 if self.color:
                     trace(green_str('.'))
                 else:
                     trace('.')
         else:
             if len(test_result.errors):
                 if recording():
                     add_cycle_result('error')
                 if self.color:
                     trace(red_str('E'))
                 else:
                     trace('E')
             else:
                 if recording():
                     add_cycle_result('failure')
                 if self.color:
                     trace(red_str('F'))
                 else:
                     trace('F')
             if self.debug:
                 for (test, error) in test_result.errors:
                     trace("ERROR %s: %s" % (str(test), str(error)))
                 for (test, error) in test_result.failures:
                     trace("FAILURE %s: %s" % (str(test), str(error)))
         thread_sleep(self.sleep_time)

 def do_link(self, attributes):
     """Process link tag."""
     newattributes = [('rel', 'stylesheet'), ('type', 'text/css')]
     for name, value in attributes:
         if name == 'href':
             # construct full url
             url = urlparse.urljoin(self.base, value)
             # make sure it's syntactically valid
             if not valid_url.match(url):
                 continue
             # TODO: figure the re-write path
             # newattributes.append((name, path))
             if not self.session.css.has_key(url):
                 self.ftestcase.logdd('    link: %s ...' % url)
                 with self.ftestcase.record_response('link', url, None):
                     try:
                         self.session.css[url] = self.session.fetch(url)
                         self.session.history.append(('link', url))
                     except HTTPError as error:
                         if self.ftestcase._accept_invalid_links:
                             if not self.ftestcase.in_bench_mode:
                                 self.ftestcase.logd('  ' + str(error))
                         else:
                             raise
                 thread_sleep()      # give a chance to other threads
         else:
             newattributes.append((name, value))
     # Write the link tag to file (with revised paths)
     self.unknown_starttag('link', newattributes)

 def do_img(self, attributes):
     """Process img tag."""
     newattributes = []
     for name, value in attributes:
         if name == "src":
             # construct full url
             url = urlparse.urljoin(self.base, value)
             # make sure it's syntactically valid
             if not valid_url.match(url):
                 continue
             # TODO: figure the re-write path
             # newattributes.append((name, path))
             if not self.session.images.has_key(url):
                 self.ftestcase.logdd("    img: %s ..." % url)
                 t_start = time.time()
                 self.session.images[url] = self.session.fetch(url)
                 t_stop = time.time()
                 self.ftestcase.logdd("     Done in %.3fs" % (t_stop - t_start))
                 self.session.history.append(("image", url))
                 self.ftestcase.total_time += t_stop - t_start
                 self.ftestcase.total_images += 1
                 self.ftestcase._log_response(self.session.images[url], "image", None, t_start, t_stop)
                 thread_sleep()  # give a chance to other threads
         else:
             newattributes.append((name, value))
     # Write the img tag to file (with revised paths)
     self.unknown_starttag("img", newattributes)

    def run(self):
        """Run a test in loop."""
        while (self.thread_signaller.running()):
            test_result = unittest.TestResult()
            self.test.clearContext()
            self.test(test_result)
            feedback = {}

            if test_result.wasSuccessful():
                if recording():
                    feedback['count'] = add_cycle_result('success')

                if self.color:
                    trace(green_str('.'))
                else:
                    trace('.')

                feedback['result'] = 'success'
            else:
                if len(test_result.errors):
                    if recording():
                        feedback['count'] = add_cycle_result('error')

                    if self.color:
                        trace(red_str('E'))
                    else:
                        trace('E')

                    feedback['result'] = 'error'

                else:
                    if recording():
                        feedback['count'] = add_cycle_result('failure')

                    if self.color:
                        trace(red_str('F'))
                    else:
                        trace('F')

                    feedback['result'] = 'failure'

                if self.debug:
                    feedback['errors'] = test_result.errors
                    feedback['failures'] = test_result.failures

                    for (test, error) in test_result.errors:
                        trace("ERROR %s: %s" % (str(test), str(error)))
                    for (test, error) in test_result.failures:
                        trace("FAILURE %s: %s" % (str(test), str(error)))

            if self.feedback is not None:
                self.feedback.test_done(feedback)

            thread_sleep(self.sleep_time)

    def run(self):
        """Run a test in loop."""
        while self.thread_signaller.running():
            test_result = unittest.TestResult()
            self.test.clearContext()
            self.test(test_result)
            feedback = {}

            if test_result.wasSuccessful():
                if recording():
                    feedback["count"] = add_cycle_result("success")

                if self.color:
                    trace(green_str("."))
                else:
                    trace(".")

                feedback["result"] = "success"
            else:
                if len(test_result.errors):
                    if recording():
                        feedback["count"] = add_cycle_result("error")

                    if self.color:
                        trace(red_str("E"))
                    else:
                        trace("E")

                    feedback["result"] = "error"

                else:
                    if recording():
                        feedback["count"] = add_cycle_result("failure")

                    if self.color:
                        trace(red_str("F"))
                    else:
                        trace("F")

                    feedback["result"] = "failure"

                if self.debug:
                    feedback["errors"] = test_result.errors
                    feedback["failures"] = test_result.failures

                    for (test, error) in test_result.errors:
                        trace("ERROR %s: %s" % (str(test), str(error)))
                    for (test, error) in test_result.failures:
                        trace("FAILURE %s: %s" % (str(test), str(error)))

            if self.feedback is not None:
                self.feedback.test_done(feedback)

            thread_sleep(self.sleep_time)

    def sleep(self):
        """Sleeps a random amount of time.

        Between the predefined sleep_time_min and sleep_time_max values.
        """
        if self._pause:
            raw_input("Press ENTER to continue ")
            return
        s_min = self.sleep_time_min
        s_max = self.sleep_time_max
        if s_max != s_min:
            s_val = s_min + abs(s_max - s_min) * random()
        else:
            s_val = s_min
        # we should always sleep something
        thread_sleep(s_val)

    def _browse(self, url_in, params_in=None,
                description=None, ok_codes=None,
                method='post',
                follow_redirect=True, load_auto_links=True,
                sleep=True):
        """Simulate a browser handle redirects, load/cache css and images."""
        self._response = None
        # Loop mode
        if self._loop_mode:
            if self.steps == self._loop_steps[0]:
                self._loop_recording = True
                self.logi('Loop mode start recording')
            if self._loop_recording:
                self._loop_records.append((url_in, params_in, description,
                                           ok_codes, method, follow_redirect,
                                           load_auto_links, False))
        # ok codes
        if ok_codes is None:
            ok_codes = self.ok_codes
        if type(params_in) is DictType:
            params_in = params_in.items()
        params = []
        if params_in:
            if isinstance(params_in, Data):
                params = params_in
            else:
                for key, value in params_in:
                    if type(value) is DictType:
                        for val, selected in value.items():
                            if selected:
                                params.append((key, val))
                    elif type(value) in (ListType, TupleType):
                        for val in value:
                            params.append((key, val))
                    else:
                        params.append((key, value))

        if method == 'get' and params:
            url = url_in + '?' + urlencode(params)
        else:
            url = url_in
        if method == 'get':
            params = None

        if method == 'get':
            if not self.in_bench_mode:
                self.logd('GET: %s\n\tPage %i: %s ...' % (url, self.steps,
                                                          description or ''))
        else:
            url = url_in
            if not self.in_bench_mode:
                self.logd('%s: %s %s\n\tPage %i: %s ...' % (
                        method.upper(), url, str(params),
                        self.steps, description or ''))
            # Fetching
        response = self._connect(url, params, ok_codes, method, description)

        # Check redirection
        if follow_redirect and response.code in (301, 302, 303, 307):
            max_redirect_count = 10
            thread_sleep()              # give a chance to other threads
            while response.code in (301, 302, 303, 307) and max_redirect_count:
                # Figure the location - which may be relative
                newurl = response.headers['Location']
                url = urljoin(url_in, newurl)
                # Save the current url as the base for future redirects
                url_in = url
                self.logd(' Load redirect link: %s' % url)
                # Use the appropriate method for redirection
                if response.code in (302, 303):
                    method = 'get'
                if response.code == 303:
                    # 303 is HTTP/1.1, make sure the connection 
                    # is not in keep alive mode
                    self.setHeader('Connection', 'close')
                response = self._connect(url, None, ok_codes, rtype=method,
                                         description=None, redirect=True)
                max_redirect_count -= 1
            if not max_redirect_count:
                self.logd(' WARNING Too many redirects give up.')

        # Load auto links (css and images)
        response.is_html = is_html(response.body)
        if load_auto_links and response.is_html and not self._simple_fetch:
            self.logd(' Load css and images...')
            page = response.body
            t_start = time.time()
            c_start = self.total_time
            try:
                # pageImages is patched to call _log_response on all links
                self._browser.pageImages(url, page, self)
            except HTTPError, error:
                if self._accept_invalid_links:
                    if not self.in_bench_mode:
                        self.logd('  ' + str(error))
                else:
                    t_stop = time.time()
                    t_delta = t_stop - t_start
                    self.step_success = False
                    self.test_status = 'Failure'
#.........这里部分代码省略.........

    def _browse(self, url_in, params_in=None,
                description=None, ok_codes=None,
                method='post',
                follow_redirect=True, load_auto_links=True,
                sleep=True):
        """Simulate a browser handle redirects, load/cache css and images."""
        self._response = None
        # Loop mode
        if self._loop_mode:
            if self.steps == self._loop_steps[0]:
                self._loop_recording = True
                self.logi('Loop mode start recording')
            if self._loop_recording:
                self._loop_records.append((url_in, params_in, description,
                                           ok_codes, method, follow_redirect,
                                           load_auto_links, False))
        # ok codes
        if ok_codes is None:
            ok_codes = self.ok_codes
        if type(params_in) is DictType:
            params_in = params_in.items()
        params = []
        if params_in:
            if isinstance(params_in, Data):
                params = params_in
            else:
                for key, value in params_in:
                    if type(value) is DictType:
                        for val, selected in value.items():
                            if selected:
                                params.append((key, val))
                    elif type(value) in (ListType, TupleType):
                        for val in value:
                            params.append((key, val))
                    else:
                        params.append((key, value))

        if method == 'get' and params:
            url = url_in + '?' + urlencode(params)
        else:
            url = url_in
        if method == 'get':
            params = None

        if method == 'get':
            if not self.in_bench_mode:
                self.logd('GET: %s\n\tPage %i: %s ...' % (url, self.steps,
                                                          description or ''))
        else:
            url = url_in
            if not self.in_bench_mode:
                self.logd('%s: %s %s\n\tPage %i: %s ...' % (
                        method.upper(), url, str(params),
                        self.steps, description or ''))
            # Fetching
        response = self._connect(url, params, ok_codes, method, description)

        # Check redirection
        if follow_redirect and response.code in (301, 302, 303, 307):
            max_redirect_count = 10
            thread_sleep()              # give a chance to other threads
            while response.code in (301, 302, 303, 307) and max_redirect_count:
                # Figure the location - which may be relative
                newurl = response.headers['Location']
                url = urljoin(url_in, newurl)
                # Save the current url as the base for future redirects
                url_in = url
                self.logd(' Load redirect link: %s' % url)
                # Use the appropriate method for redirection
                if response.code in (302, 303):
                    method = 'get'
                if response.code == 303:
                    # 303 is HTTP/1.1, make sure the connection 
                    # is not in keep alive mode
                    self.setHeader('Connection', 'close')
                response = self._connect(url, None, ok_codes, rtype=method,
                                         description=None, redirect=True)
                max_redirect_count -= 1
            if not max_redirect_count:
                self.logd(' WARNING Too many redirects give up.')

        # Load auto links (css and images)
        response.is_html = is_html(response.body)
        if load_auto_links and response.is_html and not self._simple_fetch:
            self.logd(' Load css and images...')
            page = response.body
            # pageImages is patched to call self.record on all links
            self._browser.pageImages(url, page, self)
        if sleep:
            self.sleep()
        self._response = response

        # Loop mode
        if self._loop_mode and self.steps == self._loop_steps[-1]:
            self._loop_recording = False
            self.logi('Loop mode end recording.')
            t_start = self.total_time
            count = 0
            for i in range(self._loop_number):
                self.logi('Loop mode replay %i' % i)
#.........这里部分代码省略.........