test_downloader.py 文件源码

def test_dynamic_request_concurrency(self):
        self.driver = webdriver.PhantomJS()
        dh = DownloadHandler(self.spider, self.driver, self.driver_sem)
        n = 5
        pool = Pool(n)
        urls = []
        for i in range(n):
            urls.append(HTTPBIN_URL + '/delay/1')
        time1 = time.time()
        pool.map(dh.fetch, [Request(url, dynamic=True, wait=5) for url in urls])
        self.assertGreater(time.time() - time1, n)
        self.driver.close()