test_resumable_downloads.py 文件源码

def test_broken_pipe_recovery(self):
        """
        Tests handling of a Broken Pipe (which interacts with an httplib bug)
        """
        exception = IOError(errno.EPIPE, "Broken pipe")
        harness = CallbackTestHarness(exception=exception)
        res_download_handler = ResumableDownloadHandler(num_retries=1)
        dst_fp = self.make_dst_fp()
        small_src_key_as_string, small_src_key = self.make_small_key()
        small_src_key.get_contents_to_file(
            dst_fp, cb=harness.call,
            res_download_handler=res_download_handler)
        # Ensure downloaded object has correct content.
        self.assertEqual(SMALL_KEY_SIZE,
                         get_cur_file_size(dst_fp))
        self.assertEqual(small_src_key_as_string,
                         small_src_key.get_contents_as_string())