diff --git a/matrix_webhook.py b/matrix_webhook.py
index e9f2efa..5540f59 100755
--- a/matrix_webhook.py
+++ b/matrix_webhook.py
@@ -18,11 +18,11 @@ from markdown import markdown
from nio import AsyncClient
from nio.exceptions import LocalProtocolError
-SERVER_ADDRESS = (os.environ.get('HOST', ''), int(os.environ.get('PORT', 4785)))
-MATRIX_URL = os.environ.get('MATRIX_URL', 'https://matrix.org')
-MATRIX_ID = os.environ.get('MATRIX_ID', '@wwm:matrix.org')
-MATRIX_PW = os.environ['MATRIX_PW']
-API_KEY = os.environ['API_KEY']
+SERVER_ADDRESS = (os.environ.get("HOST", ""), int(os.environ.get("PORT", 4785)))
+MATRIX_URL = os.environ.get("MATRIX_URL", "https://matrix.org")
+MATRIX_ID = os.environ.get("MATRIX_ID", "@wwm:matrix.org")
+MATRIX_PW = os.environ["MATRIX_PW"]
+API_KEY = os.environ["API_KEY"]
CLIENT = AsyncClient(MATRIX_URL, MATRIX_ID)
@@ -37,20 +37,22 @@ async def handler(request):
try:
data = json.loads(data.decode())
except json.decoder.JSONDecodeError:
- return create_json_response(HTTPStatus.BAD_REQUEST, 'Invalid JSON')
+ return create_json_response(HTTPStatus.BAD_REQUEST, "Invalid JSON")
- if not all(key in data for key in ['text', 'key']):
- return create_json_response(HTTPStatus.BAD_REQUEST, 'Missing text and/or API key property')
+ if not all(key in data for key in ["text", "key"]):
+ return create_json_response(
+ HTTPStatus.BAD_REQUEST, "Missing text and/or API key property"
+ )
- if data['key'] != API_KEY:
- return create_json_response(HTTPStatus.UNAUTHORIZED, 'Invalid API key')
+ if data["key"] != API_KEY:
+ return create_json_response(HTTPStatus.UNAUTHORIZED, "Invalid API key")
room_id = request.path[1:]
content = {
- 'msgtype': 'm.text',
- 'body': data['text'],
- 'format': 'org.matrix.custom.html',
- 'formatted_body': markdown(str(data['text']), extensions=['extra']),
+ "msgtype": "m.text",
+ "body": data["text"],
+ "format": "org.matrix.custom.html",
+ "formatted_body": markdown(str(data["text"]), extensions=["extra"]),
}
try:
await send_room_message(room_id, content)
@@ -58,18 +60,20 @@ async def handler(request):
await CLIENT.login(MATRIX_PW)
await send_room_message(room_id, content)
- return create_json_response(HTTPStatus.OK, 'OK')
+ return create_json_response(HTTPStatus.OK, "OK")
def create_json_response(status, ret):
"""Create a JSON response."""
- response_data = {'status': status, 'ret': ret}
+ response_data = {"status": status, "ret": ret}
return web.json_response(response_data, status=status)
async def send_room_message(room_id, content):
"""Send a message to a room."""
- return await CLIENT.room_send(room_id=room_id, message_type='m.room.message', content=content)
+ return await CLIENT.room_send(
+ room_id=room_id, message_type="m.room.message", content=content
+ )
async def main(event):
@@ -113,5 +117,5 @@ def run():
loop.close()
-if __name__ == '__main__':
+if __name__ == "__main__":
run()
diff --git a/tests/start.py b/tests/start.py
index 97cc0f0..27c184c 100755
--- a/tests/start.py
+++ b/tests/start.py
@@ -10,21 +10,24 @@ 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'])
+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}'
+ 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:
@@ -44,18 +47,26 @@ def wait_available(url: str, key: str, timeout: int = 10) -> bool:
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'):
+ 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:
+ 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'):
+ bot = Popen(["coverage", "run", "matrix_webhook.py"])
+ if not wait_available(BOT_URL, "status"):
return False
# Run the main unittest module
@@ -68,10 +79,10 @@ def run_and_test():
bot.terminate()
- for cmd in ['report', 'html', 'xml']:
- run(['coverage', cmd])
+ for cmd in ["report", "html", "xml"]:
+ run(["coverage", cmd])
return ret
-if __name__ == '__main__':
+if __name__ == "__main__":
exit(not run_and_test())
diff --git a/tests/tests.py b/tests/tests.py
index 3bf710e..f6ec6f9 100644
--- a/tests/tests.py
+++ b/tests/tests.py
@@ -9,25 +9,33 @@ from .start import FULL_ID, KEY, MATRIX_ID, MATRIX_PW, MATRIX_URL, bot_req
class BotTest(unittest.IsolatedAsyncioTestCase):
"""Main test class."""
+
def test_errors(self):
"""Check the bot's error paths."""
- self.assertEqual(bot_req(), {'status': 400, 'ret': 'Invalid JSON'})
- self.assertEqual(bot_req({'toto': 3}), {'status': 400, 'ret': 'Missing text and/or API key property'})
- self.assertEqual(bot_req({'text': 3, 'key': None}), {'status': 401, 'ret': 'Invalid API key'})
+ self.assertEqual(bot_req(), {"status": 400, "ret": "Invalid JSON"})
+ self.assertEqual(
+ bot_req({"toto": 3}),
+ {"status": 400, "ret": "Missing text and/or API key property"},
+ )
+ self.assertEqual(
+ bot_req({"text": 3, "key": None}), {"status": 401, "ret": "Invalid API key"}
+ )
# TODO: we are not sending to a real room, so this should not be "OK"
- self.assertEqual(bot_req({'text': 3}, KEY), {'status': 200, 'ret': 'OK'})
+ self.assertEqual(bot_req({"text": 3}, KEY), {"status": 200, "ret": "OK"})
async def test_message(self):
"""Send a markdown message, and check the result."""
- text = '# Hello'
+ text = "# Hello"
messages = []
client = nio.AsyncClient(MATRIX_URL, MATRIX_ID)
await client.login(MATRIX_PW)
room = await client.room_create()
- self.assertEqual(bot_req({'text': text}, KEY, room.room_id), {'status': 200, 'ret': 'OK'})
+ self.assertEqual(
+ bot_req({"text": text}, KEY, room.room_id), {"status": 200, "ret": "OK"}
+ )
sync = await client.sync()
messages = await client.room_messages(room.room_id, sync.next_batch)
@@ -36,4 +44,4 @@ class BotTest(unittest.IsolatedAsyncioTestCase):
message = messages.chunk[0]
self.assertEqual(message.sender, FULL_ID)
self.assertEqual(message.body, text)
- self.assertEqual(message.formatted_body, 'Hello
')
+ self.assertEqual(message.formatted_body, "Hello
")