Convert test_cookies_identity into pytest style

tests/conftest.py

@@ -68,6 +68,10 @@ class Client:
             url += '/'
         self._url = url
 
+    @property
+    def cookies(self):
+        return self._session.cookies
+
     def close(self):
         self._session.close()
 

tests/test_cookies_identity.py

@@ -1,11 +1,10 @@
 import asyncio
-import socket
+import pytest
-import unittest
 
-import aiohttp
 from aiohttp import web
-from aiohttp_security import (remember, setup, forget,
+from aiohttp_security import (remember, forget,
                               AbstractAuthorizationPolicy)
+from aiohttp_security import setup as _setup
 from aiohttp_security.cookies_identity import CookiesIdentityPolicy
 from aiohttp_security.api import IDENTITY_KEY
 
@@ -21,124 +20,83 @@ class Autz(AbstractAuthorizationPolicy):
         pass
 
 
-class TestCookiesIdentity(unittest.TestCase):
+@pytest.mark.run_loop
-
+def test_remember(create_app_and_client):
-    def setUp(self):
-        self.loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(None)
-        self.client = aiohttp.ClientSession(loop=self.loop)
-
-    def tearDown(self):
-        self.client.close()
-        self.loop.run_until_complete(self.handler.finish_connections())
-        self.srv.close()
-        self.loop.run_until_complete(self.srv.wait_closed())
-        self.loop.close()
-
-    def find_unused_port(self):
-        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        s.bind(('127.0.0.1', 0))
-        port = s.getsockname()[1]
-        s.close()
-        return port
 
     @asyncio.coroutine
-    def create_server(self):
+    def handler(request):
-        app = web.Application(loop=self.loop)
+        response = web.Response()
-        setup(app, CookiesIdentityPolicy(), Autz())
+        yield from remember(request, response, 'Andrew')
+        return response
 
-        port = self.find_unused_port()
+    app, client = yield from create_app_and_client()
-        self.handler = app.make_handler(
+    _setup(app, CookiesIdentityPolicy(), Autz())
-            debug=False, keep_alive_on=False)
+    app.router.add_route('GET', '/', handler)
-        srv = yield from self.loop.create_server(
+    resp = yield from client.get('/')
-            self.handler, '127.0.0.1', port)
+    assert 200 == resp.status
-        url = "http://127.0.0.1:{}".format(port)
+    assert 'Andrew' == client.cookies['AIOHTTP_SECURITY'].value
-        self.srv = srv
+    yield from resp.release()
-        return app, srv, url
 
-    def test_remember(self):
 
-        @asyncio.coroutine
+@pytest.mark.run_loop
-        def handler(request):
+def test_identify(create_app_and_client):
-            response = web.Response()
-            yield from remember(request, response, 'Andrew')
-            return response
 
-        @asyncio.coroutine
+    @asyncio.coroutine
-        def go():
+    def create(request):
-            app, srv, url = yield from self.create_server()
+        response = web.Response()
-            app.router.add_route('GET', '/', handler)
+        yield from remember(request, response, 'Andrew')
-            resp = yield from self.client.get(url+'/')
+        return response
-            self.assertEqual(200, resp.status)
-            self.assertEqual('Andrew',
-                             self.client.cookies['AIOHTTP_SECURITY'].value)
-            yield from resp.release()
 
-        self.loop.run_until_complete(go())
+    @asyncio.coroutine
+    def check(request):
+        policy = request.app[IDENTITY_KEY]
+        user_id = yield from policy.identify(request)
+        assert 'Andrew' == user_id
+        return web.Response()
 
-    def test_identify(self):
+    app, client = yield from create_app_and_client()
+    _setup(app, CookiesIdentityPolicy(), Autz())
+    app.router.add_route('GET', '/', check)
+    app.router.add_route('POST', '/', create)
+    resp = yield from client.post('/')
+    assert 200 == resp.status
+    yield from resp.release()
+    resp = yield from client.get('/')
+    assert 200 == resp.status
+    yield from resp.release()
 
-        @asyncio.coroutine
-        def create(request):
-            response = web.Response()
-            yield from remember(request, response, 'Andrew')
-            return response
 
-        @asyncio.coroutine
+@pytest.mark.run_loop
-        def check(request):
+def test_forget(create_app_and_client):
-            policy = request.app[IDENTITY_KEY]
-            user_id = yield from policy.identify(request)
-            self.assertEqual('Andrew', user_id)
-            return web.Response()
 
-        @asyncio.coroutine
+    @asyncio.coroutine
-        def go():
+    def index(request):
-            app, srv, url = yield from self.create_server()
+        return web.Response()
-            app.router.add_route('GET', '/', check)
-            app.router.add_route('POST', '/', create)
-            resp = yield from self.client.post(url+'/')
-            self.assertEqual(200, resp.status)
-            yield from resp.release()
-            resp = yield from self.client.get(url+'/')
-            self.assertEqual(200, resp.status)
-            yield from resp.release()
 
-        self.loop.run_until_complete(go())
+    @asyncio.coroutine
+    def login(request):
+        response = web.HTTPFound(location='/')
+        yield from remember(request, response, 'Andrew')
+        return response
 
-    def test_forget(self):
+    @asyncio.coroutine
+    def logout(request):
+        response = web.HTTPFound(location='/')
+        yield from forget(request, response)
+        return response
 
-        @asyncio.coroutine
+    app, client = yield from create_app_and_client()
-        def index(request):
+    _setup(app, CookiesIdentityPolicy(), Autz())
-            return web.Response()
+    app.router.add_route('GET', '/', index)
-
+    app.router.add_route('POST', '/login', login)
-        @asyncio.coroutine
+    app.router.add_route('POST', '/logout', logout)
-        def login(request):
+    resp = yield from client.post('/login')
-            response = web.HTTPFound(location='/')
+    assert 200 == resp.status
-            yield from remember(request, response, 'Andrew')
+    assert resp.url.endswith('/')
-            return response
+    assert 'Andrew' == client.cookies['AIOHTTP_SECURITY'].value
-
+    yield from resp.release()
-        @asyncio.coroutine
+    resp = yield from client.post('/logout')
-        def logout(request):
+    assert 200 == resp.status
-            response = web.HTTPFound(location='/')
+    assert resp.url.endswith('/')
-            yield from forget(request, response)
+    assert '' == client.cookies['AIOHTTP_SECURITY'].value
-            return response
+    yield from resp.release()
-
-        @asyncio.coroutine
-        def go():
-            app, srv, url = yield from self.create_server()
-            app.router.add_route('GET', '/', index)
-            app.router.add_route('POST', '/login', login)
-            app.router.add_route('POST', '/logout', logout)
-            resp = yield from self.client.post(url+'/login')
-            self.assertEqual(200, resp.status)
-            self.assertEqual(url+'/', resp.url)
-            self.assertEqual('Andrew',
-                             self.client.cookies['AIOHTTP_SECURITY'].value)
-            yield from resp.release()
-            resp = yield from self.client.post(url+'/logout')
-            self.assertEqual(200, resp.status)
-            self.assertEqual(url+'/', resp.url)
-            self.assertEqual('', self.client.cookies['AIOHTTP_SECURITY'].value)
-            yield from resp.release()
-
-        self.loop.run_until_complete(go())