Make mail and matrix checks synchronous

amonit/check/mail.py

@@ -36,7 +36,7 @@ def imap_login(context, host, username, password, port=143, encrypt=None):
     client.logout()
 
 
-def send_then_receive(context, account_from, account_to):
+def send_then_receive(context, account_from, account_to, timeout=120):
     """ Send a mail using SMTP, then check that it is received using IMAP
     This is useful either on a single server or across servers to ensure
     SMTP works properly from server to server.
@@ -70,7 +70,8 @@ def send_then_receive(context, account_from, account_to):
     client.login(account_to["username"], account_to["password"])
     client.select(account_to["inbox"])
     filter_string = f'(UNSEEN SUBJECT "{canary}")'
-    for _ in range(10):
+    delay = 0
+    while delay < timeout:
         status, data = client.uid("search", None, filter_string)
         delay = int(time.time() - start_time)
         if status == "OK" and data[0]:
@@ -81,7 +82,7 @@ def send_then_receive(context, account_from, account_to):
             result["message"] = f"message delivered in {delay} seconds"
             break
         time.sleep(10)
-    else:
+    if "up" not in result:
         result["up"] = False
         domain = account_to["username"].split("@")[1]
         result["message"] = f"message not delivered to {domain} ({canary})"

amonit/check/matrix.py

 import os
 import json
+import time
 
 from matrix_client import client as matrix_client
-from matrix_client import room as matrix_room
 
 
-def send_then_receive(context, account_from, account_to, roomid):
+def send_then_receive(context, account_from, account_to, roomid, timeout=120):
     """ Send a message using Matrix, then check that it is received.
     This is useful either on a single server or across homeservers to
     check that federation work properly.
 
     The canary is first sent, then looked for in the next execution.
     """
-    result = {"up": False, "message": ""}
-    if "canary" in context:
-        canary = context["canary"]
-        try:
-            client_from = matrix_client.MatrixHttpApi(
-                account_from["hs"], token=account_from["token"]
-            )
-            filter = json.dumps(
-                {"room":{"rooms":[roomid]}})
-            if "next_batch" in context:
-                since = context["next_batch"]
-                sync = client_from.sync(since=since, filter=filter)
-            else:
-                sync = client_from.sync(filter=filter)
-            result["next_batch"] = sync["next_batch"]
-            events = sync["rooms"]["join"][roomid]["timeline"]["events"]
-            for event in events:
-                if event["content"].get("body", "") == canary:
-                    result["up"] = True
-        except Exception as error:
-            result = {"up": False, "message": str(error)}
-    # Then drop a new canary
+    # Generate and store a canary
     canary = os.urandom(8).hex()
+    result = {"canary": canary}
+    # First drop the canary
     client_to = matrix_client.MatrixHttpApi(
         account_to["hs"], token=account_to["token"]
     )
     client_to.send_message(roomid, canary)
-    result["canary"] = canary
+    start_time = time.time()
+    # Then wait for the delivery
+    client_from = matrix_client.MatrixHttpApi(
+        account_from["hs"], token=account_from["token"]
+    )
+    filter_string = json.dumps({"room":{"rooms":[roomid]}})
+    delay = 0
+    next_batch = context.get("next_batch", None)
+    while delay < timeout and "up" not in result:
+        sync = client_from.sync(since=next_batch, filter=filter_string)
+        delay = int(time.time() - start_time)
+        next_batch = sync["next_batch"]
+        for event in sync["rooms"]["join"][roomid]["timeline"]["events"]:
+            if event["content"].get("body", "") == canary:
+                result["up"] = True
+                result["message"] = f"message delivered in {delay} seconds"
+    if "up" not in result:
+        result["up"] = False
+        result["message"] = f"message not delivered to {account_to['hs']}"
+    # Store the next batch for next execution
+    result["next_batch"] = next_batch
     return result