from __future__ import division, print_function, absolute_import
import unittest
import datetime
import os
import asyncio
import base
import taskcluster.aio.auth as subjectAsync
@unittest.skipIf(os.environ.get('NO_TESTS_OVER_WIRE'), "Skipping tests over wire")
class TestAuthenticationAsync(base.TCTest):
def test_async_works_with_permanent_credentials(self):
loop = asyncio.get_event_loop()
async def x():
async with subjectAsync.createSession(loop=loop) as session:
client = subjectAsync.Auth({
'rootUrl': self.real_root_url,
'credentials': {
'clientId': 'tester',
'accessToken': 'no-secret',
},
}, session=session)
result = await client.testAuthenticate({
'clientScopes': ['test:a'],
'requiredScopes': ['test:a'],
})
self.assertEqual(result, {'scopes': ['test:a'], 'clientId': 'tester'})
loop.run_until_complete(x())
def test_async_works_with_temporary_credentials(self):
loop = asyncio.get_event_loop()
async def x():
async with subjectAsync.createSession(loop=loop) as session:
tempCred = subjectAsync.createTemporaryCredentials(
'tester',
'no-secret',
datetime.datetime.utcnow(),
datetime.datetime.utcnow() + datetime.timedelta(hours=1),
['test:xyz'],
)
client = subjectAsync.Auth({
'rootUrl': self.real_root_url,
'credentials': tempCred,
}, session=session)
result = await client.testAuthenticate({
'clientScopes': ['test:*'],
'requiredScopes': ['test:xyz'],
})
self.assertEqual(result, {'scopes': ['test:xyz'], 'clientId': 'tester'})
loop.run_until_complete(x())