use the framework's own unit tests

test/aiohttp_test.py

@@ -1,33 +1,61 @@
-from multiprocessing import Process
-
 import pytest
-from common import kwargs_list
-from common import mode_list
-from common import send_requests
+from aiohttp import web
+
+from swagger_ui import aiohttp_api_doc
+from swagger_ui import api_doc
 
+from .common import config_content
+from .common import parametrize_list
 
-def server_process(port, mode, **kwargs):
-    from aiohttp import web
 
+@pytest.fixture
+def app():
     async def hello(request):
         return web.Response(text="Hello, world")
 
     app = web.Application()
     app.add_routes([web.get('/hello/world', hello)])
+    return app
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('mode, kwargs', parametrize_list)
+async def test_aiohttp(app, aiohttp_client, mode, kwargs):
+    if kwargs.get('config_rel_url'):
+        async def swagger_config_handler(request):
+            return web.Response(text=config_content)
+        app.add_routes([web.get(kwargs['config_rel_url'], swagger_config_handler)])
 
     if mode == 'auto':
-        from swagger_ui import api_doc
         api_doc(app, **kwargs)
     else:
-        from swagger_ui import aiohttp_api_doc
         aiohttp_api_doc(app, **kwargs)
-    web.run_app(app, host='localhost', port=port)
 
+    url_prefix = kwargs['url_prefix']
+    if url_prefix.endswith('/'):
+        url_prefix = url_prefix[:-1]
+
+    client = await aiohttp_client(app)
+
+    resp = await client.get('/hello/world')
+    assert resp.status == 200, await resp.text()
 
-@pytest.mark.parametrize('mode', mode_list)
-@pytest.mark.parametrize('kwargs', kwargs_list)
-def test_aiohttp(port, mode, kwargs):
-    proc = Process(target=server_process, args=(port, mode), kwargs=kwargs)
-    proc.start()
-    send_requests(port, mode, kwargs)
-    proc.terminate()
+    resp = await client.get(url_prefix)
+    assert resp.status == 200, await resp.text()
+
+    resp = await client.get(f'{url_prefix}/static/LICENSE')
+    assert resp.status == 200, await resp.text()
+
+    if kwargs.get('editor'):
+        resp = await client.get(f'{url_prefix}/editor')
+        assert resp.status == 200, await resp.text()
+    else:
+        resp = await client.get(f'{url_prefix}/editor')
+        assert resp.status == 404, await resp.text()
+
+    if kwargs.get('config_rel_url'):
+        resp = await client.get(kwargs['config_rel_url'])
+        assert resp.status == 200, await resp.text()
+    else:
+        resp = await client.get(f'{url_prefix}/swagger.json')
+        assert resp.status == 200, await resp.text()

test/bottle_test.py

@@ -1,34 +1,64 @@
-from multiprocessing import Process
-
 import pytest
-from common import kwargs_list
-from common import mode_list
-from common import send_requests
+from bottle import Bottle
+from webtest import AppError
+from webtest import TestApp
+
+from swagger_ui import api_doc
+from swagger_ui import bottle_api_doc
 
+from .common import config_content
+from .common import parametrize_list
 
-def server_process(port, mode, **kwargs):
-    from bottle import Bottle
-    from bottle import run
 
+@pytest.fixture
+def app():
     app = Bottle()
 
     @app.route('/hello/world')
     def hello():
         return 'Hello World!!!'
+    return app
+
+
+@pytest.mark.parametrize('mode, kwargs', parametrize_list)
+def test_bottle(app, mode, kwargs):
+    if kwargs.get('config_rel_url'):
+        @app.route(kwargs['config_rel_url'])
+        def swagger_config_handler():
+            return config_content
 
     if mode == 'auto':
-        from swagger_ui import api_doc
         api_doc(app, **kwargs)
     else:
-        from swagger_ui import bottle_api_doc
         bottle_api_doc(app, **kwargs)
-    run(app, host='localhost', port=port)
 
+    url_prefix = kwargs['url_prefix']
+    if url_prefix.endswith('/'):
+        url_prefix = url_prefix[:-1]
+
+    client = TestApp(app)
+
+    resp = client.get('/hello/world')
+    assert resp.status_code == 200, resp.data
 
-@pytest.mark.parametrize('mode', mode_list)
-@pytest.mark.parametrize('kwargs', kwargs_list)
-def test_bottle(port, mode, kwargs):
-    proc = Process(target=server_process, args=(port, mode), kwargs=kwargs)
-    proc.start()
-    send_requests(port, mode, kwargs)
-    proc.terminate()
+    resp = client.get(url_prefix)
+    assert resp.status_code == 200, resp.data
+
+    resp = client.get(f'{url_prefix}/static/LICENSE')
+    assert resp.status_code == 200, resp.data
+
+    if kwargs.get('editor'):
+        resp = client.get(f'{url_prefix}/editor')
+        assert resp.status_code == 200, resp.data
+    else:
+        try:
+            resp = client.get(f'{url_prefix}/editor')
+        except AppError as ex:
+            assert '404 Not Found' in str(ex), str(ex)
+
+    if kwargs.get('config_rel_url'):
+        resp = client.get(kwargs['config_rel_url'])
+        assert resp.status_code == 200, resp.data
+    else:
+        resp = client.get(f'{url_prefix}/swagger.json')
+        assert resp.status_code == 200, resp.data

test/chalice_test.py

@@ -1,40 +1,79 @@
-from multiprocessing import Process
-
 import pytest
