Add tests

Author: simolus3

packages/sqlite_async/lib/src/utils/shared_utils.dart

@@ -133,6 +133,7 @@ extension ThrottledUpdates on CommonDatabase {
         final wrapped = _UpdateListener(listener);
         addListener(wrapped);
 
+        listener.onResume = wrapped.addPending;
         listener.onCancel = () => removeListener(wrapped);
       },
       isBroadcast: true,
@@ -149,6 +150,12 @@ class _UpdateListener {
   void notify(Set<String> pendingUpdates) {
     buffered.addAll(pendingUpdates);
     if (!downstream.isPaused) {
+      addPending();
+    }
+  }
+
+  void addPending() {
+    if (buffered.isNotEmpty) {
       downstream.add(buffered);
       buffered = {};
     }

packages/sqlite_async/test/native/watch_test.dart

@@ -7,6 +7,7 @@ import 'dart:math';
 
 import 'package:sqlite3/common.dart';
 import 'package:sqlite_async/sqlite_async.dart';
+import 'package:sqlite_async/src/utils/shared_utils.dart';
 import 'package:test/test.dart';
 
 import '../utils/test_utils_impl.dart';
@@ -31,6 +32,51 @@ void main() {
       return db;
     });
 
+    test('raw update notifications', () async {
+      final factory = await testUtils.testFactory(path: path);
+      final db = factory
+          .openDB(SqliteOpenOptions(primaryConnection: true, readOnly: false));
+
+      db.execute('CREATE TABLE a (bar INTEGER);');
+      db.execute('CREATE TABLE b (bar INTEGER);');
+      final events = <Set<String>>[];
+      final subscription = db.updatedTables.listen(events.add);
+
+      db.execute('insert into a default values');
+      expect(events, isEmpty); // should be async
+      await pumpEventQueue();
+      expect(events.removeLast(), {'a'});
+
+      db.execute('begin');
+      db.execute('insert into a default values');
+      db.execute('insert into b default values');
+      await pumpEventQueue();
+      expect(events, isEmpty); // should only trigger on commit
+      db.execute('commit');
+
+      await pumpEventQueue();
+      expect(events.removeLast(), {'a', 'b'});
+
+      db.execute('begin');
+      db.execute('insert into a default values');
+      db.execute('rollback');
+      expect(events, isEmpty);
+      await pumpEventQueue();
+      expect(events, isEmpty); // should ignore cancelled transactions
+
+      // Should still listen during pause, and dispatch on resume
+      subscription.pause();
+      db.execute('insert into a default values');
+      await pumpEventQueue();
+      expect(events, isEmpty);
+
+      subscription.resume();
+      await pumpEventQueue();
+      expect(events.removeLast(), {'a'});
+
+      subscription.pause();
+    });
+
     test('watch in isolate', () async {
       final db = await testUtils.setupDatabase(path: path);
       await createTables(db);