storage: upload_chunks_concurrently ignores its timeout argument #18764

"""Show which timeout each XML MPU request gets from upload_chunks_concurrently.

Runs the real transfer_manager and XML MPU code with a transport that records
the timeout of every request and answers like Cloud Storage would. No network.
"""

import os
import tempfile

import requests
from google.auth.credentials import AnonymousCredentials
from google.cloud.storage import Client, transfer_manager

INITIATE_BODY = (
    '<?xml version="1.0" encoding="UTF-8"?>'
    '<InitiateMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
    "<UploadId>upload-1</UploadId></InitiateMultipartUploadResult>"
)


class RecordingTransport(requests.Session):
    is_mtls = False

    def __init__(self):
        super().__init__()
        self.calls = []

    def request(self, method, url, data=None, headers=None, timeout=None, **kwargs):
        query = url.split("?", 1)[1] if "?" in url else ""
        if method == "POST" and query == "uploads":
            step, body, extra = "initiate", INITIATE_BODY, {}
        elif method == "PUT":
            step, body, extra = "upload part", "", {"ETag": '"etag-1"'}
        elif method == "POST":
            step, body, extra = "finalize", "", {}
        else:
            step, body, extra = "cancel", "", {}
        self.calls.append((step, timeout))
        response = requests.Response()
        response.status_code = 204 if step == "cancel" else 200
        response._content = body.encode()
        response.headers.update(extra)
        response.url = url
        return response


transport = RecordingTransport()
client = Client(project="demo", credentials=AnonymousCredentials(), _http=transport)
blob = client.bucket("demo-bucket").blob("big-file.bin")

with tempfile.NamedTemporaryFile(delete=False) as handle:
    handle.write(b"x" * 10)
try:
    transfer_manager.upload_chunks_concurrently(
        handle.name,
        blob,
        chunk_size=5,
        worker_type=transfer_manager.THREAD,
        checksum=None,
        timeout=5,
    )
except Exception as error:
    print(f"raised {type(error).__name__}: {error}")
finally:
    os.unlink(handle.name)

for step, timeout in transport.calls:
    print(f"{step:12} timeout={timeout}")