add test case

Author: plusplusjiajia

pyiceberg/catalog/rest/__init__.py

@@ -216,6 +216,7 @@ class IdentifierKind(Enum):
 SIGV4 = "rest.sigv4-enabled"
 SIGV4_REGION = "rest.signing-region"
 SIGV4_SERVICE = "rest.signing-name"
+EMPTY_BODY_SHA256: str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
 OAUTH2_SERVER_URI = "oauth2-server-uri"
 SNAPSHOT_LOADING_MODE = "snapshot-loading-mode"
 AUTH = "auth"
@@ -565,7 +566,7 @@ def add_headers(self, request: PreparedRequest, **kwargs: Any) -> None:  # pylin
                 # For empty bodies, explicitly set the content hash header to the SHA256 of an empty string
                 body = request.body
                 if body in (None, b"", ""):
-                    request.headers["x-amz-content-sha256"] = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+                    request.headers["x-amz-content-sha256"] = EMPTY_BODY_SHA256
 
                 aws_request = AWSRequest(
                     method=request.method, url=url, params=params, data=request.body, headers=dict(request.headers)

tests/catalog/test_rest.py

@@ -23,6 +23,7 @@
 
 import pytest
 from requests.exceptions import HTTPError
+from requests.adapters import HTTPAdapter
 from requests_mock import Mocker
 
 import pyiceberg
@@ -49,6 +50,7 @@
 from pyiceberg.transforms import IdentityTransform, TruncateTransform
 from pyiceberg.typedef import RecursiveDict
 from pyiceberg.utils.config import Config
+from requests import Request
 
 TEST_URI = "https://iceberg-test-catalog/"
 TEST_CREDENTIALS = "client:secret_with:colon"
@@ -451,6 +453,62 @@ def test_list_tables_200_sigv4(rest_mock: Mocker) -> None:
     assert rest_mock.called
 
 
+def test_sigv4_sign_request_without_body(rest_mock: Mocker) -> None:
+    existing_token = "existing_token"
+    from pyiceberg.catalog.rest import EMPTY_BODY_SHA256
+
+    catalog = RestCatalog(
+        "rest",
+        **{
+            "uri": TEST_URI,
+            "token": existing_token,
+            "rest.sigv4-enabled": "true",
+            "rest.signing-region": "us-west-2",
+            "client.access-key-id": "id",
+            "client.secret-access-key": "secret",
+        },
+    )
+
+    prepared = catalog._session.prepare_request(Request("GET", f"{TEST_URI}v1/config"))
+    adapter = catalog._session.adapters[catalog.uri]
+    adapter.add_headers(prepared)
+
+    assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256")
+    assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}"
+    assert prepared.headers["x-amz-content-sha256"] == EMPTY_BODY_SHA256
+
+
+def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None:
+    existing_token = "existing_token"
+    from pyiceberg.catalog.rest import EMPTY_BODY_SHA256
+
+    catalog = RestCatalog(
+        "rest",
+        **{
+            "uri": TEST_URI,
+            "token": existing_token,
+            "rest.sigv4-enabled": "true",
+            "rest.signing-region": "us-west-2",
+            "client.access-key-id": "id",
+            "client.secret-access-key": "secret",
+        },
+    )
+
+    prepared = catalog._session.prepare_request(
+        Request(
+            "POST",
+            f"{TEST_URI}v1/namespaces",
+            data={"namespace": "asdfasd"},
+        )
+    )
+    adapter = catalog._session.adapters[catalog.uri]
+    adapter.add_headers(prepared)
+
+    assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256")
+    assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}"
+    assert prepared.headers.get("x-amz-content-sha256") != EMPTY_BODY_SHA256
+
+
 def test_list_tables_404(rest_mock: Mocker) -> None:
     namespace = "examples"
     rest_mock.get(