matrix-webhook/tests/start.py

#!/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'
MATRIX_URL, MATRIX_ID, MATRIX_PW = (environ[v] for v in ['MATRIX_URL', 'MATRIX_ID', 'MATRIX_PW'])


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


def wait_available(url: str, key: str, timeout: int = 10) -> bool:
    """Wait until a service answer correctly or timeout."""
    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."""
    if not wait_available(f'{MATRIX_URL}/_matrix/client/r0/login', 'flows'):
        return False

    # Try to register an 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)

    bot = Popen(['coverage', 'run', 'matrix_webhook.py'])

    if not wait_available(BOT_URL, 'status'):
        return False

    ret = main(module=None, exit=False).result.wasSuccessful()
    bot.terminate()
    for cmd in ['report', 'html', 'xml']:
        run(['coverage', cmd])
    return ret


if __name__ == '__main__':
    exit(not run_and_test())