Add functionality for content-disposition option in signed url.

src/gcp_storage_emulator/handlers/objects.py

@@ -471,6 +471,9 @@ def download(request, response, storage, *args, **kwargs):
  hash_header = "crc32c={},md5={}".format(obj["crc32c"], obj["md5Hash"])
  response[_HASH_HEADER] = hash_header
 
+ if 'response-content-disposition' in request.query:
+ response['Content-Disposition'] = request.query['response-content-disposition'][0]
+
  response.write_file(file, content_type=obj.get("contentType"))
  except NotFound:
  response.status = HTTPStatus.NOT_FOUND

tests/test_server.py

@@ -810,6 +810,65 @@ def test_signed_url_download(self):
  self.assertEqual(response.content, content)
  self.assertEqual(response.headers["content-type"], "text/mycustom")
 
+ def test_signed_url_download_with_content_disposition(self):
+ content = b"The quick brown fox jumps over the lazy dog"
+ bucket = self._client.create_bucket("testbucket")
+
+ blob = bucket.blob("signed-download")
+ blob.upload_from_string(content, content_type="text/mycustom")
+
+ requested_filename = 'requested_filename.cst2'
+ response_disposition = f'attachment; filename="{requested_filename}"'
+
+ url = blob.generate_signed_url(
+ api_access_endpoint="http://localhost:9023",
+ credentials=FakeSigningCredentials(),
+ version="v4",
+ expiration=datetime.timedelta(minutes=15),
+ response_disposition=response_disposition,
+ method="GET",
+ )
+
+ response = requests.get(url)
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.content, content)
+ self.assertEqual(response.headers['content-disposition'], f"{response_disposition}")
+ self.assertEqual(response.headers["content-type"], "text/mycustom")
+
+ def test_url_generation_for_browser(self):
+ self.skipTest('Used to test browser functionality with URL, not API.')
+ os.environ["STORAGE_EMULATOR_HOST"] = "http://localhost:8080"
+ content = b"The quick brown fox jumps over the lazy dog"
+
+ # Cloud Storage uses environment variables to configure API endpoints for
+ # file upload - which is read at module import time
+ from google.cloud import storage
+ http = requests.Session()
+
+ client = storage.Client(
+ project="[PROJECT]",
+ _http=http,
+ client_options={"api_endpoint": "http://localhost:8080"},
+ )
+
+ bucket = client.create_bucket("testbucket")
+
+ blob = bucket.blob("signed-download")
+ blob.upload_from_string(content, content_type="text/html")
+
+ requested_filename = 'requested_filename.cst2'
+ response_disposition = f'attachment; filename="{requested_filename}"'
+
+ url = blob.generate_signed_url(
+ api_access_endpoint="http://localhost:8080",
+ credentials=FakeSigningCredentials(),
+ version="v4",
+ expiration=datetime.timedelta(minutes=15),
+ response_disposition=response_disposition,
+ method="GET",
+ )
+ print(url)
+
  def test_signed_url_upload(self):
  bucket = self._client.create_bucket("testbucket")
  blob = bucket.blob("signed-upload")