	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > ADK Agent Testing > Testing ADK Agents with Cloud Storage Data Connector

Testing ADK Agents with Cloud Storage Data Connector

Author: Venkata Sudhakar

ShopMax India's ADK agents read product catalogs, warranty PDFs, and return policy documents from Google Cloud Storage to answer customer queries. Testing agents that access Cloud Storage requires mocking the GCS client so tests run offline without credentials, while verifying that the agent reads from the correct bucket and object path, handles missing files gracefully, and correctly parses the retrieved content before using it in a response.

Inject a fake GCS client into the agent tool that stores objects in a dict keyed by bucket/object path. The fake supports get (returns bytes), upload (stores bytes), and exists (checks presence). Assert that the agent requests the expected path, uses the file content in its response, and returns a clear error when the object does not exist. This covers the most common failure modes: wrong path, missing file, and corrupt content.

The example below mocks the GCS client for ShopMax India's warranty lookup agent. Tests verify correct path construction, content usage in the response, missing file handling, and upload of a generated certificate.

import pytest

class FakeGCSClient:
    def __init__(self): self._store = {}

def upload(self, bucket, path, data): self._store[bucket + "/" + path] = data

def get(self, bucket, path):
        key = bucket + "/" + path
        if key not in self._store:
            raise FileNotFoundError("Object not found: " + key)
        return self._store[key]

def exists(self, bucket, path): return (bucket + "/" + path) in self._store

def warranty_lookup_agent(order_id, gcs):
    bucket = "shopmax-warranties"
    path = "warranties/" + order_id + ".txt"
    try:
        content = gcs.get(bucket, path)
        return {"found": True, "response": "Warranty: " + content.decode("utf-8"), "path": path}
    except FileNotFoundError:
        return {"found": False, "response": "No warranty found for " + order_id}

def upload_certificate(order_id, text, gcs):
    bucket = "shopmax-certificates"
    path = "certs/" + order_id + ".txt"
    gcs.upload(bucket, path, text.encode("utf-8"))
    return {"uploaded": True, "path": path}

def test_agent_reads_correct_path():
    gcs = FakeGCSClient()
    gcs.upload("shopmax-warranties", "warranties/ORD-1101.txt", b"2-year warranty. Valid in India.")
    result = warranty_lookup_agent("ORD-1101", gcs)
    assert result["found"] is True
    assert result["path"] == "warranties/ORD-1101.txt"

def test_agent_uses_file_content():
    gcs = FakeGCSClient()
    gcs.upload("shopmax-warranties", "warranties/ORD-1102.txt", b"1-year warranty for Mumbai customer.")
    result = warranty_lookup_agent("ORD-1102", gcs)
    assert "Mumbai" in result["response"]

def test_agent_handles_missing_file():
    gcs = FakeGCSClient()
    result = warranty_lookup_agent("ORD-9999", gcs)
    assert result["found"] is False
    assert "No warranty found" in result["response"]

def test_upload_certificate():
    gcs = FakeGCSClient()
    result = upload_certificate("ORD-1103", "Certificate for Priya, Delhi - Rs 32000", gcs)
    assert result["uploaded"] is True
    assert gcs.exists("shopmax-certificates", result["path"])

It gives the following output,

.... (4 passed in 0.01s)

In production, ShopMax India should use IAM conditions to restrict agent access to specific GCS buckets and object prefixes, preventing the agent from inadvertently reading or overwriting files outside its intended scope. Enable object versioning on the warranties bucket so accidental overwrites can be recovered. Set signed URL expiry to 15 minutes for any warranty or certificate URLs sent to customers, and validate the order ID format before constructing GCS paths to prevent path traversal attempts.

Send your comments, suggestions or queries regarding this site to [email protected].