Add support: start and end download parameters (#20)

* Add support: `start` and `end` download parameters

gcp_storage_emulator/handlers/objects.py

@@ -2,6 +2,7 @@
 import json
 import logging
 import math
+import re
 import secrets
 import string
 import time
@@ -342,6 +343,24 @@ def download(request, response, storage, *args, **kwargs):
  obj = storage.get_file_obj(
  request.params["bucket_name"], request.params["object_id"]
  )
+ range = request.get_header("range", None)
+ if range:
+ regex = r"^\s*bytes=(?P<start>[0-9]+)-(?P<end>[0-9]*)$"
+ pattern = re.compile(regex)
+ match = pattern.fullmatch(range)
+ if match:
+ end = orig_len = len(file)
+ m_dict = match.groupdict()
+ start = int(m_dict["start"])
+ if m_dict["end"]:
+ end = min(orig_len, int(m_dict["end"]) + 1)
+ file = file[start:end]
+ end -= 1
+ response["Content-Range"] = "bytes {}-{}/{}".format(
+ start, end, orig_len
+ )
+ response.status = HTTPStatus.PARTIAL_CONTENT
+
  response.write_file(file, content_type=obj.get("contentType"))
  except NotFound:
  response.status = HTTPStatus.NOT_FOUND

tests/test_server.py

@@ -268,6 +268,39 @@ def test_download_as_bytes(self):
  fetched_content = blob.download_as_bytes()
  self.assertEqual(fetched_content, content.encode("utf-8"))
 
+ def test_download_range_start(self):
+ content = b"123456789"
+ bucket = self._client.create_bucket("testbucket")
+
+ blob = bucket.blob("iexist")
+ blob.upload_from_string(content)
+
+ blob = bucket.get_blob("iexist")
+ fetched_content = blob.download_as_bytes(start=2)
+ self.assertEqual(fetched_content, b"3456789")
+
+ def test_download_range_end(self):
+ content = b"123456789"
+ bucket = self._client.create_bucket("testbucket")
+
+ blob = bucket.blob("iexist")
+ blob.upload_from_string(content)
+
+ blob = bucket.get_blob("iexist")
+ fetched_content = blob.download_as_bytes(end=4)
+ self.assertEqual(fetched_content, b"12345")
+
+ def test_download_range_start_end(self):
+ content = b"123456789"
+ bucket = self._client.create_bucket("testbucket")
+
+ blob = bucket.blob("iexist")
+ blob.upload_from_string(content)
+
+ blob = bucket.get_blob("iexist")
+ fetched_content = blob.download_as_bytes(start=2, end=4)
+ self.assertEqual(fetched_content, b"345")
+
  def test_set_content_encoding(self):
  content = "The quick brown fox jumps over the lazy dog\n"
  bucket = self._client.create_bucket("testbucket")