python
/
aiohttp-security
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix tests

This commit is contained in:
Andrew Svetlov 2016-09-27 17:57:11 +03:00
parent aabe9282e7
commit 0a7ec47274
7 changed files with 71 additions and 212 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
requirements-dev.txt
View File

@ -11,3 +11,4 @@ aioredis==0.2.8
hiredis==0.2.0
passlib==1.6.5
aiohttp
pytest-aiohttp

156
tests/conftest.py
View File

@ -1,155 +1 @@
import gc
import socket
import asyncio
import pytest
import aiohttp
from aiohttp import web
@pytest.fixture
def unused_port():
def f():
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('127.0.0.1', 0))
return s.getsockname()[1]
return f
@pytest.yield_fixture
def loop(request):
loop = asyncio.new_event_loop()
asyncio.set_event_loop(None)
yield loop
loop.stop()
loop.run_forever()
loop.close()
gc.collect()
asyncio.set_event_loop(None)
@pytest.yield_fixture
def create_server(loop, unused_port):
app = handler = srv = None
@asyncio.coroutine
def create(*, debug=False, ssl_ctx=None, proto='http'):
nonlocal app, handler, srv
app = web.Application(loop=loop)
port = unused_port()
handler = app.make_handler(debug=debug, keep_alive_on=False)
srv = yield from loop.create_server(handler, '127.0.0.1', port,
ssl=ssl_ctx)
if ssl_ctx:
proto += 's'
url = "{}://127.0.0.1:{}".format(proto, port)
return app, url
yield create
@asyncio.coroutine
def finish():
if handler is not None:
yield from handler.finish_connections()
if app is not None:
yield from app.finish()
if srv is not None:
srv.close()
yield from srv.wait_closed()
loop.run_until_complete(finish())
class Client:
def __init__(self, session, url):
self._session = session
if not url.endswith('/'):
url += '/'
self._url = url
@property
def cookies(self):
return self._session.cookies
def close(self):
self._session.close()
def get(self, path, **kwargs):
while path.startswith('/'):
path = path[1:]
url = self._url + path
resp = self._session.get(url, **kwargs)
return resp
def post(self, path, **kwargs):
while path.startswith('/'):
path = path[1:]
url = self._url + path
return self._session.post(url, **kwargs)
def ws_connect(self, path, **kwargs):
while path.startswith('/'):
path = path[1:]
url = self._url + path
return self._session.ws_connect(url, **kwargs)
@pytest.yield_fixture
def create_app_and_client(create_server, loop):
client = None
cookie_jar = aiohttp.CookieJar(loop=loop, unsafe=True)
@asyncio.coroutine
def maker(*, server_params=None, client_params=None):
nonlocal client
if server_params is None:
server_params = {}
server_params.setdefault('debug', False)
server_params.setdefault('ssl_ctx', None)
app, url = yield from create_server(**server_params)
if client_params is None:
client_params = {}
client = Client(
aiohttp.ClientSession(loop=loop, cookie_jar=cookie_jar),
url
)
return app, client
yield maker
if client is not None:
client.close()
@pytest.mark.tryfirst
def pytest_pycollect_makeitem(collector, name, obj):
if collector.funcnamefilter(name):
if not callable(obj):
return
item = pytest.Function(name, parent=collector)
if 'run_loop' in item.keywords:
return list(collector._genfunctions(name, obj))
@pytest.mark.tryfirst
def pytest_pyfunc_call(pyfuncitem):
"""
Run asyncio marked test functions in an event loop instead of a normal
function call.
"""
if 'run_loop' in pyfuncitem.keywords:
funcargs = pyfuncitem.funcargs
loop = funcargs['loop']
testargs = {arg: funcargs[arg]
for arg in pyfuncitem._fixtureinfo.argnames}
loop.run_until_complete(pyfuncitem.obj(**testargs))
return True
def pytest_runtest_setup(item):
if 'run_loop' in item.keywords and 'loop' not in item.fixturenames:
# inject an event loop fixture for all async tests
item.fixturenames.append('loop')
# nothing to do

31
tests/test_cookies_identity.py
View File

