test_safety.py 文件源码

def test_concurrent_requests(self):
        # the application must handle concurrent calls
        def make_requests():
            # use a blocking HTTP client (we're in another thread)
            http_client = httpclient.HTTPClient()
            url = self.get_url('/nested/')
            response = http_client.fetch(url)
            eq_(200, response.code)
            eq_('OK', response.body.decode('utf-8'))
            # freeing file descriptors
            http_client.close()

        # blocking call executed in different threads
        threads = [threading.Thread(target=make_requests) for _ in range(25)]
        for t in threads:
            t.daemon = True
            t.start()

        # wait for the execution; assuming this time as a timeout
        yield web.compat.sleep(0.5)

        # the trace is created
        traces = self.tracer.writer.pop_traces()
        eq_(25, len(traces))
        eq_(2, len(traces[0]))