Select Git revision
-
Moul authored
Migrate from [tool.poetry] to [project] https://python-poetry.org/blog/announcing-poetry-2.0.0 https://python-poetry.org/docs/pyproject/ https://python-poetry.org/docs/faq/#migrate-pep621-new-static Update website, changelog urls (#433) Use dynamic Poetry classifiers and dependencies, more convenient but less generic to pyproject.toml project Change classifier to "Operating System :: Unix", since Silkaj is not supported on Windows Move pydiscourse to optional-dependencies, since Poetry extra is deprecated Set main authors, with full names, sorted by importance of their contributions Switch back pip container build to install Poetry v2
Moul authoredMigrate from [tool.poetry] to [project] https://python-poetry.org/blog/announcing-poetry-2.0.0 https://python-poetry.org/docs/pyproject/ https://python-poetry.org/docs/faq/#migrate-pep621-new-static Update website, changelog urls (#433) Use dynamic Poetry classifiers and dependencies, more convenient but less generic to pyproject.toml project Change classifier to "Operating System :: Unix", since Silkaj is not supported on Windows Move pydiscourse to optional-dependencies, since Poetry extra is deprecated Set main authors, with full names, sorted by importance of their contributions Switch back pip container build to install Poetry v2
webserver.py 1.79 KiB
import asyncio
import socket
import ssl
from typing import Tuple, Callable, Optional
from aiohttp import web
# Thanks to aiohttp tests courtesy
# Here is a nice mocking server
class WebFunctionalSetupMixin:
def setUp(self) -> None:
self.handler = None
self.app = web.Application()
self.runner = web.AppRunner(self.app)
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
def tearDown(self) -> None:
if self.handler:
self.loop.run_until_complete(self.handler.shutdown())
if self.runner:
self.loop.run_until_complete(self.runner.cleanup())
try:
self.loop.stop()
self.loop.close()
finally:
asyncio.set_event_loop(None)
def find_unused_port(self) -> int:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(('127.0.0.1', 0))
port = s.getsockname()[1]
s.close()
return port
async def create_server(self, method: str, path: str, handler: Optional[Callable] = None,
ssl_ctx: Optional[ssl.SSLContext] = None) -> Tuple[web.Application, int, str]:
"""
Create a web server for tests
:param method: HTTP method type
:param path: Url path
:param handler: Callback function
:param ssl_ctx: SSL context (https is used if not None)
:return:
"""
if handler:
self.app.router.add_route(method, path, handler)
await self.runner.setup()
port = self.find_unused_port()
site = web.TCPSite(self.runner, '127.0.0.1', port)
await site.start()
protocol = "https" if ssl_ctx else "http"
url = "{}://127.0.0.1:{}".format(protocol, port) + path
return self.app, port, url