feat(testbench): better emulation for XML requests

google/cloud/storage/testbench/testbench.py

@@ -37,7 +37,7 @@ def httpbin_error(error):
     return error.as_response()
 
 
-root = flask.Flask(__name__)
+root = flask.Flask(__name__, subdomain_matching=True)
 root.debug = True
 
 
@@ -47,6 +47,31 @@ def index():
     return "OK"
 
 
+@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(flask.request.host_url, 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(flask.request.host_url, bucket_name, object_name)
+
+
+@root.errorhandler(error_response.ErrorResponse)
+def root_error(error):
+    return error.as_response()
+
+
 # Define the WSGI application to handle bucket requests.
 GCS_HANDLER_PATH = "/storage/v1"
 gcs = flask.Flask(__name__)
@@ -865,39 +890,24 @@ def xmlapi_error(error):
     return error.as_response()
 
 
-@xmlapi.route("/<bucket_name>/<object_name>")
+@xmlapi.route("/<bucket_name>/<path:object_name>")
 def xmlapi_get_object(bucket_name, object_name):
     """Implement the 'Objects: insert' API.  Insert a new GCS Object."""
-    object_path, blob = testbench_utils.lookup_object(bucket_name, object_name)
-    if flask.request.args.get("acl") is not None:
-        raise error_response.ErrorResponse(
-            "ACL query not supported in XML API", status_code=500
-        )
-    if flask.request.args.get("encryption") is not None:
-        raise error_response.ErrorResponse(
-            "Encryption query not supported in XML API", status_code=500
-        )
-    generation_match = flask.request.headers.get("if-generation-match")
-    metageneration_match = flask.request.headers.get("if-metageneration-match")
-    blob.check_preconditions_by_value(
-        generation_match, None, metageneration_match, None
-    )
-    revision = blob.get_revision(flask.request)
-    return objects_get_common(bucket_name, object_name, revision)
+    return xml_get_object(bucket_name, object_name)
 
 
-@xmlapi.route("/<bucket_name>/<object_name>", methods=["PUT"])
+@xmlapi.route("/<bucket_name>/<path:object_name>", methods=["PUT"])
 def xmlapi_put_object(bucket_name, object_name):
     """Inserts a new GCS Object.
 
     Implement the PUT request in the XML API.
     """
-    gcs_url = flask.url_for(
-        "xmlapi_put_object",
-        bucket_name=bucket_name,
-        object_name=object_name,
-        _external=True,
-    ).replace("/xmlapi/", "/")
+    gcs_url = flask.request.host_url.replace("/xmlapi/", "/")
+    return xml_put_object(gcs_url, bucket_name, object_name)
+
+
+def xml_put_object(gcs_url, bucket_name, object_name):
+    """Implement PUT for the XML API."""
     insert_magic_bucket(gcs_url)
     object_path, blob = testbench_utils.get_object(
         bucket_name, object_name, gcs_object.GcsObject(bucket_name, object_name)
@@ -914,6 +924,26 @@ def xmlapi_put_object(bucket_name, object_name):
     return response
 
 
+def xml_get_object(bucket_name, object_name):
+    """Implement the 'Objects: insert' API.  Insert a new GCS Object."""
+    object_path, blob = testbench_utils.lookup_object(bucket_name, object_name)
+    if flask.request.args.get("acl") is not None:
+        raise error_response.ErrorResponse(
+            "ACL query not supported in XML API", status_code=500
+        )
+    if flask.request.args.get("encryption") is not None:
+        raise error_response.ErrorResponse(
+            "Encryption query not supported in XML API", status_code=500
+        )
+    generation_match = flask.request.headers.get("if-generation-match")
+    metageneration_match = flask.request.headers.get("if-metageneration-match")
+    blob.check_preconditions_by_value(
+        generation_match, None, metageneration_match, None
+    )
+    revision = blob.get_revision(flask.request)
+    return objects_get_common(bucket_name, object_name, revision)
+
+
 # Define the WSGI application to handle HMAC key requests
 (PROJECTS_HANDLER_PATH, projects_app) = gcs_project.get_projects_app()
 
@@ -939,7 +969,7 @@ def main():
     parser = argparse.ArgumentParser(
         description="A testbench for the Google Cloud C++ Client Library"
     )
-    parser.add_argument("--host", default="localhost", help="The listening port")
+    parser.add_argument("--host", default="localhost", help="The listening address")
     parser.add_argument("--port", help="The listening port")
     # By default we do not turn on the debugging. This typically runs inside a
     # Docker image, with a uid that has not entry in /etc/passwd, and the
@@ -949,6 +979,8 @@ def main():
     )
     arguments = parser.parse_args()
 
+    root.config.update(SERVER_NAME=arguments.host)
+
     # Compose the different WSGI applications.
     serving.run_simple(
         arguments.host,