checkpoint event for tests

deltachat-jsonrpc/src/api.rs

@@ -152,6 +152,13 @@ impl CommandApi {
         tokio::time::sleep(std::time::Duration::from_secs_f64(delay)).await
     }
 
+    /// Emit the test checkpoint event, used for tests.
+    /// Returns a unique id of the emitted `TestCheckpointEvent`.
+    async fn trigger_checkpoint_event(&self, account_id: u32) -> Result<usize> {
+        let ctx = self.get_context(account_id).await?;
+        Ok(ctx.emit_test_checkpoint_event().await)
+    }
+
     // ---------------------------------------------
     //  Misc top level functions
     // ---------------------------------------------

deltachat-jsonrpc/src/api/types/events.rs

@@ -259,6 +259,13 @@ pub enum EventType {
     /// If `chat_id` is set to None, then all currently visible chats need to be rerendered, and all not-visible items need to be cleared from cache if the UI has a cache.
     #[serde(rename_all = "camelCase")]
     ChatlistItemChanged { chat_id: Option<u32> },
+
+    /// Tests can trigger this event and then wait for it, to make sure all events before it were consumed.
+    /// Useful for tests that test whether a specific event is emitted,
+    TestCheckpointEvent {
+        /// unique id to recognize the event
+        id: usize,
+    },
 }
 
 impl From<CoreEventType> for EventType {
@@ -370,6 +377,7 @@ impl From<CoreEventType> for EventType {
                 chat_id: chat_id.map(|id| id.to_u32()),
             },
             CoreEventType::ChatlistChanged => ChatlistChanged,
+            CoreEventType::TestCheckpointEvent { id } => TestCheckpointEvent { id },
         }
     }
 }

deltachat-rpc-client/src/deltachat_rpc_client/const.py

@@ -61,6 +61,7 @@ class EventType(str, Enum):
     WEBXDC_INSTANCE_DELETED = "WebxdcInstanceDeleted"
     CHATLIST_CHANGED = "ChatlistChanged"
     CHATLIST_ITEM_CHANGED = "ChatlistItemChanged"
+    TEST_CHECKPOINT_EVENT = "TestCheckpointEvent"
 
 
 class ChatId(IntEnum):

deltachat-rpc-client/src/deltachat_rpc_client/rpc.py

@@ -6,10 +6,12 @@
 import os
 import subprocess
 import sys
-from queue import Empty, Queue
+from queue import Queue
 from threading import Event, Thread
 from typing import Any, Iterator, Optional
 
+from .const import EventType
+
 
 class JsonRpcError(Exception):
     pass
@@ -190,12 +192,11 @@ def wait_for_event(self, account_id: int) -> Optional[dict]:
 
     def clear_all_events(self, account_id: int):
         """Removes all queued-up events for a given account. Useful for tests."""
-        queue = self.get_queue(account_id)
-        try:
-            while True:
-                queue.get_nowait()
-        except Empty:
-            pass
+        id = self.trigger_checkpoint_event(account_id)
+        while True:
+            event = self.wait_for_event(account_id)
+            if event.kind == EventType.TEST_CHECKPOINT_EVENT and event.id == id:
+                break
 
     def __getattr__(self, attr: str):
         return RpcMethod(self, attr)

src/context.rs

@@ -288,6 +288,9 @@ pub struct InnerContext {
 
     /// True if account has subscribed to push notifications via IMAP.
     pub(crate) push_subscribed: AtomicBool,
+
+    /// Counter for the id for the test checkpoint event that is used by tests.
+    pub(crate) test_event_checkpoint_counter: Mutex<usize>,
 }
 
 /// The state of ongoing process.
@@ -445,6 +448,7 @@ impl Context {
             debug_logging: std::sync::RwLock::new(None),
             push_subscriber,
             push_subscribed: AtomicBool::new(false),
+            test_event_checkpoint_counter: Mutex::new(0),
         };
 
         let ctx = Context {
@@ -1362,6 +1366,15 @@ impl Context {
         wal_fname.push("-wal");
         dbfile.with_file_name(wal_fname)
     }
+
+    /// Emit the test checkpoint event, used for tests.
+    /// Returns a unique id of the emitted `TestCheckpointEvent`.
+    pub async fn emit_test_checkpoint_event(&self) -> usize {
+        let mut counter = self.inner.test_event_checkpoint_counter.lock().await;
+        *counter = counter.wrapping_add(1);
+        self.emit_event(EventType::TestCheckpointEvent { id: *counter });
+        *counter
+    }
 }
 
 /// Returns core version as a string.

src/events/payload.rs

@@ -302,4 +302,11 @@ pub enum EventType {
         /// ID of the changed chat
         chat_id: Option<ChatId>,
     },
+
+    /// Tests can trigger this event and then wait for it, to make sure all events before it were consumed.
+    /// Useful for tests that test whether a specific event is emitted,
+    TestCheckpointEvent {
+        /// unique id to recognize the event
+        id: usize,
+    },
 }

src/test_utils.rs

@@ -785,6 +785,17 @@ impl TestContext {
 
         chat_id
     }
+
+    /// Clears event queue.
+    /// Works by emitting a `TestCheckpointEvent` and consuming all events until.
+    /// times out after 10 seconds.
+    pub async fn clear_events(&self) {
+        let event_id = self.emit_test_checkpoint_event().await;
+        self.evtracker.get_matching(|ev| match ev {
+            EventType::TestCheckpointEvent { id } => event_id == *id,
+            _ => false
+        }).await;
+    }
 }
 
 impl Deref for TestContext {