test_resumable_uploads.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_upload_handler = ResumableUploadHandler(num_retries=1)
        small_src_file_as_string, small_src_file = self.make_small_file()
        small_src_file.seek(0)
        dst_key = self._MakeKey(set_contents=False)
        dst_key.set_contents_from_file(
            small_src_file, cb=harness.call,
            res_upload_handler=res_upload_handler)
        # Ensure uploaded object has correct content.
        self.assertEqual(SMALL_KEY_SIZE, dst_key.size)
        self.assertEqual(small_src_file_as_string,
                         dst_key.get_contents_as_string())