Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
#!/usr/bin/env python
import sys, os
import types
import py
import time
import re
import pyral
from pyral import Rally
RallyRESTAPIError = pyral.context.RallyRESTAPIError
RallyResponseError = pyral.rallyresp.RallyResponseError
##################################################################################################
from rally_targets import AGICEN, AGICEN_USER, AGICEN_PSWD, HTTPS_PROXY
from rally_targets import PROD, API_KEY
from rally_targets import PROD_USER, PROD_PSWD
##################################################################################################
def test_basic_connection():
"""
Using a known valid Rally server and access credentials, issue a simple query
request against a known valid Rally entity.
"""
rally = Rally(server=AGICEN, user=AGICEN_USER, password=AGICEN_PSWD)
response = rally.get('Project', fetch=False, limit=10)
cgt = CargoTruck(page_urls, num_threads)
try:
cgt.load(self.session, 'get', 15)
payload = cgt.dump()
success = True
break
except:
exc_name, exc_desc = sys.exc_info()[:2]
anomaly = "RallyResponse.next.__retrieveNextPage.__retrievePages caught exception " +\
"in threaded request/response activity: %s, %s" % (exc_name, exc_desc)
pg1, pg2 = (self.startIndex + self.pageSize), (self.startIndex + (self.pageSize*num_threads))
notice = "Retrying the page_urls for the page group startIndexes: %d -> %d" % (pg1, pg2)
print(notice)
if not success:
raise RallyResponseError("Unable to retrieve %d chunks of data" % num_threads)
chapter = []
for chunk in payload:
chapter.extend(chunk.json()['QueryResult']['Results'])
self.startIndex += len(chapter)
return chapter
# or a Rally "sub-type" like PortfolioItem/Feature
# which is context dependent and has a singleton result
target = self.target
if target.endswith('.x'):
target = target[:-2]
##
## print("ImpliedQuery presumed target: |%s|" % target)
## print("")
##
if target not in list(self.content.keys()):
# check to see if there is a case-insensitive match before upchucking...
ckls = [k.lower() for k in list(self.content.keys())]
if target.lower() not in ckls:
forensic_info = "%s\n%s\n" % (response.status_code, response.content)
problem = 'missing _Xx_Result specifier for target %s in following:' % target
raise RallyResponseError('%s\n%s' % (problem, forensic_info))
else:
matching_item_ix = ckls.index(target.lower())
target = list(self.content.keys())[matching_item_ix]
self.target = target
self._stdFormat = False
# fudge in the QueryResult.Results. dict keychain
self._item_type = target
self.data = {'QueryResult': {'Results' : { target: self.content[target] }}}
self.data['Errors'] = self.content[target]['Errors']
self.data['Warnings'] = self.content[target]['Warnings']
del self.content[target]['Errors'] # we just snagged this and repositioned it
del self.content[target]['Warnings'] # ditto
self.data['PageSize'] = 1
self.data['TotalResultCount'] = 1
qr = self.data
## print("RallyRESTResponse.status_code is %s" % self.status_code)
## print(response.content)
## print("x" * 80)
##
if isinstance(response, ErrorResponse):
if 'OperationResult' in response.content:
if 'Errors' in response.content['OperationResult']:
self.errors = response.content['OperationResult']['Errors']
return
self._stdFormat = True
try:
self.content = response.json()
except:
problem = "Response for request: {0} either was not JSON content or was an invalidly formed/incomplete JSON structure".format(self.resource)
raise RallyResponseError(problem)
##
## print("response content: %s" % self.content)
##
self.request_type, self.data = self._determineRequestResponseType(request)
##
## print("RallyRESTResponse request_type: %s for %s" % (self.request_type, self._item_type))
##
if self.request_type == 'ImpliedQuery':
# the request is against a Rally Type name, ie. 'Subscription', 'Workspace', 'UserProfile', etc.
# or a Rally "sub-type" like PortfolioItem/Feature
# which is context dependent and has a singleton result
target = self.target
if target.endswith('.x'):
target = target[:-2]
##