From 7addb68468c0ff0e09d7188b53e4552908508639 Mon Sep 17 00:00:00 2001 From: RomanZhukov Date: Sun, 2 Feb 2025 00:56:23 +0500 Subject: [PATCH] moved ssl tests to test_runserver_main.py --- tests/test_runserver_main.py | 88 +++++++++++++++++++++++++ tests/test_runserver_with_ssl.py | 107 ------------------------------- 2 files changed, 88 insertions(+), 107 deletions(-) delete mode 100644 tests/test_runserver_with_ssl.py diff --git a/tests/test_runserver_main.py b/tests/test_runserver_main.py index a2f4602f..9bb7a5ce 100644 --- a/tests/test_runserver_main.py +++ b/tests/test_runserver_main.py @@ -14,6 +14,7 @@ from aiohttp_devtools.runserver.serve import ( WS, create_auxiliary_app, create_main_app, modify_main_app, src_reload, start_main_app) from aiohttp_devtools.runserver.watch import AppTask +import ssl from .conftest import SIMPLE_APP, forked @@ -308,3 +309,90 @@ async def test_websocket_reload(aux_cli): assert reloads == 1 finally: await ws.close() + + +async def check_ssl_server_running(check_callback, sslcontext): + port_open = False + async with aiohttp.ClientSession(timeout=ClientTimeout(total=1)) as session: + for i in range(50): # pragma: no branch + try: + async with session.get('https://localhost:8443/', ssl=sslcontext): + pass + except OSError: + await asyncio.sleep(0.1) + else: + port_open = True + break + assert port_open + await check_callback(session, sslcontext) + await asyncio.sleep(.25) # TODO(aiohttp 4): Remove this hack + + +@pytest.mark.filterwarnings(r"ignore:unclosed:ResourceWarning") +@pytest.mark.datafiles('tests/test_certs', keep_top_dir=True) +def test_start_runserver_ssl(datafiles, tmpworkdir, smart_caplog): + mktree(tmpworkdir, { + 'app.py': """\ +from aiohttp import web +import ssl +async def hello(request): + return web.Response(text='

hello world

', content_type='text/html') + +async def has_error(request): + raise ValueError() + +def create_app(): + app = web.Application() + app.router.add_get('/', hello) + app.router.add_get('/error', has_error) + return app + +def get_ssl_context(): + ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) + ssl_context.load_cert_chain('test_certs/server.crt', 'test_certs/server.key') + return ssl_context + """, + 'static_dir/foo.js': 'var bar=1;', + }) + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + args = runserver(app_path="app.py", static_path="static_dir", + bind_address="0.0.0.0", ssl_context_factory_name='get_ssl_context') + aux_app = args["app"] + aux_port = args["port"] + runapp_host = args["host"] + assert isinstance(aux_app, aiohttp.web.Application) + assert aux_port == 8001 + assert runapp_host == "0.0.0.0" + for startup in aux_app.on_startup: + loop.run_until_complete(startup(aux_app)) + + ssl_context = ssl.create_default_context() + ssl_context.load_verify_locations('test_certs/rootCA.pem') + + async def check_callback(session, ssl_context): + print(session, ssl_context) + async with session.get('https://localhost:8443/', ssl=ssl_context) as r: + assert r.status == 200 + assert r.headers['content-type'].startswith('text/html') + text = await r.text() + print(text) + assert '

hello world

