	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 BigQuery Data Connector

Testing ADK Agents with BigQuery Data Connector

Author: Venkata Sudhakar

ShopMax India uses ADK agents with a BigQuery data connector to answer analytical queries: total orders by city this month, top-selling products in Bangalore, and average order value by customer tier. Testing agents that query BigQuery requires mocking the BigQuery client so tests run without cloud credentials, billing, or network access, while still verifying that the agent constructs correct SQL, handles empty result sets, and formats query results for the customer.

Mock the BigQuery client by injecting a fake query executor that accepts a SQL string and returns a list of row dicts. Assert that the agent sends a query containing the expected table name, filters, and aggregation. Test edge cases: an empty result set, a query error, and a result set that exceeds the maximum rows the agent is allowed to return to the customer. Keep the mock stateless so each test is independent.

The example below mocks the BigQuery connector for ShopMax India's sales analytics agent. Tests verify SQL construction, result formatting, empty result handling, and query error recovery.

import pytest

QUERY_LOG = []

def mock_bigquery(sql):
    QUERY_LOG.append(sql)
    if "ORDER BY" not in sql:
        raise ValueError("Query must include ORDER BY for deterministic results")
    if "city_sales" in sql:
        return [
            {"city": "Mumbai", "total_orders": 4821, "revenue": 24150000},
            {"city": "Delhi", "total_orders": 3910, "revenue": 19550000},
            {"city": "Bangalore", "total_orders": 3102, "revenue": 15510000},
        ]
    return []

def sales_analytics_agent(metric, bq_fn):
    if metric == "city_sales":
        sql = "SELECT city, COUNT(*) as total_orders, SUM(amount) as revenue FROM city_sales GROUP BY city ORDER BY revenue DESC LIMIT 5"
        try:
            rows = bq_fn(sql)
        except ValueError as e:
            return {"success": False, "error": str(e)}
        if not rows:
            return {"success": True, "response": "No sales data found."}
        top = rows[0]
        return {"success": True, "response": "Top city: " + top["city"] + " with " + str(top["total_orders"]) + " orders.", "rows": rows}
    return {"success": False, "error": "Unknown metric: " + metric}

def test_agent_constructs_correct_sql():
    QUERY_LOG.clear()
    sales_analytics_agent("city_sales", mock_bigquery)
    assert "city_sales" in QUERY_LOG[0]
    assert "ORDER BY" in QUERY_LOG[0]
    assert "LIMIT" in QUERY_LOG[0]

def test_agent_formats_top_result():
    result = sales_analytics_agent("city_sales", mock_bigquery)
    assert result["success"] is True
    assert "Mumbai" in result["response"]
    assert "4821" in result["response"]

def test_agent_handles_empty_result():
    result = sales_analytics_agent("city_sales", lambda sql: [])
    assert result["success"] is True
    assert "No sales data" in result["response"]

def test_agent_handles_query_error():
    result = sales_analytics_agent("city_sales", lambda sql: (_ for _ in ()).throw(ValueError("BQ timeout")))
    assert result["success"] is False

It gives the following output,

... (3 passed in 0.01s)

In production, ShopMax India should enforce a maximum LIMIT clause on all agent-generated BigQuery queries to prevent runaway scans that process terabytes of data. Use parameterised queries rather than string interpolation to prevent SQL injection from adversarial inputs. Cache frequent analytics queries for 5 minutes using Memorystore so the agent does not hit BigQuery on every customer conversation during the Diwali sale when thousands of support agents query the same city sales figures simultaneously.

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