Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_validate_token_true(self):
header = httplib2.Http(".cache")
url = '%stokens/%s?belongsTo=%s' % (utils.URL, self.token, self.tenant)
resp, content = header.request(url, "GET", body='',
headers={"Content-Type": "application/json",
"X-Auth-Token": self.auth_token})
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
self.fail('Service Not Available')
self.assertEqual(200, int(resp['status']))
self.assertEqual('application/json', utils.content_type(resp))
#verify content
obj = json.loads(content)
if not "auth" in obj:
raise self.fail("Expecting Auth")
role_refs = obj["auth"]["user"]["roleRefs"]
def delete_token(token, auth_token):
header = httplib2.Http(".cache")
url = '%stoken/%s' % (URL, token)
resp, content = header.request(url, "DELETE", body='',
headers={"Content-Type": "application/json",
"X-Auth-Token": auth_token})
return (resp, content)
def fetch():
h = httplib2.Http()
credentials = get_credentials()
credentials.authorize(h)
try:
agendas = {}
for key, cal_id in settings.GOOGLE_CALENDAR_IDS.items():
agendas[key] = fetch_agenda(h, cal_id)
return agendas
except httplib2.ServerNotFoundError:
return None
global verbose
verbose = args.v
# Load the inboice and select the one to process
invoices = load_invoices(args.data_file)
if not str(args.n_invoice) in invoices:
print("Unknown invoice %i. Known invoices: %s" % (args.n_invoice, invoices.keys()))
return
invoice = invoices[str(args.n_invoice)]
invoice['number'] = args.n_invoice
if 'template' not in invoice:
invoice['template'] = 'Template'
# find the 'DriveInvoicing' directory and look for the file 'Template' inside it
credentials = get_credentials()
http = credentials.authorize(httplib2.Http())
drive_service = discovery.build('drive', 'v3', http=http)
folder_id = get_folder(drive_service, 'DriveInvoicing')['id']
template_id = get_content(drive_service, invoice['template'], folder_id)['id']
# Copy the template
invoice_doc_id = copy_file(drive_service, template_id, 'Invoice_%i' % invoice['number'], folder_id)['id']
# Run the script to fill the template
script_service = discovery.build('script', 'v1', http=http)
paymentDays = None
currency = DEFAULT_CURRENCY
if 'currency' in invoice:
currency = invoice['currency']
if 'paymentDays' in invoice:
paymentDays = invoice['paymentDays']
def get(self, path='me', fql=None, connection_type=None, metadata=False,
format='json'):
token_url = self._create_token_url(path, fql, connection_type, metadata,
format)
h = httplib2.Http()
try:
headers, response = h.request(token_url, 'GET')
except Exception, e:
logging.error(e)
return HttpResponseServerError
if response in ['true', 'false']:
response = {'error':
'Facebook Graph API returned unexpected boolean.'}
return self._error_handler(response, fql, token_url)
response = simplejson.loads(response)
if 'error' in response:
return self._error_handler(response, fql, token_url)
elif 'data' in response:
return response['data']
def get_user_email(oauth_credentials):
http = httplib2.Http()
http = oauth_credentials.authorize(http)
user_info = build('oauth2', 'v2').tokeninfo().execute(http)
if not user_info['verified_email']:
return None
return user_info['email'].lower()
def main(argv):
# Parse the command-line flags.
flags = parser.parse_args(argv[1:])
# Obtain service account credentials from virtual machine environement.
credentials = AppAssertionCredentials(['https://www.googleapis.com/auth/compute'])
# Create an httplib2.Http object to handle our HTTP requests and authorize
# it with our good Credentials.
http = httplib2.Http()
http = credentials.authorize(http)
# Construct the service object for the interacting with the Compute Engine
# API.
service = discovery.build('compute', 'v1', http=http)
# Set project, zone, and other constants.
URL_PREFIX = 'https://www.googleapis.com/compute'
API_VERSION = 'v1'
PROJECT_ID = 'your-project-id'
PROJECT_URL = '%s/%s/projects/%s' % (URL_PREFIX, API_VERSION, PROJECT_ID)
INSTANCE_NAME = 'test-vm-serv-acct'
ZONE = 'us-central1-a'
MACHINE_TYPE = 'n1-standard-1'
IMAGE_PROJECT_ID = 'debian-cloud'
IMAGE_PROJECT_URL = '%s/%s/projects/%s' % (
doc_headers = doc['headers']
if 'date-modified' in doc_headers:
req_headers['If-Modified-Since'] = \
doc_headers['date-modified'].encode('utf-8')
if 'etag' in doc_headers:
req_headers['If-None-Match'] = doc_headers['etag'].encode('utf-8')
# Issue the request.
if parsed.scheme != 'http':
logger.error("Can't fetch URI %r - unknown scheme", uri)
return
# Note we *must* use the full uri here and not just the path portion
# or getsatisfaction returns invalid urls...
try:
conn = httplib2.Http()
response, content = conn.request(uri, headers=req_headers)
except Exception:
logger.exception("fetching rss feed %r failed", uri)
return
if response.status == 304:
# yay - nothing to update!
logger.info('rss feed %r is up-to-date', uri)
response.close()
return
if response.status != 200:
logger.exception("bad response fetching rss feed %r: %s (%s)",
uri, response.status, response.reason)
return
logger.debug('rss feed %r has changed', uri)
Used by `gcs.GCSClient` and `bigquery.BigQueryClient` to initiate the API Client
"""
if oauth_credentials:
authenticate_kwargs = {"credentials": oauth_credentials}
elif http_:
authenticate_kwargs = {"http": http_}
else:
# neither http_ or credentials provided
try:
# try default credentials
credentials, _ = google.auth.default()
authenticate_kwargs = {"credentials": credentials}
except google.auth.exceptions.DefaultCredentialsError:
# try http using httplib2
authenticate_kwargs = {"http": httplib2.Http()}
return authenticate_kwargs
def OAuth2Login(client_secrets, credential_store, email):
scope='https://picasaweb.google.com/data/'
user_agent='picasawebuploader'
storage = Storage(credential_store)
credentials = storage.get()
if credentials is None or credentials.invalid:
flow = flow_from_clientsecrets(client_secrets, scope=scope, redirect_uri='urn:ietf:wg:oauth:2.0:oob')
uri = flow.step1_get_authorize_url()
webbrowser.open(uri)
code = raw_input('\nEnter the authentication code, then press Enter: ').strip()
credentials = flow.step2_exchange(code)
if (credentials.token_expiry - datetime.utcnow()) < timedelta(minutes=5):
http = httplib2.Http()
http = credentials.authorize(http)
credentials.refresh(http)
storage.put(credentials)
gd_client = gdata.photos.service.PhotosService(source=user_agent,
email=email,
additional_headers={'Authorization' : 'Bearer %s' % credentials.access_token})
return gd_client