Use connection manager for tests

simolus3 · simolus3 · commit 34659a55cf54 · 2025-08-12T15:08:49.000+02:00
diff --git a/packages/powersync_core/lib/src/sync/connection_manager.dart b/packages/powersync_core/lib/src/sync/connection_manager.dart
@@ -25,7 +25,8 @@ final class ConnectionManager {
   final Map<_RawStreamKey, _ActiveSubscription> _locallyActiveSubscriptions =
       {};
 
-  final StreamController<SyncStatus> _statusController = StreamController();
+  final StreamController<SyncStatus> _statusController =
+      StreamController.broadcast();
 
   /// Fires when an entry is added or removed from [_locallyActiveSubscriptions]
   /// while we're connected.
diff --git a/packages/powersync_core/lib/src/sync/streaming_sync.dart b/packages/powersync_core/lib/src/sync/streaming_sync.dart
@@ -622,6 +622,7 @@ final class _ActiveRustStreamingIteration {
           'include_defaults': sync.options.includeDefaultStreams,
           'active_streams': sync._activeSubscriptions
               .map((s) => {'name': s.name, 'params': s.parameters})
+              .toList(),
         }),
       );
       assert(_completedStream.isCompleted, 'Should have started streaming');
diff --git a/packages/powersync_core/test/in_memory_sync_test.dart b/packages/powersync_core/test/in_memory_sync_test.dart
@@ -5,7 +5,6 @@ import 'package:async/async.dart';
 import 'package:logging/logging.dart';
 import 'package:powersync_core/powersync_core.dart';
 import 'package:powersync_core/sqlite3_common.dart';
-import 'package:powersync_core/src/sync/streaming_sync.dart';
 import 'package:powersync_core/src/sync/protocol.dart';
 import 'package:test/test.dart';
 
@@ -53,35 +52,32 @@ void _declareTests(String name, SyncOptions options, bool bson) {
 
     late TestPowerSyncFactory factory;
     late CommonDatabase raw;
-    late PowerSyncDatabase database;
+    late TestDatabase database;
     late MockSyncService syncService;
     late Logger logger;
 
-    late StreamingSync syncClient;
     var credentialsCallbackCount = 0;
     Future<void> Function(PowerSyncDatabase) uploadData = (db) async {};
 
-    void createSyncClient({Schema? schema}) {
+    Future<void> connect() async {
       final (client, server) = inMemoryServer();
       server.mount(syncService.router.call);
 
-      final thisSyncClient = syncClient = database.connectWithMockService(
-        client,
-        TestConnector(() async {
-          credentialsCallbackCount++;
-          return PowerSyncCredentials(
-            endpoint: server.url.toString(),
-            token: 'token$credentialsCallbackCount',
-            expiresAt: DateTime.now(),
-          );
-        }, uploadData: (db) => uploadData(db)),
+      database.httpClient = client;
+      await database.connect(
+        connector: TestConnector(
+          () async {
+            credentialsCallbackCount++;
+            return PowerSyncCredentials(
+              endpoint: server.url.toString(),
+              token: 'token$credentialsCallbackCount',
+              expiresAt: DateTime.now(),
+            );
+          },
+          uploadData: (db) => uploadData(db),
+        ),
         options: options,
-        customSchema: schema,
       );
-
-      addTearDown(() async {
-        await thisSyncClient.abort();
-      });
     }
 
     setUp(() async {
@@ -92,7 +88,6 @@ void _declareTests(String name, SyncOptions options, bool bson) {
       factory = await testUtils.testFactory();
       (raw, database) = await factory.openInMemoryDatabase();
       await database.initialize();
-      createSyncClient();
     });
 
     tearDown(() async {
@@ -109,7 +104,7 @@ void _declareTests(String name, SyncOptions options, bool bson) {
           }
         });
       }
-      syncClient.streamingSync();
+      await connect();
       await syncService.waitForListener;
 
       expect(database.currentStatus.lastSyncedAt, isNull);
@@ -144,7 +139,7 @@ void _declareTests(String name, SyncOptions options, bool bson) {
       });
       await expectLater(
           status, emits(isSyncStatus(downloading: false, hasSynced: true)));
-      await syncClient.abort();
+      await database.disconnect();
 
       final independentDb = factory.wrapRaw(raw, logger: ignoredLogger);
       addTearDown(independentDb.close);
@@ -155,7 +150,7 @@ void _declareTests(String name, SyncOptions options, bool bson) {
       // A complete sync also means that all partial syncs have completed
       expect(
           independentDb.currentStatus
-              .statusForPriority(BucketPriority(3))
+              .statusForPriority(StreamPriority(3))
               .hasSynced,
           isTrue);
     });
@@ -249,7 +244,7 @@ void _declareTests(String name, SyncOptions options, bool bson) {
             database.watch('SELECT * FROM lists', throttle: Duration.zero));
         await expectLater(query, emits(isEmpty));
 
-        createSyncClient(schema: schema);
+        await database.updateSchema(schema);
         await waitForConnection();
 
         syncService
@@ -374,13 +369,13 @@ void _declareTests(String name, SyncOptions options, bool bson) {
             status,
             emitsThrough(
                 isSyncStatus(downloading: true, hasSynced: false).having(
-              (e) => e.statusForPriority(BucketPriority(0)).hasSynced,
+              (e) => e.statusForPriority(StreamPriority(0)).hasSynced,
               'status for $prio',
               isTrue,
             )),
           );
 
-          await database.waitForFirstSync(priority: BucketPriority(prio));
+          await database.waitForFirstSync(priority: StreamPriority(prio));
           expect(await database.getAll('SELECT * FROM customers'),
               hasLength(prio + 1));
         }