-from common import kwargs_list
-from common import mode_list
-from common import send_requests
+from chalice import Chalice
+from chalice.config import Config
+from chalice.local import ForbiddenError
+from chalice.local import LocalGateway
+
+from swagger_ui import api_doc
+from swagger_ui import chalice_api_doc
 
+from .common import config_content
+from .common import parametrize_list
 
-def server_process(port, mode, **kwargs):
-    from chalice import Chalice
-    from chalice.config import Config
-    from chalice.local import create_local_server
 
+@pytest.fixture
+def app():
     app = Chalice(__name__)
 
     @app.route('/hello/world')
-    def index():
-        return {'hello': 'world'}
+    def hello():
+        return 'Hello World!!!'
+    return app
+
+
+@pytest.mark.parametrize('mode, kwargs', parametrize_list)
+def test_chalice(app, mode, kwargs):
+    if kwargs.get('config_rel_url'):
+        @app.route(kwargs['config_rel_url'])
+        def swagger_config_handler():
+            return config_content
 
     if mode == 'auto':
-        from swagger_ui import api_doc
         api_doc(app, **kwargs)
     else:
-        from swagger_ui import chalice_api_doc
         chalice_api_doc(app, **kwargs)
 
-    config = Config()
-    create_local_server(app, config, 'localhost', port).serve_forever()
+    url_prefix = kwargs['url_prefix']
+    if url_prefix.endswith('/'):
+        url_prefix = url_prefix[:-1]
+
+    client = LocalGateway(app, config=Config())
 
+    headers = {'Host': 'localhost'}
 
-@pytest.mark.parametrize('mode', mode_list)
-@pytest.mark.parametrize('kwargs', kwargs_list)
-def test_chalice(port, mode, kwargs):
-    if kwargs['url_prefix'] in ('/', ''):
-        return
+    resp = client.handle_request(
+        method='GET', path='/hello/world', headers=headers, body=None)
+    assert resp['statusCode'] == 200, resp['body']
 
-    proc = Process(target=server_process, args=(port, mode), kwargs=kwargs)
-    proc.start()
-    send_requests(port, mode, kwargs)
-    proc.terminate()
+    resp = client.handle_request(
+        method='GET', path=url_prefix, headers=headers, body=None)
+    assert resp['statusCode'] == 200, resp['body']
+
+    resp = client.handle_request(
+        method='GET', path=f'{url_prefix}/static/LICENSE',
+        headers=headers, body=None)
+    assert resp['statusCode'] == 200, resp['body']
+
+    if kwargs.get('editor'):
+        resp = client.handle_request(
+            method='GET', path=f'{url_prefix}/editor',
+            headers=headers, body=None)
+        assert resp['statusCode'] == 200, resp['body']
+    else:
+        try:
+            resp = client.handle_request(
+                method='GET', path=f'{url_prefix}/editor',
+                headers=headers, body=None)
+        except ForbiddenError as ex:
+            assert "Missing Authentication Token" in str(ex), str(ex)
+
+    if kwargs.get('config_rel_url'):
+        resp = client.handle_request(
+            method='GET', path=kwargs['config_rel_url'],
+            headers=headers, body=None)
+        assert resp['statusCode'] == 200, resp['body']
+    else:
+        resp = client.handle_request(
+            method='GET', path=f'{url_prefix}/swagger.json',
+            headers=headers, body=None)
+        assert resp['statusCode'] == 200, resp['body']

test/common.py

@@ -1,12 +1,10 @@
-import socket
-import time
+from itertools import product
 from pathlib import Path
 
-import requests
-
 cur_dir = Path(__file__).resolve().parent
 
 config_path = str(cur_dir.joinpath('conf/test3.yaml'))
+config_content = Path(config_path).read_text()
 
 mode_list = ['auto', None]
 
@@ -35,48 +33,4 @@
     },
 ]
 
-
-def detect_port(port):
-    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    try:
-        s.connect(('localhost', int(port)))
-        s.shutdown(2)
-        return True
-    except Exception:
-        return False
-
-
-def wait_port_listen(port):
-    counter = 0
-    while counter <= 50:
-        if detect_port(port):
-            return True
-        time.sleep(0.1)
-        counter += 1
-    return False
-
-
-def send_requests(port, mode, kwargs):
-    assert wait_port_listen(port), 'port: {} not listen!'.format(port)
-
-    url_prefix = 'http://localhost:{}{}'.format(port, kwargs['url_prefix'])
-    if url_prefix.endswith('/'):
-        url_prefix = url_prefix[:-1]
-    server_url = 'http://localhost:{}/hello/world'.format(port)
-
-    # Step 1: test server
-    assert requests.get(server_url).status_code == 200
-
-    # Step 2: test root
-    assert requests.get(url_prefix).status_code == 200
-    assert requests.get(url_prefix + '/').status_code == 200
-
-    # Step 3: test static file
-    assert requests.get(url_prefix + '/static/LICENSE').status_code == 200
-
-    # Step 4: test editor
-    if kwargs.get('editor', False):
-        assert requests.get(url_prefix + '/editor').status_code == 200
-
-    # Step 5: test swagger.json
-    assert requests.get(url_prefix + '/swagger.json').status_code == 200
+parametrize_list = list(product(mode_list, kwargs_list))

test/conftest.py

@@ -1,12 +0,0 @@
-import socket
-
-import pytest
-
-
-@pytest.fixture(scope='function')
-def port():
-    sock = socket.socket()
-    sock.bind(('', 0))
-    ip, port = sock.getsockname()
-    sock.close()
-    return port