test(storage): add system tests for Bidi Read/Write (gRPC)

[nidhiii-27]

Implement parameterized integration tests for Bidirectional Read and Write
features, porting test specifications from the Java reference PR.
Tests cover standard, RCU, and zonal buckets, and include checksum
validation and stream closure scenarios.

[Generated-by: AI]

[gemini-code-assist]

Code Review

This pull request introduces a new system test suite (test_bidi.py) to verify bidirectional streaming read and write operations using AsyncAppendableObjectWriter and AsyncMultiRangeDownloader. The feedback suggests several improvements to the test suite: explicitly catching asyncio.CancelledError (which inherits from BaseException rather than Exception), implementing a pytest fixture to handle temporary object cleanup robustly on test failures, and checking for exceptions returned by asyncio.gather to prevent masking failures during assertions.

[gemini-code-assist]

In Python 3.8+, asyncio.CancelledError inherits from BaseException rather than Exception. As a result, pytest.raises(Exception) will not catch asyncio.CancelledError, which can cause the test to fail with an uncaught exception if the task is cancelled. To ensure both ServiceUnavailable and CancelledError are correctly caught, specify them explicitly in pytest.raises.

Suggested change

	with pytest.raises(Exception) as excinfo:
	await task
	# The exception could be ServiceUnavailable or CancelledError
	assert isinstance(excinfo.value, (exceptions.ServiceUnavailable, asyncio.CancelledError))
	with pytest.raises((exceptions.ServiceUnavailable, asyncio.CancelledError)) as excinfo:
	await task

[gemini-code-assist]

To prevent leaving orphaned objects in the preprod/prod bucket when assertions fail, consider using a pytest fixture to track and automatically clean up created objects. This ensures cleanup runs even if a test fails mid-execution.

Suggested change

	# Helper to create objects using sync client if needed,
	@pytest.fixture
	async def temp_objects(grpc_client, bidi_bucket):
	created_objects = []
	yield created_objects
	for object_name in created_objects:
	try:
	await grpc_client.delete_object(bidi_bucket, object_name)
	except Exception:
	pass
	# Helper to create objects using sync client if needed,

[gemini-code-assist]

When using asyncio.gather with return_exceptions=True, any exception raised by valid_task will be returned in results[0] instead of being raised. If valid_task fails, the assertion valid_buffer.getvalue() == data[:100] will fail, masking the actual exception and making debugging difficult. Checking and raising the exception from results[0] first ensures the real failure is reported with its stack trace.

Suggested change

	# Verify valid one processed correctly
	assert valid_buffer.getvalue() == data[:100]
	# Verify valid one processed correctly
	if isinstance(results[0], Exception):
	raise results[0]
	assert valid_buffer.getvalue() == data[:100]