feat: add XML RPCs to emulator (#40)

emulator.py

@@ -38,6 +38,44 @@ def index():
     return "OK"
 
 
+def xml_put_object(bucket_name, object_name):
+    db.insert_test_bucket(None)
+    bucket = db.get_bucket_without_generation(bucket_name, None).metadata
+    blob, fake_request = gcs_type.object.Object.init_xml(
+        flask.request, bucket, object_name
+    )
+    db.insert_object(fake_request, bucket_name, blob, None)
+    response = flask.make_response("")
+    response.headers["x-goog-hash"] = fake_request.headers.get("x-goog-hash")
+    return response
+
+
+def xml_get_object(bucket_name, object_name):
+    fake_request = testbench.common.FakeRequest.init_xml(flask.request)
+    blob = db.get_object(fake_request, bucket_name, object_name, False, None)
+    return blob.rest_media(fake_request)
+
+
+@root.route("/<path:object_name>", subdomain="<bucket_name>")
+def root_get_object(bucket_name, object_name):
+    return xml_get_object(bucket_name, object_name)
+
+
+@root.route("/<bucket_name>/<path:object_name>", subdomain="")
+def root_get_object_with_bucket(bucket_name, object_name):
+    return xml_get_object(bucket_name, object_name)
+
+
+@root.route("/<path:object_name>", subdomain="<bucket_name>", methods=["PUT"])
+def root_put_object(bucket_name, object_name):
+    return xml_put_object(bucket_name, object_name)
+
+
+@root.route("/<bucket_name>/<path:object_name>", subdomain="", methods=["PUT"])
+def root_put_object_with_bucket(bucket_name, object_name):
+    return xml_put_object(bucket_name, object_name)
+
+
 # === WSGI APP TO HANDLE JSON API === #
 GCS_HANDLER_PATH = "/storage/v1"
 gcs = flask.Flask(__name__)

tests/test_emulator_object_xml.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+#
+# Copyright 2021 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Unit test for emulator."""
+
+import json
+import os
+import unittest
+from werkzeug.test import create_environ
+
+import emulator
+import testbench
+
+
+class TestEmulator(unittest.TestCase):
+    def setUp(self):
+        emulator.db = testbench.database.Database.init()
+        emulator.server.config["PREFERRED_URL_SCHEME"] = "https"
+        emulator.server.config["SERVER_NAME"] = "storage.googleapis.com"
+        emulator.root.config["PREFERRED_URL_SCHEME"] = "https"
+        emulator.root.config["SERVER_NAME"] = "storage.googleapis.com"
+        self.client = emulator.server.test_client(allow_subdomain_redirects=True)
+        # Avoid magic buckets in the test
+        os.environ.pop("GOOGLE_CLOUD_CPP_STORAGE_TEST_BUCKET_NAME", None)
+
+    def test_object_xml_put_get_with_subdomain(self):
+        response = self.client.post(
+            "/storage/v1/b", data=json.dumps({"name": "bucket-name"})
+        )
+        self.assertEqual(response.status_code, 200)
+
+        response = self.client.put(
+            "/fox.txt",
+            base_url="https://bucket-name.storage.googleapis.com",
+            content_type="text/plain",
+            data="The quick brown fox jumps over the lazy dog",
+        )
+        self.assertEqual(response.status_code, 200, msg=response.data)
+
+        response = self.client.get(
+            "/fox.txt", base_url="https://bucket-name.storage.googleapis.com"
+        )
+        self.assertEqual(response.status_code, 200, msg=response.data)
+        self.assertEqual(response.data, b"The quick brown fox jumps over the lazy dog")
+
+    def test_object_xml_put_get_with_bucket(self):
+        response = self.client.post(
+            "/storage/v1/b", data=json.dumps({"name": "bucket-name"})
+        )
+        self.assertEqual(response.status_code, 200)
+
+        response = self.client.put(
+            "/bucket-name/fox.txt",
+            content_type="text/plain",
+            data="The quick brown fox jumps over the lazy dog",
+        )
+        self.assertEqual(response.status_code, 200)
+
+        response = self.client.get("/bucket-name/fox.txt")
+        self.assertEqual(response.status_code, 200, msg=response.data)
+        self.assertEqual(response.data, b"The quick brown fox jumps over the lazy dog")
+
+
+if __name__ == "__main__":
+    unittest.main()