feat(firestore): configure gRPC message length limits for large documents

[ohmayr]

• Configure grpc.max_send_message_length and grpc.max_receive_message_length options on production and emulator gRPC channels.
• Update unit tests in test_base_client.py to assert gRPC message size options.
• Add system tests for large document CRUD and pipeline execution across sync and async clients.

[gemini-code-assist]

Code Review

This pull request configures unlimited gRPC send and receive message sizes for Firestore clients, including emulator channels, to support handling large documents. It also adds system tests for both synchronous and asynchronous operations with large documents on Enterprise DB, along with corresponding unit test updates. The review feedback highlights a critical issue in the new asynchronous system tests: passing the asynchronous doc_ref.delete coroutine to a synchronous cleanup fixture will cause cleanup to fail and raise a RuntimeWarning. It is recommended to use try...finally blocks to explicitly await the deletion of the documents.

[gemini-code-assist]

In async tests, doc_ref.delete is an asynchronous coroutine function. Passing it directly to a synchronous cleanup fixture will result in the coroutine being called but never awaited, which triggers a RuntimeWarning: coroutine 'AsyncDocumentReference.delete' was never awaited and fails to clean up the document in the database. Instead, use a try...finally block to explicitly await the deletion of the document.

Suggested change

	async def test_large_document_standard_writes_async(client, cleanup, database):
	"""Test standard write and read operations for large document on Enterprise DB (async)."""
	collection_id = "large_docs_async_" + UNIQUE_RESOURCE_ID
	doc_ref = client.collection(collection_id).document("large_doc")
	cleanup(doc_ref.delete)
	large_payload = "c" * (900 * 1024)
	await doc_ref.set({"payload": large_payload})
	snapshot = await doc_ref.get()
	assert snapshot.exists
	assert snapshot.to_dict() == {"payload": large_payload}
	async def test_large_document_standard_writes_async(client, database):
	"""Test standard write and read operations for large document on Enterprise DB (async)."""
	collection_id = "large_docs_async_" + UNIQUE_RESOURCE_ID
	doc_ref = client.collection(collection_id).document("large_doc")
	try:
	large_payload = "c" * (900 * 1024)
	await doc_ref.set({"payload": large_payload})
	snapshot = await doc_ref.get()
	assert snapshot.exists
	assert snapshot.to_dict() == {"payload": large_payload}
	finally:
	await doc_ref.delete()

[gemini-code-assist]

Similar to the standard writes test, doc_ref.delete is an asynchronous coroutine function. Passing it to the synchronous cleanup fixture will not await it, leading to un-deleted documents and RuntimeWarnings. Please use a try...finally block to ensure the document is properly cleaned up.

Suggested change

	async def test_large_document_pipeline_async(client, cleanup, database, method):
	"""Test async pipeline execution over large document on Enterprise DB."""
	collection_id = "large_pipeline_async_" + UNIQUE_RESOURCE_ID
	col_ref = client.collection(collection_id)
	doc_ref = col_ref.document("large_doc")
	cleanup(doc_ref.delete)
	large_payload = "d" * (900 * 1024)
	await doc_ref.set({"payload": large_payload})
	pipeline = client.pipeline().collection(collection_id)
	if method == "execute":
	results = await pipeline.execute()
	else:
	results = [doc async for doc in pipeline.stream()]
	assert [doc.data() for doc in results] == [{"payload": large_payload}]
	async def test_large_document_pipeline_async(client, database, method):
	"""Test async pipeline execution over large document on Enterprise DB."""
	collection_id = "large_pipeline_async_" + UNIQUE_RESOURCE_ID
	col_ref = client.collection(collection_id)
	doc_ref = col_ref.document("large_doc")
	try:
	large_payload = "d" * (900 * 1024)
	await doc_ref.set({"payload": large_payload})
	pipeline = client.pipeline().collection(collection_id)
	if method == "execute":
	results = await pipeline.execute()
	else:
	results = [doc async for doc in pipeline.stream()]
	assert [doc.data() for doc in results] == [{"payload": large_payload}]
	finally:
	await doc_ref.delete()