Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_auth_ticket(self):
"""
Test ticket authentication.
"""
shark = SocketShark(TEST_CONFIG)
client = MockClient(shark)
session = client.session
await session.on_client_event({'event': 'auth'})
assert client.log.pop() == {
'status': 'error',
'event': 'auth',
'error': c.ERR_NEEDS_TICKET,
}
await session.on_client_event({'event': 'auth', 'method': 'ticket'})
assert client.log.pop() == {
'status': 'error',
'event': 'auth',
'error': c.ERR_NEEDS_TICKET,
}
with aioresponses() as mock:
# Auth endpoint unreachable
await session.on_client_event({
'event': 'auth',
'method': 'ticket',
'ticket': 'the_ticket',
})
async def process(self):
if self.method not in self.auth_config:
raise EventError(c.ERR_AUTH_UNSUPPORTED)
# The only supported method.
assert self.method == 'ticket'
auth_method_config = self.auth_config[self.method]
ticket = self.data.get('ticket')
if not ticket:
raise EventError(c.ERR_NEEDS_TICKET)
auth_url = auth_method_config['validation_url']
auth_fields = auth_method_config['auth_fields']
result = await http_post(self.shark, auth_url, {'ticket': ticket})
if result.get('status') != 'ok':
raise EventError(result.get('error', c.ERR_AUTH_FAILED))
auth_info = {field: result[field] for field in auth_fields}
self.session.auth_info = auth_info
self.session.log.debug('auth info', auth_info=auth_info)
await self.send_ok()
return True