77 lines
2.4 KiB
Python
Executable file
77 lines
2.4 KiB
Python
Executable file
#!/usr/bin/env python
|
|
"""Entry point to start an instrumentalized bot for coverage and run tests."""
|
|
|
|
from os import environ
|
|
from subprocess import Popen, run
|
|
from time import time
|
|
from unittest import main
|
|
|
|
import httpx
|
|
import yaml
|
|
from synapse._scripts.register_new_matrix_user import request_registration
|
|
|
|
BOT_URL = 'http://localhost:4785'
|
|
KEY, MATRIX_URL, MATRIX_ID, MATRIX_PW = (environ[v] for v in ['API_KEY', 'MATRIX_URL', 'MATRIX_ID', 'MATRIX_PW'])
|
|
FULL_ID = f'@{MATRIX_ID}:{MATRIX_URL.split("/")[2]}'
|
|
|
|
|
|
def bot_req(req=None, key=None, room_id=None):
|
|
"""Bot requests boilerplate."""
|
|
if key is not None:
|
|
req['key'] = key
|
|
url = BOT_URL if room_id is None else f'{BOT_URL}/{room_id}'
|
|
return httpx.post(url, json=req).json()
|
|
|
|
|
|
def wait_available(url: str, key: str, timeout: int = 10) -> bool:
|
|
"""Wait until a service answer correctly or timeout."""
|
|
def check_json(url: str, key: str) -> bool:
|
|
"""Ensure a service at a given url answers with valid json containing a certain key."""
|
|
try:
|
|
data = httpx.get(url).json()
|
|
return key in data
|
|
except httpx.ConnectError:
|
|
return False
|
|
|
|
start = time()
|
|
while True:
|
|
if check_json(url, key):
|
|
return True
|
|
if time() > start + timeout:
|
|
return False
|
|
|
|
|
|
def run_and_test():
|
|
"""Launch the bot and its tests."""
|
|
# Start the server, and wait for it
|
|
srv = Popen(['python', '-m', 'synapse.app.homeserver', '--config-path', '/srv/homeserver.yaml'])
|
|
if not wait_available(f'{MATRIX_URL}/_matrix/client/r0/login', 'flows'):
|
|
return False
|
|
|
|
# Register a user for the bot.
|
|
with open('/srv/homeserver.yaml') as f:
|
|
secret = yaml.safe_load(f.read()).get("registration_shared_secret", None)
|
|
request_registration(MATRIX_ID, MATRIX_PW, MATRIX_URL, secret, admin=True)
|
|
|
|
# Start the bot, and wait for it
|
|
bot = Popen(['coverage', 'run', 'matrix_webhook.py'])
|
|
if not wait_available(BOT_URL, 'status'):
|
|
return False
|
|
|
|
# Run the main unittest module
|
|
ret = main(module=None, exit=False).result.wasSuccessful()
|
|
|
|
srv.terminate()
|
|
|
|
# TODO Check what the bot says when the server is offline
|
|
# print(bot_req({'text': 'bye'}, KEY), {'status': 200, 'ret': 'OK'})
|
|
|
|
bot.terminate()
|
|
|
|
for cmd in ['report', 'html', 'xml']:
|
|
run(['coverage', cmd])
|
|
return ret
|
|
|
|
|
|
if __name__ == '__main__':
|
|
exit(not run_and_test())
|