Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
mock_request.path = path.split('?')[0]
path = mock_request.path
except Exception:
pass
if isinstance(path, bytes):
path = path.decode('utf8')
for (method, pattern, func) in self.callbacks:
if http_method != method:
continue
matcher = pattern.match(path)
if matcher:
try:
args = [urlparse.unquote(u) for u in matcher.groups()]
(code, response) = yield func(mock_request, *args)
defer.returnValue((code, response))
except CodeMessageException as e:
defer.returnValue((e.code, cs_error(e.msg, code=e.errcode)))
raise KeyError("No event can handle %s" % path)
def extract_properties_from_topic(topic):
"""
Topics for responses from DPS are of the following format:
$dps/registrations/res//?$=&=&=
Extract key=value pairs from the latter part of the topic.
:param topic: The topic string
:return key_values_dict : a dictionary of key mapped to a list of values.
"""
topic_parts = topic.split("$")
key_value_dict = urllib.parse.parse_qs(topic_parts[2])
return key_value_dict
"""
Extract authentication (user name and password) credentials from the
given URL.
>>> extract_credentials('http://localhost:5984/_config/')
('http://localhost:5984/_config/', None)
>>> extract_credentials('http://joe:secret@localhost:5984/_config/')
('http://localhost:5984/_config/', ('joe', 'secret'))
>>> extract_credentials('http://joe%40example.com:secret@'
... 'localhost:5984/_config/')
('http://localhost:5984/_config/', ('joe@example.com', 'secret'))
"""
parts = parse.urlsplit(url)
netloc = parts[1]
if '@' in netloc:
creds, netloc = netloc.split('@')
credentials = tuple(parse.unquote(i) for i in creds.split(':'))
parts = list(parts)
parts[1] = netloc
else:
credentials = None
return parse.urlunsplit(parts), credentials
def _get_absolute_url(self, location):
# If 'location' is an absolute URL (includes host), this will be a no-op.
return six.moves.urllib.parse.urljoin(
base=self._scheme_and_host(), url=location)
socket.timeout)
3. Network IO error (e.reason is socket.error)
urllib2.HTTPError
1. HTTP 404, HTTP 500 etc.
CoraidJsonEncodeFailure - bad REST response
"""
# Handle must be simple path, for example:
# /configure
if '?' in handle or '&' in handle:
raise ValueError(_('Invalid REST handle name. Expected path.'))
# Request url includes base ESM url, handle path and optional
# URL params.
rest_url = urlparse.urljoin(self._esm_url, handle)
encoded_url_params = urllib.urlencode(url_params)
if encoded_url_params:
rest_url += '?' + encoded_url_params
if data is None:
json_request = None
else:
json_request = jsonutils.dumps(data)
request = urllib2.Request(rest_url, json_request)
response = self._url_opener.open(request).read()
try:
if not response and allow_empty_response:
reply = {}
else:
def _get_absolute_url(self, location):
# If 'location' is an absolute URL (includes host), this will be a no-op.
return six.moves.urllib.parse.urljoin(
base=self._scheme_and_host(), url=location)
user_agent=None, redirect=None, endpoint_filter=None,
raise_exc=True, log=True, microversion = None,
endpoint_override=None,connect_retries=0,
allow=None, client_name=None, client_version=None,
**kwargs):
headers = kwargs.setdefault('headers', dict())
if microversion:
self._set_microversion_headers(headers, microversion, None, endpoint_filter)
if not urllib.parse.urlparse(url).netloc:
if endpoint_override:
base_url = endpoint_override % {"project_id": self.project_id}
elif endpoint_filter:
base_url = self.get_endpoint(interface=endpoint_filter.interface,
service_type=endpoint_filter.service_type)
if not urllib.parse.urlparse(base_url).netloc:
raise exceptions.EndpointNotFound()
url = '%s/%s' % (base_url.rstrip('/'), url.lstrip('/'))
headers.setdefault("Host", urllib.parse.urlparse(url).netloc)
if self.cert:
kwargs.setdefault('cert', self.cert)
if self.timeout is not None:
kwargs.setdefault('timeout', self.timeout)
if user_agent:
headers['User-Agent'] = user_agent
elif self.user_agent:
user_agent = headers.setdefault('User-Agent', self.user_agent)
else:
# Per RFC 7231 Section 5.5.3, identifiers in a user-agent should be
def get_bing_access_token(self):
params = {
'client_id': 'beets',
'client_secret': self.config['bing_client_secret'],
'scope': "https://api.microsofttranslator.com",
'grant_type': 'client_credentials',
}
oauth_url = 'https://datamarket.accesscontrol.windows.net/v2/OAuth2-13'
oauth_token = json.loads(requests.post(
oauth_url,
data=urllib.parse.urlencode(params)).content)
if 'access_token' in oauth_token:
return "Bearer " + oauth_token['access_token']
else:
self._log.warning(u'Could not get Bing Translate API access token.'
u' Check your "bing_client_secret" password')
def _unquote_domain(domain):
"""Unquoting function for receiving a domain name in a URL.
Domain names tend to have .'s in them. Urllib doesn't quote dots,
but Routes tends to choke on them, so we need an extra level of
by-hand quoting here.
"""
return urllib.parse.unquote(domain).replace('%2E', '.')
def send(self, msg, *args, **kwargs):
config = self.config
params = {
'id': config.account_id,
'pw': config.password,
'snr': config.sender_id,
'dnr': msg.phone_number,
'split': '10',
}
try:
text = msg.text.encode('iso-8859-1')
params['msg'] = text
except UnicodeEncodeError:
params['msg'] = msg.text.encode('utf-16-be').encode('hex')
params['encoding'] = 'ucs'
url = '%s?%s' % (MACH_URL, six.moves.urllib.parse.urlencode(params))
resp = six.moves.urllib.request.urlopen(url, timeout=settings.SMS_GATEWAY_TIMEOUT).read()
self.handle_response(msg, resp)
return resp