snap

python/tests/stress_test_db.py

@@ -1,18 +1,23 @@
+
 import time
+import threading
+import pytest
 import os
-from queue import Queue
+from queue import Queue, Empty
 
 import deltachat
 
 
 def test_db_busy_error(acfactory, tmpdir):
     starttime = time.time()
+    log_lock = threading.RLock()
 
     def log(string):
-        print("%3.2f %s" % (time.time() - starttime, string))
+        with log_lock:
+            print("%3.2f %s" % (time.time() - starttime, string))
 
     # make a number of accounts
-    accounts = acfactory.get_many_online_accounts(5, quiet=False)
+    accounts = acfactory.get_many_online_accounts(5, quiet=True)
     log("created %s accounts" % len(accounts))
 
     # put a bigfile into each account
@@ -36,7 +41,7 @@ def test_db_busy_error(acfactory, tmpdir):
     # each replier receives all events and sends report events to receive_queue
     repliers = []
     for acc in accounts:
-        replier = AutoReplier(acc, num_send=1000, num_bigfiles=0, report_func=report_func)
+        replier = AutoReplier(acc, log=log, num_send=1000, num_bigfiles=0, report_func=report_func)
         acc.add_account_plugin(replier)
         repliers.append(replier)
 
@@ -46,37 +51,59 @@ def test_db_busy_error(acfactory, tmpdir):
 
     alive_count = len(accounts)
     while alive_count > 0:
-        replier, report_type, report_args = report_queue.get(10)
-        addr = replier.account.get_self_contact().addr
-        assert addr
+        try:
+            replier, report_type, report_args = report_queue.get(timeout=10)
+        except Empty:
+            log("timeout waiting for next event")
+            pytest.fail("timeout exceeded")
         if report_type == ReportType.exit:
-            alive_count -= 1
-            log("{} EXIT -- remaining: {}".format(addr, alive_count))
-            replier.account.shutdown(wait=True)
-        elif report_type == ReportType.message_sent:
-            log("{} sent message: {}".format(addr, report_args[0].text))
-        elif report_type == ReportType.message_incoming:
-            log("{} incoming message: {}".format(addr, report_args[0].text))
+            replier.log("EXIT".format(alive_count))
         elif report_type == ReportType.ffi_error:
-            log("{} ERROR: {}".format(addr, report_args[0]))
-            replier.account.shutdown(wait=True)
-            alive_count -= 1
+            replier.log("ERROR: {}".format(addr, report_args[0]))
+        elif report_type == ReportType.message_echo:
+            continue
+        else:
+            raise ValueError("{} unknown report type {}, args={}".format(
+                addr, report_type, report_args
+            ))
+        alive_count -= 1
+        replier.log("shutting down")
+        replier.account.shutdown(wait=False)
+        replier.log("shut down complete, remaining={}".format(alive_count))
 
 
 class ReportType:
     exit = "exit"
-    message_sent = "message-sent"
     ffi_error = "ffi-error"
-    message_incoming = "message-incoming"
+    message_echo = "message-echo"
 
 
 class AutoReplier:
-    def __init__(self, account, report_func, num_send, num_bigfiles):
+    def __init__(self, account, log, num_send, num_bigfiles, report_func):
         self.account = account
+        self._log = log
         self.report_func = report_func
         self.num_send = num_send
         self.num_bigfiles = num_bigfiles
         self.current_sent = 0
+        self.addr = self.account.get_self_contact().addr
+
+        self._thread = threading.Thread(
+            name="Stats{}".format(self.account),
+            target=self.thread_stats
+        )
+        self._thread.setDaemon(True)
+        self._thread.start()
+
+    def log(self, message):
+        self._log("{} {}".format(self.addr, message))
+
+    def thread_stats(self):
+        # XXX later use, for now we just quit
+        return
+        while 1:
+            time.sleep(1.0)
+            break
 
     @deltachat.account_hookimpl
     def ac_incoming_message(self, message):
@@ -84,7 +111,7 @@ class AutoReplier:
             return
         message.accept_sender_contact()
         message.mark_seen()
-        self.report_func(self, ReportType.message_incoming, message)
+        self.log("incoming message: {}".format(message))
 
         self.current_sent += 1
         # we are still alive, let's send a reply
@@ -94,12 +121,14 @@ class AutoReplier:
         else:
             msg = message.chat.send_text("got message id {}, small text reply".format(message.id))
             assert msg.text
-        self.report_func(self, ReportType.message_sent, msg)
+        self.log("message-sent: {}".format(msg))
+        self.report_func(self, ReportType.message_echo)
         if self.current_sent >= self.num_send:
             self.report_func(self, ReportType.exit)
             return
 
     @deltachat.account_hookimpl
     def ac_process_ffi_event(self, ffi_event):
+        self.log(ffi_event)
         if ffi_event.name == "DC_EVENT_ERROR":
             self.report_func(self, ReportType.ffi_error, ffi_event)