how to mock googleapiclient.discovery.build

  • Last Update :
  • Techknowledgy :

code.py:

import googleapiclient.discovery
import logging

class Service:
   def __init__(self, project, event):
   self.project_id = project
self.compute = googleapiclient.discovery.build('compute', 'v1', cache_discovery = False)
self.event = event
self.zones = self._validate_event()

def _validate_event(self):
   if "jsonPayload" not in self.event:
   zones = self.compute.zones().list(project = self.project_id).execute()['items']
else:
   zones = self.compute.zones().get(project = self.project_id,
      zone = self.event["jsonPayload"]["resource"]["zone"]).execute()

logging.debug(f "Identified Zones are {zones}")
return [zone["name"]
   for zone in zones
]

test_code.py:

from unittest
import TestCase, main
from unittest.mock
import patch
import code

class TestService(TestCase):
   def setUp(self):
   self.project_id = "sample-project-id"

@patch('code.googleapiclient.discovery')
def test__validate_event_with_empty_inputs(self, mock_discovery):
   # Arrange
mock_discovery.build.return_value.zones.return_value.list.return_value.execute.return_value = {
   "items": [{
      "name": "eu-west-1"
   }]
}

# Act
obj = code.Service(event = {}, project = self.project_id)

# Assert
mock_discovery.build.assert_called_once_with('compute', 'v1', cache_discovery = False)
mock_discovery.build.return_value.zones.assert_called_once()
mock_discovery.build.return_value.zones.return_value.list.assert_called_once_with(project = 'sample-project-id')
mock_discovery.build.return_value.zones.return_value.list.return_value.execute.assert_called_once()
self.assertEqual(obj.zones, ["eu-west-1"])

if __name__ == '__main__':
   main()

unit test result with coverage report:

.
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
Ran 1 test in 0.002 s

OK
Name Stmts Miss Cover Missing
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
src / stackoverflow / 56794377 / code.py 14 1 93 % 16
src / stackoverflow / 56794377 / test_code.py 16 0 100 %
   -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
   TOTAL 30 1 97 %

Suggestion : 2

This example uses HttpMock to simulate the basic steps necessary to complete an API call to the Google Books API. The first Mock HTTP returns a status code of 200 and a file named books-discovery.json, which is the discovery document that describes the Books API. This file is a necessary part of the building of the service object, which takes place in the next few lines. The actual request is executed using the second Mock object. This returns the contents of books-android.json, the simulated response.,As you develop and test your application, it is a good idea to save actual API responses in files like books-discovery.json or books-android.json for use in testing.,This class simulates the response to a single HTTP request. As arguments, the constructor for the HttpMock object takes a dictionary object representing the response header and the path to a file. When this resource built on this object is executed, it simply returns contents of the file.,Instead of a pre-defined object, your code can use one of the following keywords as the content object of a simulated HTTP response. At runtime, the Mock object returns the information described in the table.

from googleapiclient.discovery
import build
from googleapiclient.http
import HttpMock
import pprint

http = HttpMock('books-discovery.json', {
   'status': '200'
})
api_key = 'your_api_key'
service = build('books', 'v1', http = http, developerKey = api_key)
request = service.volumes().list(source = 'public', q = 'android')
http = HttpMock('books-android.json', {
   'status': '200'
})
response = request.execute(http = http)
pprint.pprint(response)
from googleapiclient.discovery
import build
from googleapiclient.http
import HttpMockSequence

books_discovery = # Saved data from a build response
books_android = # Saved data from a request to list android volumes

http = HttpMockSequence([
   ({
      'status': '200'
   }, books_discovery),
   ({
      'status': '200'
   }, books_android)
])
api_key = 'your_api_key'
service = build('books', 'v1',
   http = http,
   developerKey = your_api_key)
request = service.volumes().list(source = 'public', q = 'android')
response = request.execute()

Suggestion : 3

I am trying to mock the result of API call anycodings_python-3.7 made to compute engine to list VMs. But anycodings_python-3.7 unfortunately couldn't mock an exact anycodings_python-3.7 function.,You didn't mock the anycodings_python googleapiclient.discovery.build method anycodings_python correctly. Here is the unit test anycodings_python solution:,I've tried using PATCH and MOCK methods to anycodings_python-3.7 mock specific calls made, still unsuccessful,Getting "django.core.exceptions.AppRegistryNotReady" when trying to use Django contrib User module

code.py file looks likes this

import googleapiclient.discovery
import logging

class Service:
   def __init__(self, project, event):
   self.project_id = project
self.compute = googleapiclient.discovery.build('compute', 'v1',
   cache_discovery = False)
self.event = event
self.zones = self._validate_event()

def _validate_event(self):
   if "jsonPayload" not in self.event:
   zones = self.compute.zones().list(
      project = self.project_id).execute()['items']

else:
   zones = self.compute.zones().get(project = self.project_id,
      zone = self.event["jsonPayload"]
      ["resource"]["zone"]).execute()

logging.debug(f "Identified Zones are {zones}")
return [zone["name"]
   for zone in zones
]

My test file looks like this

# in -built
from unittest
import TestCase
from unittest.mock
import patch

# custom
import code

