30 lines
819 B
Python
30 lines
819 B
Python
from abc import ABC
|
|
import asyncio
|
|
from threading import Thread
|
|
from typing import List
|
|
|
|
from aiohttp import web
|
|
|
|
|
|
class MockHTTPServerBase(ABC):
|
|
def __init__(self, port: int, routes: List):
|
|
self.port = port
|
|
self.app = web.Application()
|
|
self.app.add_routes(routes)
|
|
self.runner = web.AppRunner(self.app)
|
|
self.loop = asyncio.new_event_loop()
|
|
|
|
def _start_loop(self):
|
|
asyncio.set_event_loop(self.loop)
|
|
self.loop.run_until_complete(self.runner.setup())
|
|
site = web.TCPSite(self.runner, '127.0.0.1', self.port)
|
|
self.loop.run_until_complete(site.start())
|
|
self.loop.run_forever()
|
|
|
|
def start(self):
|
|
t = Thread(target=self._start_loop)
|
|
t.start()
|
|
|
|
def stop(self):
|
|
self.loop.call_soon_threadsafe(self.loop.stop)
|