Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_credential_client(self):
""" New property added for using LocustEdxRestApiClient.
Default locust client will remain same for using auto_auth().
"""
return LocustEdxRestApiClient(
settings.data['credentials']['url']['api'],
session=HttpSession(base_url=self.locust.host),
jwt=self.get_access_token()
)
access_token_endpoint = '{}/oauth2/access_token'.format(
settings.secrets['oauth']['provider_url'].strip('/')
)
access_token, __ = LocustEdxRestApiClient.get_oauth_access_token(
access_token_endpoint,
settings.secrets['oauth']['client_id'],
settings.secrets['oauth']['client_secret'],
)
api_url = self.host.strip('/')
self.client = LocustEdxRestApiClient(
api_url,
session=HttpSession(base_url=self.host),
oauth_access_token=access_token
)
_userEnvironment.append(up[ENVIRONMENT_INDEX])
# Async lock user to prevent two threads runing with the same user
_userLock.acquire()
self.taskUser = _currentUser
self.taskUserToken = _userTokens[_currentUser]
self.taskUserName = _userNames[_currentUser]
self.taskUserEnvironment = _userEnvironment[_currentUser]
print("Spawning user ["+str(self.taskUser)+"] on ["+self.taskUserEnvironment+"]")
if _currentUser < _users - 1:
_currentUser += 1
else:
_currentUser = 0
_userLock.release()
# User lock released, critical section end
host = "https://che.prod-preview.openshift.io" if "prod-preview" in self.taskUserEnvironment else "https://che.openshift.io"
self.client = HttpSession(base_url=host)
def get_kv_version(client: requests.Session=None, host: str=None, token: str=None) -> int:
if isinstance(client, HttpSession):
s = client
r = s.get('/v1/sys/mounts')
elif isinstance(client, requests.Session):
s = client
r = s.get(f'{host}/v1/sys/mounts')
else:
s = requests.Session()
s.headers = {'X-Vault-Token': token}
r = s.get(f'{host}/v1/sys/mounts')
r.raise_for_status()
version = 1
for key, val in r.json().items():
if key == 'secret/':
if 'options' in val:
def __init__(self):
super(ProgramsUser, self).__init__()
if not self.host:
raise LocustError(
'You must specify a base host, either in the host attribute in the Locust class, '
'or on the command line using the --host option.'
)
self.client = LocustEdxRestApiClient(
PROGRAMS_API_URL,
session=HttpSession(base_url=self.host),
jwt=self._get_token()
)
def setup(self):
# Create a fleet.
client = clients.HttpSession(base_url=self.host)
self.create_fleet(client, FLEET_NAME, FLEET_SIZE)
def __init__(self, base_url, *args, **kwargs):
super(HttpSession, self).__init__(*args, **kwargs)
self.base_url = base_url
# Check for basic authentication
parsed_url = urlparse(self.base_url)
if parsed_url.username and parsed_url.password:
netloc = parsed_url.hostname
if parsed_url.port:
netloc += ":%d" % parsed_url.port
# remove username and password from the base_url
self.base_url = urlunparse((parsed_url.scheme, netloc, parsed_url.path, parsed_url.params, parsed_url.query, parsed_url.fragment))
# configure requests to use basic auth
self.auth = HTTPBasicAuth(parsed_url.username, parsed_url.password)
def __init__(self, *args, **kwargs):
super(MultipleHostsLocust, self).__init__(*args, **kwargs)
self.api_client = HttpSession(base_url=os.environ["API_HOST"])
def __init__(self):
super(HttpLocust, self).__init__()
if self.host is None:
raise LocustError("You must specify the base host. Either in the host attribute in the Locust class, or on the command line using the --host option.")
session = HttpSession(base_url=self.host)
session.trust_env = self.trust_env
self.client = session