Expand tests

Author: simolus3

crates/core/src/sync/storage_adapter.rs

@@ -5,7 +5,7 @@ use serde::Serialize;
 use sqlite_nostd::{self as sqlite, Connection, ManagedStmt, ResultCode};
 
 use crate::{
-    error::PowerSyncError,
+    error::{PSResult, PowerSyncError},
     ext::SafeManagedStmt,
     operations::delete_bucket,
     schema::Schema,
@@ -31,10 +31,11 @@ pub struct StorageAdapter {
 }
 
 impl StorageAdapter {
-    pub fn new(db: *mut sqlite::sqlite3) -> Result<Self, ResultCode> {
+    pub fn new(db: *mut sqlite::sqlite3) -> Result<Self, PowerSyncError> {
         // language=SQLite
-        let progress =
-            db.prepare_v2("SELECT name, count_at_last, count_since_last FROM ps_buckets")?;
+        let progress = db
+            .prepare_v2("SELECT name, count_at_last, count_since_last FROM ps_buckets")
+            .into_db_result(db)?;
 
         // language=SQLite
         let time = db.prepare_v2("SELECT unixepoch()")?;
@@ -46,11 +47,11 @@ impl StorageAdapter {
         })
     }
 
-    pub fn collect_bucket_requests(&self) -> Result<Vec<BucketRequest>, ResultCode> {
+    pub fn collect_bucket_requests(&self) -> Result<Vec<BucketRequest>, PowerSyncError> {
         // language=SQLite
         let statement = self.db.prepare_v2(
             "SELECT name, last_op FROM ps_buckets WHERE pending_delete = 0 AND name != '$local'",
-        )?;
+        ).into_db_result(self.db)?;
 
         let mut requests = Vec::<BucketRequest>::new();
 
@@ -97,23 +98,27 @@ impl StorageAdapter {
         }
     }
 
-    pub fn reset_progress(&self) -> Result<(), ResultCode> {
+    pub fn reset_progress(&self) -> Result<(), PowerSyncError> {
         self.db
-            .exec_safe("UPDATE ps_buckets SET count_since_last = 0, count_at_last = 0;")?;
+            .exec_safe("UPDATE ps_buckets SET count_since_last = 0, count_at_last = 0;")
+            .into_db_result(self.db)?;
         Ok(())
     }
 
-    pub fn lookup_bucket(&self, bucket: &str) -> Result<BucketInfo, ResultCode> {
+    pub fn lookup_bucket(&self, bucket: &str) -> Result<BucketInfo, PowerSyncError> {
         // We do an ON CONFLICT UPDATE simply so that the RETURNING bit works for existing rows.
         // We can consider splitting this into separate SELECT and INSERT statements.
         // language=SQLite
-        let bucket_statement = self.db.prepare_v2(
-            "INSERT INTO ps_buckets(name)
+        let bucket_statement = self
+            .db
+            .prepare_v2(
+                "INSERT INTO ps_buckets(name)
                             VALUES(?)
                         ON CONFLICT DO UPDATE
                             SET last_applied_op = last_applied_op
                         RETURNING id, last_applied_op",
-        )?;
+            )
+            .into_db_result(self.db)?;
         bucket_statement.bind_text(1, bucket, sqlite::Destructor::STATIC)?;
         let res = bucket_statement.step()?;
         debug_assert_matches!(res, ResultCode::ROW);
@@ -147,7 +152,8 @@ impl StorageAdapter {
 
         let update_bucket = self
             .db
-            .prepare_v2("UPDATE ps_buckets SET last_op = ? WHERE name = ?")?;
+            .prepare_v2("UPDATE ps_buckets SET last_op = ? WHERE name = ?")
+            .into_db_result(self.db)?;
 
         for bucket in checkpoint.buckets.values() {
             if bucket.is_in_priority(priority) {
@@ -214,7 +220,7 @@ impl StorageAdapter {
                 // partial completions.
                 let update = self.db.prepare_v2(
                     "UPDATE ps_buckets SET count_since_last = 0, count_at_last = ? WHERE name = ?",
-                )?;
+                ).into_db_result(self.db)?;
 
                 for bucket in checkpoint.buckets.values() {
                     if let Some(count) = bucket.count {

crates/core/src/sync/streaming_sync.rs

@@ -17,7 +17,7 @@ use futures_lite::FutureExt;
 
 use crate::{
     bson,
-    error::PowerSyncError,
+    error::{PowerSyncError, PowerSyncErrorCause},
     kv::client_id,
     state::DatabaseState,
     sync::{checkpoint::OwnedBucketChecksum, interface::StartSyncStream},
@@ -130,7 +130,7 @@ impl SyncIterationHandle {
         db: *mut sqlite::sqlite3,
         options: StartSyncStream,
         state: Arc<DatabaseState>,
-    ) -> Result<Self, ResultCode> {
+    ) -> Result<Self, PowerSyncError> {
         let runner = StreamingSyncIteration {
             db,
             options,
@@ -307,8 +307,9 @@ impl StreamingSyncIteration {
                 }
                 SyncLine::CheckpointDiff(diff) => {
                     let Some(target) = target.target_checkpoint_mut() else {
-                        return Err(PowerSyncError::argument_error(
+                        return Err(PowerSyncError::sync_protocol_error(
                             "Received checkpoint_diff without previous checkpoint",
+                            PowerSyncErrorCause::Unknown,
                         ));
                     };
 
@@ -325,8 +326,9 @@ impl StreamingSyncIteration {
                 }
                 SyncLine::CheckpointComplete(_) => {
                     let Some(target) = target.target_checkpoint_mut() else {
-                        return Err(PowerSyncError::argument_error(
+                        return Err(PowerSyncError::sync_protocol_error(
                             "Received checkpoint complete without previous checkpoint",
+                            PowerSyncErrorCause::Unknown,
                         ));
                     };
                     let result =

dart/test/error_test.dart

@@ -28,14 +28,47 @@ void main() {
         ]),
         throwsA(isSqliteException(
           1,
-          'powersync_replace_schema: internal SQLite call returned ERROR',
+          'powersync_replace_schema: internal SQLite call returned ERROR: malformed JSON',
         )),
       );
     });
+
+    test('missing client id', () {
+      db
+        ..execute('SELECT powersync_init()')
+        ..execute('DELETE FROM ps_kv;');
+
+      expect(
+        () => db.execute('SELECT powersync_client_id()'),
+        throwsA(isSqliteException(
+          4,
+          'powersync_client_id: No client_id found in ps_kv',
+        )),
+      );
+    });
+
+    group('sync protocol', () {
+      setUp(() => db.execute('SELECT powersync_init()'));
+
+      test('invalid json', () {
+        const stmt = 'SELECT powersync_control(?,?)';
+        db.execute('BEGIN');
+        final control = db.prepare(stmt);
+
+        control.execute(['start', null]);
+        expect(
+          () => control.execute(['line_text', 'invalid sync line']),
+          throwsA(isSqliteException(
+            4,
+            'powersync_control: Sync protocol error: invalid text line. cause: expected value at line 1 column 1',
+          )),
+        );
+      });
+    });
   });
 }
 
-Matcher isSqliteException(int code, String message) {
+Matcher isSqliteException(int code, dynamic message) {
   return isA<SqliteException>()
       .having((e) => e.extendedResultCode, 'extendedResultCode', code)
       .having((e) => e.message, 'message', message);