' in text + assert '' in text + + async with session.get('https://localhost:8443/error', ssl=ssl_context) as r: + assert r.status == 500 + assert 'raise ValueError()' in (await r.text()) + + try: + loop.run_until_complete(check_ssl_server_running(check_callback, ssl_context)) + finally: + for shutdown in aux_app.on_shutdown: + loop.run_until_complete(shutdown(aux_app)) + loop.run_until_complete(aux_app.cleanup()) + assert ( + 'adev.server.dft INFO: Starting aux server at http://localhost:8001 ◆\n' + 'adev.server.dft INFO: serving static files from ./static_dir/ at http://localhost:8001/static/\n' + 'adev.server.dft INFO: Starting dev server at https://localhost:8443 ●\n' + ) in smart_caplog + loop.run_until_complete(asyncio.sleep(.25)) # TODO(aiohttp 4): Remove this hack diff --git a/tests/test_runserver_with_ssl.py b/tests/test_runserver_with_ssl.py deleted file mode 100644 index d47dd59c..00000000 --- a/tests/test_runserver_with_ssl.py +++ /dev/null @@ -1,107 +0,0 @@ -import asyncio - -import aiohttp -import pytest -from aiohttp import ClientTimeout -from pytest_toolbox import mktree -import ssl - -from aiohttp_devtools.runserver import runserver -from aiohttp_devtools.runserver.config import Config - -from aiohttp_devtools.exceptions import AiohttpDevConfigError - - -async def test_load_invalid_app(tmpworkdir): - mktree(tmpworkdir, { - 'invalid': "it's not python file)" - }) - with pytest.raises(AiohttpDevConfigError): - Config(app_path='invalid') - - -async def check_server_running(check_callback, sslcontext): - port_open = False - async with aiohttp.ClientSession(timeout=ClientTimeout(total=1)) as session: - for i in range(50): # pragma: no branch - try: - async with session.get('https://localhost:8443/', ssl=sslcontext): - pass - except OSError: - await asyncio.sleep(0.1) - else: - port_open = True - break - assert port_open - await check_callback(session, sslcontext) - await asyncio.sleep(.25) # TODO(aiohttp 4): Remove this hack - - -@pytest.mark.filterwarnings(r"ignore:unclosed:ResourceWarning") -@pytest.mark.datafiles('tests/test_certs', keep_top_dir=True) -def test_start_runserver_ssl(datafiles, tmpworkdir, smart_caplog): - mktree(tmpworkdir, { - 'app.py': """\ -from aiohttp import web -import ssl -async def hello(request): - return web.Response(text='

hello world

', content_type='text/html') - -async def has_error(request): - raise ValueError() - -def create_app(): - app = web.Application() - app.router.add_get('/', hello) - app.router.add_get('/error', has_error) - return app - -def get_ssl_context(): - ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) - ssl_context.load_cert_chain('test_certs/server.crt', 'test_certs/server.key') - return ssl_context - """, - 'static_dir/foo.js': 'var bar=1;', - }) - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - args = runserver(app_path="app.py", static_path="static_dir", - bind_address="0.0.0.0", ssl_context_factory_name='get_ssl_context') - aux_app = args["app"] - aux_port = args["port"] - runapp_host = args["host"] - assert isinstance(aux_app, aiohttp.web.Application) - assert aux_port == 8001 - assert runapp_host == "0.0.0.0" - for startup in aux_app.on_startup: - loop.run_until_complete(startup(aux_app)) - - ssl_context = ssl.create_default_context() - ssl_context.load_verify_locations('test_certs/rootCA.pem') - - async def check_callback(session, ssl_context): - print(session, ssl_context) - async with session.get('https://localhost:8443/', ssl=ssl_context) as r: - assert r.status == 200 - assert r.headers['content-type'].startswith('text/html') - text = await r.text() - print(text) - assert '

hello world

' in text - assert '' in text - - async with session.get('https://localhost:8443/error', ssl=ssl_context) as r: - assert r.status == 500 - assert 'raise ValueError()' in (await r.text()) - - try: - loop.run_until_complete(check_server_running(check_callback, ssl_context)) - finally: - for shutdown in aux_app.on_shutdown: - loop.run_until_complete(shutdown(aux_app)) - loop.run_until_complete(aux_app.cleanup()) - assert ( - 'adev.server.dft INFO: Starting aux server at http://localhost:8001 ◆\n' - 'adev.server.dft INFO: serving static files from ./static_dir/ at http://localhost:8001/static/\n' - 'adev.server.dft INFO: Starting dev server at https://localhost:8443 ●\n' - ) in smart_caplog - loop.run_until_complete(asyncio.sleep(.25)) # TODO(aiohttp 4): Remove this hack