4

Could anybody help with GCP API mocking? Here's the function func.py:

import re
from google.cloud import storage
def is_all_log_entries_sink(sink):
    storage_client = storage.Client()
    if 'filter' not in sink and 'storage.googleapis.com' in sink.get('destination', ''):
        bucket_name = re.search(r'^storage.googleapis.com/([^/]+)$', sink.get('destination')).group(1)  
        bucket = storage_client.lookup_bucket(bucket_name)
        if bucket is not None:
            return True
    return False 

Here's the test:

import mock
from mock import patch, MagicMock
with mock.patch('oauth2client.client.GoogleCredentials.get_application_default') as mock_method:
    import func
@patch('func.storage_client')
def test_is_all_log_entries_sink(mock_obj):
    mock_obj.lookup_bucket = MagicMock(return_value='bucket-name')
    sink = {'name': 'testname', 'destination': 'storage.googleapis.com/bucket-name'}
    assert func.is_all_log_entries_sink(sink) == 1
    assert mock_obj.lookup_bucket.called
    sink = {'name': 'testname', 'destination': 'storage.googleapis.com/bucket-name', 'filter': 'testfilter'}
    assert func.is_all_log_entries_sink(sink) == 0
    sink = {'name': 'testname', 'destination': 'storage.googleapis.com/bucket-name'}
    mock_obj.lookup_bucket = MagicMock(return_value=None)
    assert func.is_all_log_entries_sink(sink) == 0

When PyTest ran, the following error was received:

E   google.auth.exceptions.DefaultCredentialsError: Could not automatically determine credentials. Please set GOOGLE_APPLICATION_CREDENTIALS or explicitly create credentials and
 re-run the application.

I tried to mock Google's authentication, but was unsuccessful. Any assistance would be appreciated.

Maxim
  • 4,075
  • 1
  • 14
  • 23
kenwynejohnes
  • 53
  • 1
  • 6

2 Answers2

2

One possible solution: to mock GOOGLE_APPLICATION_CREDENTIALS as env variable in os module. I consider GOOGLE_APPLICATION_CREDENTIALS as dict, so mocking dict may help you.

e.g. here

Oleksiy
  • 794
  • 2
  • 14
  • 39
0
from unittest import mock

@mock.patch('func.storage')
def test_is_all_log_entries_sink(self, mock_storage):
    # Set up the mock objects
    client_mock = mock_storage.Client.return_value
    # call your main function and do your assertions next

You can mock Google storage using mock.patch like this.

Abhishek Poojary
  • 749
  • 9
  • 10