Add functionality for content-disposition option in signed url. (#253)

src/gcp_storage_emulator/handlers/objects.py

@@ -510,6 +510,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")