transfer_test_7.py 1.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940
  1. def testDownloadThenStream(self):
  2. bytes_http = object()
  3. http = object()
  4. download_stream = six.StringIO()
  5. download = transfer.Download.FromStream(download_stream,
  6. total_size=26)
  7. download.bytes_http = bytes_http
  8. base_url = 'https://part.one/'
  9. with mock.patch.object(http_wrapper, 'MakeRequest',
  10. autospec=True) as make_request:
  11. make_request.return_value = http_wrapper.Response(
  12. info={
  13. 'content-range': 'bytes 0-25/26',
  14. 'status': http_client.OK,
  15. },
  16. content=string.ascii_lowercase,
  17. request_url=base_url,
  18. )
  19. request = http_wrapper.Request(url='https://part.one/')
  20. download.InitializeDownload(request, http=http)
  21. self.assertEqual(1, make_request.call_count)
  22. received_request = make_request.call_args[0][1]
  23. self.assertEqual(base_url, received_request.url)
  24. self.assertRangeAndContentRangeCompatible(
  25. received_request, make_request.return_value)
  26. with mock.patch.object(http_wrapper, 'MakeRequest',
  27. autospec=True) as make_request:
  28. make_request.return_value = http_wrapper.Response(
  29. info={
  30. 'status': http_client.REQUESTED_RANGE_NOT_SATISFIABLE,
  31. },
  32. content='error',
  33. request_url=base_url,
  34. )
  35. download.StreamInChunks()
  36. self.assertEqual(1, make_request.call_count)
  37. received_request = make_request.call_args[0][1]
  38. self.assertEqual('bytes=26-', received_request.headers['range'])