class TestServiceModule(TestCase):
   def setUp(self):
   self.project_id = "sample-project-id"

@patch('code.googleapiclient.discovery')
def test__validate_event_with_empty_inputs(self, mock_discovery):
   mock_discovery.build.zones.list.execute.return_value = {
      "items": [{
         "name": "eu-west-1"
      }]
   }

obj = code.Service(event = {}, project = self.project_id)

print(obj.zones)

code.py:

import googleapiclient.discovery
import logging

class Service:
   def __init__(self, project, event):
   self.project_id = project
self.compute = googleapiclient.discovery.build('compute', 'v1', cache_discovery = False)
self.event = event
self.zones = self._validate_event()

def _validate_event(self):
   if "jsonPayload" not in self.event:
   zones = self.compute.zones().list(project = self.project_id).execute()['items']
else:
   zones = self.compute.zones().get(project = self.project_id,
      zone = self.event["jsonPayload"]["resource"]["zone"]).execute()

logging.debug(f "Identified Zones are {zones}")
return [zone["name"]
   for zone in zones
]

test_code.py:

from unittest
import TestCase, main
from unittest.mock
import patch
import code

class TestService(TestCase):
   def setUp(self):
   self.project_id = "sample-project-id"

@patch('code.googleapiclient.discovery')
def test__validate_event_with_empty_inputs(self, mock_discovery):
   # Arrange
mock_discovery.build.return_value.zones.return_value.list.return_value.execute.return_value = {
   "items": [{
      "name": "eu-west-1"
   }]
}

# Act
obj = code.Service(event = {}, project = self.project_id)

# Assert
mock_discovery.build.assert_called_once_with('compute', 'v1', cache_discovery = False)
mock_discovery.build.return_value.zones.assert_called_once()
mock_discovery.build.return_value.zones.return_value.list.assert_called_once_with(project = 'sample-project-id')
mock_discovery.build.return_value.zones.return_value.list.return_value.execute.assert_called_once()
self.assertEqual(obj.zones, ["eu-west-1"])

if __name__ == '__main__':
   main()

unit test result with coverage report:

.
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- --
Ran 1 test in 0.002 s

OK
Name Stmts Miss Cover Missing
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
src / stackoverflow / 56794377 / code.py 14 1 93 % 16
src / stackoverflow / 56794377 / test_code.py 16 0 100 %
   -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
   TOTAL 30 1 97 %

Suggestion : 4

Here are the examples of the python api googleapiclient.discovery.build taken from open source projects. By voting up you can indicate which examples are most useful and appropriate., googleapiclient.discovery

def fetch_youtube_url(search_term):
   ""
"For each song name/artist name combo, fetch the YouTube URL
and
return the list of URLs ""
"
YOUTUBE_DEV_KEY = getenv('YOUTUBE_DEV_KEY')
youtube = build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION,
   developerKey = YOUTUBE_DEV_KEY)
log.info(u "Searching for {}".format(search_term))
search_response = youtube.search().list(q = search_term,
   part = 'id, snippet').execute()
for v in search_response['items']:
   if v['id']['kind'] == VIDEO:
   log.debug("Adding Video id {}".format(v['id']['videoId']))
return YOUTUBE_VIDEO_URL + v['id']['videoId']
def _get_drive_service(settings):
   credentials_file = _default_drive_credentials_path(settings)
storage = Storage(credentials_file)
credentials = storage.get()
auth_http = credentials.authorize(httplib2.Http())
return build('drive', 'v2', http = auth_http)
def create_service():
   # Get the application
default credentials.When running locally, these are
# available after running `gcloud init`.When running on compute
# engine, these are available from the environment.
credentials = GoogleCredentials.get_application_default()

# Construct the service object
for interacting with the Cloud Storage API -
   # the 'storage'
service, at version 'v1'.
# You can browse other available api services and versions here:
   # http: //g.co/dv/api-client-library/python/apis/
   return discovery.build('storage', 'v1', credentials = credentials)
    def __init__(self, flags):
       self.__flags = flags

    # Perform OAuth 2.0 authorization.
    project_dir = os.path.join(
       os.getenv('HOME'), 'cloud', 'projects', flags.project)
    client_secrets = os.path.join(project_dir, 'client_secrets.json')
    oauth2_storage = os.path.join(project_dir, 'oauth2.dat')
    flow = flow_from_clientsecrets(client_secrets, scope = GCE_SCOPE)
    storage = Storage(oauth2_storage)
    credentials = storage.get()

    if credentials is None or credentials.invalid:
       credentials = run_flow(flow, storage, flags)
    self.http = credentials.authorize(httplib2.Http())
    self.compute = build('compute', API_VERSION)
def get_client():
   ""
"Builds an http client authenticated with the application default
credentials.
""
"
credentials = GoogleCredentials.get_application_default()
client = discovery.build(
   'cloudmonitoring', 'v2beta2',
   credentials = credentials)
return client
def get_speech_service():
   credentials = GoogleCredentials.get_application_default().create_scoped(
      ['https://www.googleapis.com/auth/cloud-platform'])
http = httplib2.Http()
credentials.authorize(http)

return discovery.build(
   'speech', 'v1beta1', http = http, discoveryServiceUrl = DISCOVERY_URL)