Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
'error': c.ERR_AUTH_REQUIRED,
}
await self._auth_session(session)
with aioresponses() as mock:
# Authorizer is unavailable.
await session.on_client_event({
'event': 'subscribe',
'subscription': 'authorizer.topic',
})
assert client.log.pop() == {
'event': 'subscribe',
'subscription': 'authorizer.topic',
'status': 'error',
'error': c.ERR_SERVICE_UNAVAILABLE,
}
with aioresponses() as mock:
# Mock authorizer
authorizer_url = 'http://auth-service/auth/authorizer/'
mock.post(authorizer_url, payload={
'status': 'error',
})
mock.post(authorizer_url, payload={
'status': 'error',
'error': 'test error',
})
mock.post(authorizer_url, payload={
for n in range(opts['tries']):
if n > 0:
await asyncio.sleep(opts['wait'])
try:
shark.log.debug('http request', url=url, data=data)
async with session.post(url, json=data,
timeout=opts['timeout']) as resp:
resp.raise_for_status()
data = await resp.json()
shark.log.debug('http response', data=data)
return data
except aiohttp.ClientError:
shark.log.exception('unhandled exception in http_post')
except asyncio.TimeoutError:
shark.log.exception('timeout in http_post')
return {'status': 'error', 'error': c.ERR_SERVICE_UNAVAILABLE}