Use asynctest built-in loop

Author: Ben Picolo

kubernetes_asyncio/config/kube_config_test.py

@@ -14,14 +14,13 @@
 
 import base64
 import datetime
-import json
 import os
 import shutil
 import tempfile
 from types import SimpleNamespace
 
 import yaml
-from asynctest import Mock, PropertyMock, TestCase, main, patch
+from asynctest import Mock, TestCase, main, patch
 from six import PY3
 
 from .config_exception import ConfigException

kubernetes_asyncio/config/openid_test.py

@@ -1,10 +1,12 @@
+import asyncio
 import json
+from contextlib import contextmanager
 
-import pytest
 from aiohttp import web
-from aiohttp.test_utils import TestClient
-from aiohttp.test_utils import TestServer
+from aiohttp.test_utils import TestClient as _TestClient
+from aiohttp.test_utils import TestServer as _TestServer
 from asynctest import patch
+from asynctest import TestCase
 
 from .config_exception import ConfigException
 from .openid import OpenIDRequestor
@@ -27,77 +29,72 @@ def respond_json(data):
     )
 
 
-@pytest.fixture
-def requestor():
-    return OpenIDRequestor(
-        'client-id',
-        'client-secret',
-        '',
-    )
-
+@contextmanager
+def working_client():
+    loop = asyncio.get_event_loop()
 
-@pytest.yield_fixture
-def working_client(loop, aiohttp_client):
     app = web.Application()
 
     app.router.add_get('/.well-known/openid-configuration', respond_json({'token_endpoint': '/token'}))
     app.router.add_post('/token', respond_json({'id-token': 'id-token-data', 'refresh-token': 'refresh-token-data'}))
 
-    with patch('kubernetes_asyncio.config.openid.OpenIDRequestor._client_session') as _client_session:
-        client = TestClient(TestServer(app, loop=loop), loop=loop)
+    with patch('kubernetes_asyncio.config.openid.aiohttp.ClientSession') as _client_session:
+        client = _TestClient(_TestServer(app, loop=loop), loop=loop)
         _client_session.return_value = client
-        loop.run_until_complete(client.start_server())
 
         yield client
 
-        loop.run_until_complete(client.close())
 
-
-@pytest.yield_fixture
-def fail_well_known_client(loop, aiohttp_client):
+@contextmanager
+def fail_well_known_client():
+    loop = asyncio.get_event_loop()
     app = web.Application()
 
     app.router.add_get('/.well-known/openid-configuration', make_responder(web.Response(status=500)))
 
-    with patch('kubernetes_asyncio.config.openid.OpenIDRequestor._client_session') as _client_session:
-        client = TestClient(TestServer(app, loop=loop), loop=loop)
+    with patch('kubernetes_asyncio.config.openid.aiohttp.ClientSession') as _client_session:
+        client = _TestClient(_TestServer(app, loop=loop), loop=loop)
         _client_session.return_value = client
-        loop.run_until_complete(client.start_server())
-
         yield client
 
-        loop.run_until_complete(client.close())
-
 
-@pytest.yield_fixture
-def fail_token_request_client(loop, aiohttp_client):
+@contextmanager
+def fail_token_request_client():
+    loop = asyncio.get_event_loop()
     app = web.Application()
 
     app.router.add_get('/.well-known/openid-configuration', respond_json({'token_endpoint': '/token'}))
     app.router.add_post('/token', make_responder(web.Response(status=500)))
 
     with patch('kubernetes_asyncio.config.openid.aiohttp.ClientSession') as _client_session:
-        client = TestClient(TestServer(app, loop=loop), loop=loop)
+        client = _TestClient(_TestServer(app, loop=loop), loop=loop)
         _client_session.return_value = client
-        loop.run_until_complete(client.start_server())
 
         yield client
 
-        loop.run_until_complete(client.close())
 
+class OpenIDRequestorTest(TestCase):
 
-async def test_refresh_token_success(requestor, working_client):
-    resp = await requestor.refresh_token('my-refresh-token')
-
-    assert resp['id-token'] == 'id-token-data'
-    assert resp['refresh-token'] == 'refresh-token-data'
+    def setUp(self):
+        self.requestor = OpenIDRequestor(
+            'client-id',
+            'client-secret',
+            '',
+        )
 
+    async def test_refresh_token_success(self):
+        with working_client():
+            resp = await self.requestor.refresh_token('my-refresh-token')
 
-async def test_failed_well_known(requestor, fail_well_known_client):
-    with pytest.raises(ConfigException):
-        await requestor.refresh_token('my-refresh-token')
+            assert resp['id-token'] == 'id-token-data'
+            assert resp['refresh-token'] == 'refresh-token-data'
 
+    async def test_failed_well_known(self):
+        with fail_well_known_client():
+            with self.assertRaises(ConfigException):
+                await self.requestor.refresh_token('my-refresh-token')
 
-async def test_failed_refresh_token(requestor, fail_token_request_client):
-    with pytest.raises(ConfigException):
-        await requestor.refresh_token('my-refresh-token')
+    async def test_failed_refresh_token(self):
+        with fail_token_request_client():
+            with self.assertRaises(ConfigException):
+                await self.requestor.refresh_token('my-refresh-token')

test-requirements.txt

@@ -8,7 +8,6 @@ nose>=1.3.7
 pluggy>=0.3.1
 py>=1.4.31
 pytest
-pytest-aiohttp
 pytest-xdist
 randomize>=0.13
 recommonmark

tox.ini

@@ -9,7 +9,7 @@ deps = -r{toxinidir}/test-requirements.txt
        -r{toxinidir}/requirements.txt
 commands =
    python -V
-   py.test -vvv -s --ignore=kubernetes/e2e_test {posargs}
+   py.test -vvv -s --ignore=kubernetes/e2e_test
 
 [testenv:docs]
 commands =