Python Unit Testing Google Bigquery

Question

I am having trouble in unit testing the following code block:

from google.cloud import bigquery
from google.oauth2 import service_account

def run_query(query, gcp_ser_acc):
    credentials = 
    service_account.Credentials.from_service_account_info(gcp_ser_acc)
    client = bigquery.Client(gcp_ser_acc['project_id'], credentials)
    query_job = client.query(query)
    results = query_job.result()
    return results

I am new to mocking and I have tried the following test:

def test_run_a_query_with_real_key(self):
    gcp_ser_acc = {
                'project_id': 'my_project_id',
                'private_key': 'my_private_key',
                'token_uri': 'my_token_uri',
                'client_email': 'my_client_email'
                }
    with mock.patch('service_account.Credentials', call_args=gcp_ser_acc, return_value={}):
        with mock.patch('bigquery.Client', call_args=(gcp_ser_acc['project_id'], {}), return_value={}):
            run_query('SELECT 1+1 as col', gcp_ser_acc)
            assert service_account.Credentials.called
            assert bigquery.Client.called

Can anybody mock the google stuff and write a unit test please?

Answer 1

This is how you mock google.cloud.bigquery with pytest, pytest-mock

from google.cloud import bigquery


schema = [
    bigquery.SchemaField("full_name", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("age", "INTEGER", mode="REQUIRED"),
]


def some_query(table_name='blahblahbloo'):

    client = bigquery.Client()
    table_id = f"project.dataset.{table_name}"
    table = bigquery.Table(table_id, schema=schema)
    table = client.create_table(table)


def test_some_query(mocker):
    mock_table = mocker.patch('google.cloud.bigquery.Table', autospec=True)
    mock_client = mocker.patch('google.cloud.bigquery.Client', autospec=True)

    some_query()  # run with mocked objects

    mock_table.assert_called_with('project.dataset.blahblahbloo', schema=schema)
    mock_client().create_table.assert_called_with(mock_table.return_value)