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 %
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()
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 %
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)