Fix: Add logs on AssertionError for issue #1512 (#1518)

chandra-siri · web-flow · commit 6a9923e4fc94 · 2025-07-30T10:51:23.000+05:30
Adding Response.content and headers will help debug this flaky issue which throws 403, having more verbose information would be helpful. Closes #1512
diff --git a/tests/system/test__signing.py b/tests/system/test__signing.py
@@ -51,10 +51,17 @@ def _create_signed_list_blobs_url_helper(
     )
 
     response = requests.get(signed_url)
-    assert response.status_code == 200
+    assert (
+        response.status_code == 200
+    ), f"Response content start: {response.content} \
+        :Response content end. \
+        Response headers start: {response.headers} \
+        :Response headers end."
 
 
-def test_create_signed_list_blobs_url_v2(storage_client, signing_bucket, no_mtls):
+def test_create_signed_list_blobs_url_v2(
+    storage_client, signing_bucket, no_mtls
+):
     _create_signed_list_blobs_url_helper(
         storage_client,
         signing_bucket,
@@ -76,7 +83,9 @@ def test_create_signed_list_blobs_url_v2_w_expiration(
     )
 
 
-def test_create_signed_list_blobs_url_v4(storage_client, signing_bucket, no_mtls):
+def test_create_signed_list_blobs_url_v4(
+    storage_client, signing_bucket, no_mtls
+):
     _create_signed_list_blobs_url_helper(
         storage_client,
         signing_bucket,
@@ -137,7 +146,12 @@ def _create_signed_read_url_helper(
         headers["x-goog-encryption-key-sha256"] = key_hash
 
     response = requests.get(signed_url, headers=headers)
-    assert response.status_code == 200
+    assert (
+        response.status_code == 200
+    ), f"Response content start: {response.content} \
+        :Response content end. \
+        Response headers start: {response.headers} \
+        :Response headers end."
 
     if payload is not None:
         assert response.content == payload
@@ -215,7 +229,9 @@ def test_create_signed_read_url_v4_w_non_ascii_name(
     )
 
 
-def test_create_signed_read_url_v2_w_csek(storage_client, signing_bucket, no_mtls):
+def test_create_signed_read_url_v2_w_csek(
+    storage_client, signing_bucket, no_mtls
+):
     encryption_key = os.urandom(32)
     _create_signed_read_url_helper(
         storage_client,
@@ -226,7 +242,9 @@ def test_create_signed_read_url_v2_w_csek(storage_client, signing_bucket, no_mtl
     )
 
 
-def test_create_signed_read_url_v4_w_csek(storage_client, signing_bucket, no_mtls):
+def test_create_signed_read_url_v4_w_csek(
+    storage_client, signing_bucket, no_mtls
+):
     encryption_key = os.urandom(32)
     _create_signed_read_url_helper(
         storage_client,
@@ -309,7 +327,9 @@ def test_create_signed_read_url_v4_w_access_token_universe_domain(
         "https://www.googleapis.com/auth/devstorage.read_write",
         "https://www.googleapis.com/auth/iam",
     ]
-    response = universe_domain_iam_client.generate_access_token(name=name, scope=scope)
+    response = universe_domain_iam_client.generate_access_token(
+        name=name, scope=scope
+    )
 
     _create_signed_read_url_helper(
         universe_domain_client,
@@ -320,7 +340,9 @@ def test_create_signed_read_url_v4_w_access_token_universe_domain(
     )
 
 
-def _create_signed_delete_url_helper(client, bucket, version="v2", expiration=None):
+def _create_signed_delete_url_helper(
+    client, bucket, version="v2", expiration=None
+):
     expiration = _morph_expiration(version, expiration)
 
     blob = bucket.blob("DELETE_ME.txt")
@@ -335,7 +357,12 @@ def _create_signed_delete_url_helper(client, bucket, version="v2", expiration=No
 
     response = requests.request("DELETE", signed_delete_url)
 
-    assert response.status_code == 204
+    assert (
+        response.status_code == 204
+    ), f"Response content start: {response.content} \
+        :Response content end. \
+        Response headers start: {response.headers} \
+        :Response headers end."
     assert response.content == b""
     assert not blob.exists()
 
@@ -345,7 +372,9 @@ def test_create_signed_delete_url_v2(storage_client, signing_bucket, no_mtls):
 
 
 def test_create_signed_delete_url_v4(storage_client, signing_bucket, no_mtls):
-    _create_signed_delete_url_helper(storage_client, signing_bucket, version="v4")
+    _create_signed_delete_url_helper(
+        storage_client, signing_bucket, version="v4"
+    )
 
 
 def _create_signed_resumable_upload_url_helper(
@@ -364,22 +393,39 @@ def _create_signed_resumable_upload_url_helper(
     )
 
     post_headers = {"x-goog-resumable": "start"}
-    post_response = requests.post(signed_resumable_upload_url, headers=post_headers)
-    assert post_response.status_code == 201
+    post_response = requests.post(
+        signed_resumable_upload_url, headers=post_headers
+    )
+    assert (
+        post_response.status_code == 201
+    ), f"Response content start: {post_response.content} \
+        :Response content end. \
+        Response headers start: {post_response.headers} \
+        :Response headers end."
 
     # Finish uploading the body.
     location = post_response.headers["Location"]
     put_headers = {"content-length": str(len(payload))}
     put_response = requests.put(location, headers=put_headers, data=payload)
-    assert put_response.status_code == 200
+    assert (
+        put_response.status_code == 200
+    ), f"Response content start: {put_response.content} \
+        :Response content end. \
+        Response headers start: {put_response.headers} \
+        :Response headers end."
 
     # Download using a signed URL and verify.
     signed_download_url = blob.generate_signed_url(
         expiration=expiration, method="GET", client=client, version=version
     )
 
     get_response = requests.get(signed_download_url)
-    assert get_response.status_code == 200
+    assert (
+        get_response.status_code == 200
+    ), f"Response content start: {get_response.content} \
+        :Response content end. \
+        Response headers start: {get_response.headers} \
+        :Response headers end."
     assert get_response.content == payload
 
     # Finally, delete the blob using a signed URL.
@@ -391,18 +437,27 @@ def _create_signed_resumable_upload_url_helper(
     )
 
     delete_response = requests.delete(signed_delete_url)
-    assert delete_response.status_code == 204
+    assert (
+        delete_response.status_code == 204
+    ), f"Response content start: {delete_response.content} \
+        :Response content end. \
+        Response headers start: {delete_response.headers} \
+        :Response headers end."
 
 
-def test_create_signed_resumable_upload_url_v2(storage_client, signing_bucket, no_mtls):
+def test_create_signed_resumable_upload_url_v2(
+    storage_client, signing_bucket, no_mtls
+):
     _create_signed_resumable_upload_url_helper(
         storage_client,
         signing_bucket,
         version="v2",
     )
 
 
-def test_create_signed_resumable_upload_url_v4(storage_client, signing_bucket, no_mtls):
+def test_create_signed_resumable_upload_url_v4(
+    storage_client, signing_bucket, no_mtls
+):
     _create_signed_resumable_upload_url_helper(
         storage_client,
         signing_bucket,
@@ -439,10 +494,17 @@ def test_generate_signed_post_policy_v4(
     )
     with open(blob_name, "r") as f:
         files = {"file": (blob_name, f)}
-        response = requests.post(policy["url"], data=policy["fields"], files=files)
+        response = requests.post(
+            policy["url"], data=policy["fields"], files=files
+        )
 
     os.remove(blob_name)
-    assert response.status_code == 204
+    assert (
+        response.status_code == 204
+    ), f"Response content start: {response.content} \
+        :Response content end. \
+        Response headers start: {response.headers} \
+        :Response headers end."
 
     blob = bucket.get_blob(blob_name)
     assert blob.download_as_bytes() == payload
@@ -487,10 +549,17 @@ def test_generate_signed_post_policy_v4_access_token_sa_email(
     )
     with open(blob_name, "r") as f:
         files = {"file": (blob_name, f)}
-        response = requests.post(policy["url"], data=policy["fields"], files=files)
+        response = requests.post(
+            policy["url"], data=policy["fields"], files=files
+        )
 
     os.remove(blob_name)
-    assert response.status_code == 204
+    assert (
+        response.status_code == 204
+    ), f"Response content start: {response.content} \
+        :Response content end. \
+        Response headers start: {response.headers} \
+        :Response headers end."
 
     blob = signing_bucket.get_blob(blob_name)
     blobs_to_delete.append(blob)
@@ -522,9 +591,16 @@ def test_generate_signed_post_policy_v4_invalid_field(
     )
     with open(blob_name, "r") as f:
         files = {"file": (blob_name, f)}
-        response = requests.post(policy["url"], data=policy["fields"], files=files)
+        response = requests.post(
+            policy["url"], data=policy["fields"], files=files
+        )
 
     os.remove(blob_name)
-    assert response.status_code == 400
+    assert (
+        response.status_code == 400
+    ), f"Response content start: {response.content} \
+        :Response content end. \
+        Response headers start: {response.headers} \
+        :Response headers end."
 
     assert list(bucket.list_blobs()) == []
