test_downloader.py 文件源码

def test_concurrency_with_delayed_url(self):
        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')
        time_start = time.time()
        pool.map(dh.fetch, [Request(url) for url in urls])
        time_total = time.time() - time_start
        self.assertLess(time_total, n)