@@ -417,9 +412,9 @@ void _declareTests(String name, SyncOptions options, bool bson) {
             'priority': 1,
           }
         });
-        await database.waitForFirstSync(priority: BucketPriority(1));
+        await database.waitForFirstSync(priority: StreamPriority(1));
         expect(database.currentStatus.hasSynced, isFalse);
-        await syncClient.abort();
+        await database.disconnect();
 
         final independentDb = factory.wrapRaw(raw, logger: ignoredLogger);
         addTearDown(independentDb.close);
@@ -428,12 +423,12 @@ void _declareTests(String name, SyncOptions options, bool bson) {
         // Completing a sync for prio 1 implies a completed sync for prio 0
         expect(
             independentDb.currentStatus
-                .statusForPriority(BucketPriority(0))
+                .statusForPriority(StreamPriority(0))
                 .hasSynced,
             isTrue);
         expect(
             independentDb.currentStatus
-                .statusForPriority(BucketPriority(3))
+                .statusForPriority(StreamPriority(3))
                 .hasSynced,
             isFalse);
       });
@@ -681,10 +676,9 @@ void _declareTests(String name, SyncOptions options, bool bson) {
         await expectProgress(status, total: progress(5, 10));
 
         // Emulate the app closing - create a new independent sync client.
-        await syncClient.abort();
+        await database.disconnect();
         syncService.endCurrentListener();
 
-        createSyncClient();
         status = await waitForConnection();
 
         // Send same checkpoint again
@@ -715,10 +709,9 @@ void _declareTests(String name, SyncOptions options, bool bson) {
         await expectProgress(status, total: progress(5, 10));
 
         // Emulate the app closing - create a new independent sync client.
-        await syncClient.abort();
+        await database.disconnect();
         syncService.endCurrentListener();
 
-        createSyncClient();
         status = await waitForConnection();
 
         // Send checkpoint with additional data
@@ -749,9 +742,9 @@ void _declareTests(String name, SyncOptions options, bool bson) {
 
         // A sync rule deploy could reset buckets, making the new bucket smaller
         // than the existing one.
-        await syncClient.abort();
+        await database.disconnect();
         syncService.endCurrentListener();
-        createSyncClient();
+
         status = await waitForConnection();
         syncService.addLine({
           'checkpoint': Checkpoint(
@@ -770,8 +763,8 @@ void _declareTests(String name, SyncOptions options, bool bson) {
           await expectProgress(
             status,
             priorities: {
-              BucketPriority(0): prio0,
-              BucketPriority(2): prio2,
+              StreamPriority(0): prio0,
+              StreamPriority(2): prio2,
             },
             total: prio2,
           );
@@ -835,7 +828,7 @@ void _declareTests(String name, SyncOptions options, bool bson) {
       });
 
       await expectLater(status, emits(isSyncStatus(downloading: true)));
-      await syncClient.abort();
+      await database.disconnect();
 
       expect(syncService.controller.hasListener, isFalse);
     });
@@ -854,9 +847,6 @@ void _declareTests(String name, SyncOptions options, bool bson) {
       syncService.addLine({
         'checkpoint_complete': {'last_op_id': '10'}
       });
-
-      await pumpEventQueue();
-      expect(syncService.controller.hasListener, isFalse);
       syncService.endCurrentListener();
 
       // Should reconnect after delay.
@@ -876,9 +866,6 @@ void _declareTests(String name, SyncOptions options, bool bson) {
 
       await expectLater(status, emits(isSyncStatus(downloading: true)));
       syncService.addKeepAlive(0);
-
-      await pumpEventQueue();
-      expect(syncService.controller.hasListener, isFalse);
       syncService.endCurrentListener();
 
       // Should reconnect after delay.
diff --git a/packages/powersync_core/test/utils/abstract_test_utils.dart b/packages/powersync_core/test/utils/abstract_test_utils.dart
@@ -1,8 +1,11 @@
+import 'dart:async';
 import 'dart:convert';
 
 import 'package:http/http.dart';
 import 'package:logging/logging.dart';
 import 'package:powersync_core/powersync_core.dart';
+import 'package:powersync_core/src/abort_controller.dart';
+import 'package:powersync_core/src/database/powersync_db_mixin.dart';
 import 'package:powersync_core/src/sync/bucket_storage.dart';
 import 'package:powersync_core/src/sync/internal_connector.dart';
 import 'package:powersync_core/src/sync/options.dart';
@@ -63,20 +66,20 @@ Logger _makeTestLogger({Level level = Level.ALL, String? name}) {
 abstract mixin class TestPowerSyncFactory implements PowerSyncOpenFactory {
   Future<CommonDatabase> openRawInMemoryDatabase();
 
-  Future<(CommonDatabase, PowerSyncDatabase)> openInMemoryDatabase() async {
+  Future<(CommonDatabase, TestDatabase)> openInMemoryDatabase() async {
     final raw = await openRawInMemoryDatabase();
     return (raw, wrapRaw(raw));
   }
 
-  PowerSyncDatabase wrapRaw(
+  TestDatabase wrapRaw(
     CommonDatabase raw, {
     Logger? logger,
   }) {
-    return PowerSyncDatabase.withDatabase(
-      schema: schema,
+    return TestDatabase(
       database: SqliteDatabase.singleConnection(
           SqliteConnection.synchronousWrapper(raw)),
-      logger: logger,
+      logger: logger ?? Logger.detached('PowerSync.test'),
+      schema: schema,
     );
   }
 }
@@ -147,6 +150,78 @@ class TestConnector extends PowerSyncBackendConnector {
   }
 }
 
+final class TestDatabase
+    with SqliteQueries, PowerSyncDatabaseMixin
+    implements PowerSyncDatabase {
+  @override
+  final SqliteDatabase database;
+  @override
+  final Logger logger;
+  @override
+  Schema schema;
+
+  @override
+  late final Future<void> isInitialized;
+
+  Client? httpClient;
+
+  TestDatabase({
+    required this.database,
+    required this.logger,
+    required this.schema,
+  }) {
+    isInitialized = baseInit();
+  }
+
+  @override
+  Future<void> connectInternal({
+    required PowerSyncBackendConnector connector,
+    required ResolvedSyncOptions options,
+    required List<SubscribedStream> initiallyActiveStreams,
+    required Stream<List<SubscribedStream>> activeStreams,
+    required AbortController abort,
+    required Zone asyncWorkZone,
+  }) async {
+    final impl = StreamingSyncImplementation(
+      adapter: BucketStorage(this),
+      schemaJson: jsonEncode(schema),
+      client: httpClient!,
+      options: options,
+      connector: InternalConnector.wrap(connector, this),
+      logger: logger,
+      crudUpdateTriggerStream: database
+          .onChange(['ps_crud'], throttle: const Duration(milliseconds: 10)),
+      activeSubscriptions: initiallyActiveStreams,
+    );
+    impl.statusStream.listen(setStatus);
+
+    asyncWorkZone.run(impl.streamingSync);
+    final subscriptions = activeStreams.listen(impl.updateSubscriptions);
+
+    abort.onAbort.then((_) async {
+      subscriptions.cancel();
+      await impl.abort();
+      abort.completeAbort();
+    }).ignore();
+  }
+
+  @override
+  Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
+      {String? debugContext, Duration? lockTimeout}) async {
+    await isInitialized;
+    return database.readLock(callback,
+        debugContext: debugContext, lockTimeout: lockTimeout);
+  }
+
+  @override
+  Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
+      {String? debugContext, Duration? lockTimeout}) async {
+    await isInitialized;
+    return database.writeLock(callback,
+        debugContext: debugContext, lockTimeout: lockTimeout);
+  }
+}
+
 extension MockSync on PowerSyncDatabase {
   StreamingSyncImplementation connectWithMockService(
     Client client,
