Below I am using a python client library to connect to bigquery and a service account to connect to the bigquery (see this link to get more info about the library).
from google.cloud.bigquery import Client, LoadJobConfig, LoadJob
from google.oauth2 import service_account
def fetch_last_modified_date_from_bq():
service_account_json_string = "json to connect"
service_account_json = json.loads(service_account_json_string)
credentials = service_account.Credentials.from_service_account_info(service_account_json)
client = Client(credentials=credentials)
# Run a SQL query on the table
sql = """
SELECT last_modified_date FROM `project.dataset.table`
order by last_modified_date desc LIMIT 1
"""
query_job = client.query(sql)
# Print the results
for row in query_job:
return row.result //this is string
How to mock the client?
I tried to mock in this way but as client needs "credentials" to be passed and it was not successful.
Is the code below the right way?
@mock.patch('google.cloud.bigquery.Client')
@mock.patch('google.oauth2.service_account.Credentials.from_service_account_info')
@mock.patch('google.auth.credentials.Credentials')
def test_fetch_last_modified_date_from_bq(self, mock_credentials ,mock_service_account, mock_client):
#arrange
mock_service_account.return_value = mock_credentials
row={}
row['last_modified_date'] = ''
rows = [row]
mock_client.return_value.query.return_value = rows
#act
fetch_last_modified_date_from_bq()
biquery_store.fetch_last_modified_date_from_bq()
which you call by the test method? – Liskfetch_last_modified_date_from_bq()
you assign somethin tocredentials
two times? Why? – Liskfetch_last_modified_date_from_bq()
and the testtest_fetch_last_modified_date_from_bq()
are in the same file? – Lisk