REST: Keep vended storage credentials for staged tables (#4329)

pyiceberg/catalog/rest/__init__.py

diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py
index bca9bfe..d9f0ced 100644
--- a/pyiceberg/catalog/rest/__init__.py
+++ b/pyiceberg/catalog/rest/__init__.py
@@ -1198,9 +1198,8 @@
 
     def _response_to_staged_table(self, identifier_tuple: tuple[str, ...], table_response: TableResponse) -> StagedTable:
         # Per Iceberg spec: storage-credentials take precedence over config
-        credential_config = self._resolve_storage_credentials(
-            table_response.storage_credentials, table_response.metadata_location
-        )
+        credential_location = table_response.metadata_location or f"{table_response.metadata.location.rstrip('/')}/"
+        credential_config = self._resolve_storage_credentials(table_response.storage_credentials, credential_location)
         return StagedTable(
             identifier=identifier_tuple,
             metadata_location=table_response.metadata_location,  # type: ignore

tests/catalog/test_rest.py

diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py
index a6ef7a7..901b66e 100644
--- a/tests/catalog/test_rest.py
+++ b/tests/catalog/test_rest.py
@@ -1798,6 +1798,33 @@
     assert actual_response == expected
 
 
+def test_create_staged_table_with_storage_credentials(
+    rest_mock: Mocker,
+    table_schema_simple: Schema,
+    example_table_metadata_no_snapshot_v1: dict[str, Any],
+) -> None:
+    table_location = "gs://bucket/ns/t"
+    rest_mock.post(
+        f"{TEST_URI}v1/namespaces/fokko/tables",
+        json={
+            "metadata": {**example_table_metadata_no_snapshot_v1, "location": table_location},
+            "config": {},
+            "storage-credentials": [
+                {
+                    "prefix": f"{table_location}/",
+                    "config": {"gcs.oauth2.token": "vended"},
+                }
+            ],
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    txn = catalog.create_table_transaction(identifier=("fokko", "fokko2"), schema=table_schema_simple)
+
+    assert txn._table.io.properties["gcs.oauth2.token"] == "vended"
+
+
 def test_create_table_409(rest_mock: Mocker, table_schema_simple: Schema) -> None:
     rest_mock.post(
         f"{TEST_URI}v1/namespaces/fokko/tables",