pyceo/tests/MockMailmanServer.py

76 lines
2.4 KiB
Python
Raw Normal View History

2021-08-13 20:11:56 -04:00
import asyncio
from threading import Thread
from aiohttp import web
class MockMailmanServer:
def __init__(self, port=8001, prefix='/3.1'):
self.port = port
2021-08-13 20:11:56 -04:00
self.app = web.Application()
self.app.add_routes([
web.post(prefix + '/members', self.subscribe),
web.delete(prefix + '/lists/{mailing_list}/member/{address}', self.unsubscribe),
2021-08-13 20:11:56 -04:00
])
self.runner = web.AppRunner(self.app)
self.loop = asyncio.new_event_loop()
2021-08-17 21:59:24 -04:00
# add more as necessary
self.subscriptions = {
'csc-general': [],
'exec': [],
'syscom': [],
'syscom-alerts': [],
}
2021-08-13 20:11:56 -04:00
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)
2021-08-13 20:11:56 -04:00
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)
def clear(self):
for key in self.subscriptions:
self.subscriptions[key].clear()
2021-08-13 20:11:56 -04:00
async def subscribe(self, request):
body = await request.post()
subscriber = body['subscriber']
2021-08-17 21:59:24 -04:00
list_id = body['list_id']
list_id = list_id[:list_id.index('.')]
if list_id not in self.subscriptions:
return web.json_response({
'description': 'No such list',
}, status=400)
subscribers = self.subscriptions[list_id]
if subscriber in subscribers:
2021-08-13 20:11:56 -04:00
return web.json_response({
'description': 'user is already subscribed',
}, status=409)
2021-08-17 21:59:24 -04:00
subscribers.append(subscriber)
2021-08-13 20:11:56 -04:00
return web.json_response({'status': 'OK'})
async def unsubscribe(self, request):
subscriber = request.match_info['address']
2021-08-17 21:59:24 -04:00
list_id = request.match_info['mailing_list']
list_id = list_id[:list_id.index('@')]
subscribers = self.subscriptions.get(list_id, [])
if subscriber not in subscribers:
2021-08-13 20:11:56 -04:00
return web.json_response({
'description': 'user is not subscribed',
}, status=404)
2021-08-17 21:59:24 -04:00
subscribers.remove(subscriber)
2021-08-13 20:11:56 -04:00
return web.json_response({'status': 'OK'})
if __name__ == '__main__':
server = MockMailmanServer()
server.start()