@ -1,5 +1,4 @@
import asyncio
import pytest
from aiohttp import web
from aiohttp_security import (remember, forget,
@ -20,8 +19,8 @@ class Autz(AbstractAuthorizationPolicy):
pass
@pytest.mark.run_loop
def test_remember(create_app_and_client):
@asyncio.coroutine
def test_remember(loop, test_client):
@asyncio.coroutine
def handler(request):
@ -29,17 +28,18 @@ def test_remember(create_app_and_client):
yield from remember(request, response, 'Andrew')
return response
app, client = yield from create_app_and_client()
app = web.Application(loop=loop)
_setup(app, CookiesIdentityPolicy(), Autz())
app.router.add_route('GET', '/', handler)
client = yield from test_client(app)
resp = yield from client.get('/')
assert 200 == resp.status
assert 'Andrew' == resp.cookies['AIOHTTP_SECURITY'].value
yield from resp.release()
@pytest.mark.run_loop
def test_identify(create_app_and_client):
@asyncio.coroutine
def test_identify(loop, test_client):
@asyncio.coroutine
def create(request):
@ -54,10 +54,11 @@ def test_identify(create_app_and_client):
assert 'Andrew' == user_id
return web.Response()
app, client = yield from create_app_and_client()
app = web.Application(loop=loop)
_setup(app, CookiesIdentityPolicy(), Autz())
app.router.add_route('GET', '/', check)
app.router.add_route('POST', '/', create)
client = yield from test_client(app)
resp = yield from client.post('/')
assert 200 == resp.status
yield from resp.release()
@ -66,8 +67,8 @@ def test_identify(create_app_and_client):
yield from resp.release()
@pytest.mark.run_loop
def test_forget(create_app_and_client):
@asyncio.coroutine
def test_forget(loop, test_client):
@asyncio.coroutine
def index(request):
@ -85,19 +86,23 @@ def test_forget(create_app_and_client):
yield from forget(request, response)
return response
app, client = yield from create_app_and_client()
app = web.Application(loop=loop)
_setup(app, CookiesIdentityPolicy(), Autz())
app.router.add_route('GET', '/', index)
app.router.add_route('POST', '/login', login)
app.router.add_route('POST', '/logout', logout)
client = yield from test_client(app)
resp = yield from client.post('/login')
assert 200 == resp.status
assert resp.url.endswith('/')
assert 'Andrew' == client.cookies['AIOHTTP_SECURITY'].value
cookies = client.session.cookie_jar.filter_cookies(
client.make_url('/'))
assert 'Andrew' == cookies['AIOHTTP_SECURITY'].value
yield from resp.release()
resp = yield from client.post('/logout')
assert 200 == resp.status
assert resp.url.endswith('/')
with pytest.raises(KeyError):
_ = client.cookies['AIOHTTP_SECURITY'] # noqa
cookies = client.session.cookie_jar.filter_cookies(
client.make_url('/'))
assert 'AIOHTTP_SECURITY' not in cookies
yield from resp.release()

30
tests/test_dict_autz.py
View File

@ -1,5 +1,4 @@
import asyncio
import pytest
from aiohttp import web
from aiohttp_security import (remember,
@ -26,8 +25,8 @@ class Autz(AbstractAuthorizationPolicy):
return None
@pytest.mark.run_loop
def test_authorized_userid(create_app_and_client):
@asyncio.coroutine
def test_authorized_userid(loop, test_client):
@asyncio.coroutine
def login(request):
@ -41,10 +40,11 @@ def test_authorized_userid(create_app_and_client):
assert 'Andrew' == userid
return web.Response(text=userid)
app, client = yield from create_app_and_client()
app = web.Application(loop=loop)
_setup(app, CookiesIdentityPolicy(), Autz())
app.router.add_route('GET', '/', check)
app.router.add_route('POST', '/login', login)
client = yield from test_client(app)
resp = yield from client.post('/login')
assert 200 == resp.status
@ -53,8 +53,8 @@ def test_authorized_userid(create_app_and_client):
yield from resp.release()
@pytest.mark.run_loop
def test_authorized_userid_not_authorized(create_app_and_client):
@asyncio.coroutine
def test_authorized_userid_not_authorized(loop, test_client):
@asyncio.coroutine
def check(request):
@ -62,16 +62,18 @@ def test_authorized_userid_not_authorized(create_app_and_client):
assert userid is None
return web.Response()
app, client = yield from create_app_and_client()
app = web.Application(loop=loop)
_setup(app, CookiesIdentityPolicy(), Autz())
app.router.add_route('GET', '/', check)
client = yield from test_client(app)
resp = yield from client.get('/')
assert 200 == resp.status
yield from resp.release()
@pytest.mark.run_loop
def test_permits(create_app_and_client):
@asyncio.coroutine
def test_permits(loop, test_client):
@asyncio.coroutine
def login(request):
@ -89,17 +91,18 @@ def test_permits(create_app_and_client):
assert not ret
return web.Response()
app, client = yield from create_app_and_client()
app = web.Application(loop=loop)
_setup(app, CookiesIdentityPolicy(), Autz())
app.router.add_route('GET', '/', check)
app.router.add_route('POST', '/login', login)
client = yield from test_client(app)
resp = yield from client.post('/login')
assert 200 == resp.status
yield from resp.release()
@pytest.mark.run_loop
def test_permits_unauthorized(create_app_and_client):
@asyncio.coroutine
def test_permits_unauthorized(loop, test_client):
@asyncio.coroutine
def check(request):
@ -111,9 +114,10 @@ def test_permits_unauthorized(create_app_and_client):
assert not ret
return web.Response()
app, client = yield from create_app_and_client()
app = web.Application(loop=loop)
_setup(app, CookiesIdentityPolicy(), Autz())
app.router.add_route('GET', '/', check)
client = yield from test_client(app)
resp = yield from client.get('/')
assert 200 == resp.status
yield from resp.release()

15
tests/test_no_auth.py
View File

@ -1,12 +1,11 @@
import asyncio
import pytest
from aiohttp import web
from aiohttp_security import authorized_userid, permits
@pytest.mark.run_loop
def test_authorized_userid(create_app_and_client):
@asyncio.coroutine
def test_authorized_userid(loop, test_client):
@asyncio.coroutine
def check(request):
@ -14,15 +13,16 @@ def test_authorized_userid(create_app_and_client):
assert userid is None
return web.Response()
app, client = yield from create_app_and_client()
app = web.Application(loop=loop)
app.router.add_route('GET', '/', check)
client = yield from test_client(app)
resp = yield from client.get('/')
assert 200 == resp.status
yield from resp.release()
@pytest.mark.run_loop
def test_permits(create_app_and_client):
@asyncio.coroutine
def test_permits(loop, test_client):
@asyncio.coroutine
def check(request):
@ -34,8 +34,9 @@ def test_permits(create_app_and_client):
assert ret
return web.Response()
app, client = yield from create_app_and_client()
app = web.Application(loop=loop)
app.router.add_route('GET', '/', check)
client = yield from test_client(app)
resp = yield from client.get('/')
assert 200 == resp.status
yield from resp.release()

15
tests/test_no_identity.py
View File

@ -1,20 +1,20 @@
import asyncio
import pytest
from aiohttp import web
from aiohttp_security import remember, forget
@pytest.mark.run_loop
def test_remember(create_app_and_client):
@asyncio.coroutine
def test_remember(loop, test_client):
@asyncio.coroutine
def do_remember(request):
response = web.Response()
yield from remember(request, response, 'Andrew')
app, client = yield from create_app_and_client()
app = web.Application(loop=loop)
app.router.add_route('POST', '/', do_remember)
client = yield from test_client(app)
resp = yield from client.post('/')
assert 500 == resp.status
assert (('Security subsystem is not initialized, '
@ -23,16 +23,17 @@ def test_remember(create_app_and_client):
yield from resp.release()
@pytest.mark.run_loop
def test_forget(create_app_and_client):
@asyncio.coroutine
def test_forget(loop, test_client):
@asyncio.coroutine
def do_forget(request):
response = web.Response()
yield from forget(request, response)
app, client = yield from create_app_and_client()
app = web.Application(loop=loop)
app.router.add_route('POST', '/', do_forget)
client = yield from test_client(app)
resp = yield from client.post('/')
assert 500 == resp.status
assert (('Security subsystem is not initialized, '

35
tests/test_session_identity.py
View File

@ -23,18 +23,15 @@ class Autz(AbstractAuthorizationPolicy):
@pytest.fixture
def create_app_and_client2(create_app_and_client):
@asyncio.coroutine
def maker(*args, **kwargs):
app, client = yield from create_app_and_client(*args, **kwargs)
setup_session(app, SimpleCookieStorage())
setup_security(app, SessionIdentityPolicy(), Autz())
return app, client
return maker
def make_app(loop):
app = web.Application(loop=loop)
setup_session(app, SimpleCookieStorage())
setup_security(app, SessionIdentityPolicy(), Autz())
return app
@pytest.mark.run_loop
def test_remember(create_app_and_client2):
@asyncio.coroutine
def test_remember(make_app, test_client):
@asyncio.coroutine
def handler(request):
@ -48,9 +45,10 @@ def test_remember(create_app_and_client2):
assert session['AIOHTTP_SECURITY'] == 'Andrew'
return web.HTTPOk()
app, client = yield from create_app_and_client2()
app = make_app()
app.router.add_route('GET', '/', handler)
app.router.add_route('GET', '/check', check)
client = yield from test_client(app)
resp = yield from client.get('/')
assert 200 == resp.status
yield from resp.release()
@ -60,8 +58,8 @@ def test_remember(create_app_and_client2):
yield from resp.release()
@pytest.mark.run_loop
def test_identify(create_app_and_client2):
@asyncio.coroutine
def test_identify(make_app, test_client):
@asyncio.coroutine
def create(request):
@ -76,9 +74,10 @@ def test_identify(create_app_and_client2):
assert 'Andrew' == user_id
return web.Response()
app, client = yield from create_app_and_client2()
app = make_app()
app.router.add_route('GET', '/', check)
app.router.add_route('POST', '/', create)
client = yield from test_client(app)
resp = yield from client.post('/')
assert 200 == resp.status
yield from resp.release()
@ -88,8 +87,8 @@ def test_identify(create_app_and_client2):
yield from resp.release()
@pytest.mark.run_loop
def test_forget(create_app_and_client2):
@asyncio.coroutine
def test_forget(make_app, test_client):
@asyncio.coroutine
def index(request):
@ -108,11 +107,13 @@ def test_forget(create_app_and_client2):
yield from forget(request, response)
return response
app, client = yield from create_app_and_client2()
app = make_app()
app.router.add_route('GET', '/', index)
app.router.add_route('POST', '/login', login)
app.router.add_route('POST', '/logout', logout)
client = yield from test_client(app)
resp = yield from client.post('/login')
assert 200 == resp.status
assert resp.url.endswith('/')