Skip to content

Commit

Permalink
Merge pull request #50 from xsnippet/aiohttp-pytest
Browse files Browse the repository at this point in the history
Use pytest-aiohttp instead of custom helpers
  • Loading branch information
malor committed Dec 24, 2017
2 parents ea5a0b0 + dc0a109 commit b58ce84
Show file tree
Hide file tree
Showing 9 changed files with 940 additions and 1,143 deletions.
3 changes: 2 additions & 1 deletion setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,14 +39,15 @@ def find_packages(namespace):
'pytest-runner',
],
install_requires=[
'aiohttp >= 1.2',
'aiohttp >= 2.3.5',
'cerberus >= 0.9.2',
'motor >= 1.1',
'python-jose >= 1.3.2',
'werkzeug >= 0.11.4',
],
tests_require=[
'pytest >= 2.8.7',
'pytest-aiohttp >= 0.3.0',
],
entry_points={
'console_scripts': [
Expand Down
221 changes: 0 additions & 221 deletions tests/__init__.py

This file was deleted.

147 changes: 70 additions & 77 deletions tests/middlewares/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,86 +15,79 @@
import pytest

from xsnippet.api import middlewares
from tests import AIOTestMeta, AIOTestApp


class TestAuthMiddleware(metaclass=AIOTestMeta):
@pytest.fixture(scope='function')
async def testapp(test_client):
app = web.Application(middlewares=[
functools.partial(middlewares.auth.auth, {'secret': 'SWORDFISH'})
])

conf = {
'auth': {
'secret': 'SWORDFISH'
}
}
async def handler(request):
return web.Response(text='success')
app.router.add_get('/', handler)

def setup(self):
async def handler(request):
return web.Response(text='success')
async def handler(request):
assert request['auth']['user'] == 'john'
return web.Response(text='success')
app.router.add_get('/success', handler)

self.app = web.Application(middlewares=[
functools.partial(middlewares.auth.auth, self.conf['auth'])
])
self.app.router.add_get('/', handler)
async def handler(request):
assert request['auth'] is None
return web.Response(text='success')
app.router.add_get('/success-no-token', handler)

@pytest.mark.parametrize('token', [
'malformed_token_value',
jwt.encode({'valid': 'token'}, 'wrong secret')
])
async def test_unauthorized_is_raised_for_invalid_tokens(self, token):
async with AIOTestApp(self.app) as testapp:
resp = await testapp.get('/', headers={
'Authorization': 'bearer ' + token
})

assert resp.status == 401
assert 'passed token is invalid' in await resp.text()

async def test_authentication_succeeds(self):
async def handler(request):
assert request['auth']['user'] == 'john'
return web.Response(text='success')
self.app.router.add_get('/test_auth', handler)

async with AIOTestApp(self.app) as testapp:
token = jwt.encode({'user': 'john'}, 'SWORDFISH')
resp = await testapp.get('/test_auth', headers={
'Authorization': 'bearer ' + token
})

async with resp:
assert resp.status == 200

async def test_authentication_token_not_passed(self):
async def handler(request):
assert request['auth'] is None
return web.Response(text='success')
self.app.router.add_get('/test_auth', handler)

async with AIOTestApp(self.app) as testapp:
resp = await testapp.get('/test_auth')

async with resp:
assert resp.status == 200

async def test_unauthorized_is_raised_for_invalid_type(self):
token = jwt.encode({'user': 'john'}, 'SWORDFISH')

async with AIOTestApp(self.app) as testapp:
resp = await testapp.get('/', headers={
'Authorization': 'token ' + token
})

assert resp.status == 401
assert 'Unsupported auth type' in await resp.text()

@pytest.mark.parametrize('header, error', [
('bearer', 'Token missing'),
('bearer t oken', 'Token contains spaces'),
])
async def test_unauthorized_for_invalid_header(self, header, error):
async with AIOTestApp(self.app) as testapp:
resp = await testapp.get('/', headers={
'Authorization': header
})

assert resp.status == 401
assert error in await resp.text()
return await test_client(app)


@pytest.mark.parametrize('token', [
'malformed_token_value',
jwt.encode({'valid': 'token'}, 'wrong secret')
])
async def test_unauthorized_is_raised_for_invalid_tokens(token, testapp):
resp = await testapp.get('/', headers={
'Authorization': 'bearer ' + token
})

assert resp.status == 401
assert 'passed token is invalid' in await resp.text()


async def test_authentication_succeeds(testapp):
token = jwt.encode({'user': 'john'}, 'SWORDFISH')
resp = await testapp.get('/success', headers={
'Authorization': 'bearer ' + token
})

async with resp:
assert resp.status == 200


async def test_authentication_token_not_passed(testapp):
resp = await testapp.get('/success-no-token')

async with resp:
assert resp.status == 200


async def test_unauthorized_is_raised_for_invalid_type(testapp):
token = jwt.encode({'user': 'john'}, 'SWORDFISH')
resp = await testapp.get('/', headers={
'Authorization': 'token ' + token
})

assert resp.status == 401
assert 'Unsupported auth type' in await resp.text()


@pytest.mark.parametrize('header, error', [
('bearer', 'Token missing'),
('bearer t oken', 'Token contains spaces'),
])
async def test_unauthorized_for_invalid_header(header, error, testapp):
resp = await testapp.get('/', headers={
'Authorization': header
})

assert resp.status == 401
assert error in await resp.text()

0 comments on commit b58ce84

Please sign in to comment.