switchy_database 0.5.0

Switchy database package
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

#![cfg(all(feature = "sqlite-rusqlite", feature = "schema"))]

use std::sync::Arc;
use switchy_database::{
    Database, DatabaseError,
    rusqlite::RusqliteDatabase,
    schema::{Column, DataType},
};

fn open(path: &std::path::Path) -> RusqliteDatabase {
    RusqliteDatabase::new(vec![Arc::new(switchy_async::sync::Mutex::new(
        rusqlite::Connection::open(path).unwrap(),
    ))])
}

#[tokio::test]
async fn observations_are_connection_affine_non_mutating_and_fail_closed() {
    let path = std::path::PathBuf::from(format!(
        "file:change-observation-{}?mode=memory&cache=shared",
        std::process::id()
    ));
    let db = open(&path);
    db.create_table("items")
        .column(Column {
            name: "id".into(),
            data_type: DataType::BigInt,
            nullable: false,
            auto_increment: false,
            default: None,
        })
        .execute(&db)
        .await
        .unwrap();
    let before = db.sqlite_change_observation().await.unwrap();
    assert_eq!(before, db.sqlite_change_observation().await.unwrap());
    db.insert("items")
        .value("id", 1)
        .execute(&db)
        .await
        .unwrap();
    let local = db.sqlite_change_observation().await.unwrap();
    assert_eq!(local.data_version, before.data_version);
    assert_eq!(local.total_changes, before.total_changes + 1);

    let external = open(&path);
    external
        .insert("items")
        .value("id", 2)
        .execute(&external)
        .await
        .unwrap();
    let remote = db.sqlite_change_observation().await.unwrap();
    assert_ne!(remote.data_version, local.data_version);
    assert_eq!(remote.total_changes, local.total_changes);

    let tx = db.begin_transaction().await.unwrap();
    assert!(matches!(
        db.sqlite_change_observation().await,
        Err(DatabaseError::Busy)
    ));
    tx.rollback().await.unwrap();
    assert_eq!(remote, db.sqlite_change_observation().await.unwrap());
    for commit in [false, true] {
        let before = db.sqlite_change_observation().await.unwrap();
        let tx = db.begin_transaction().await.unwrap();
        assert_eq!(before, tx.sqlite_change_observation().await.unwrap());
        tx.select("items").execute(tx.as_ref()).await.unwrap();
        assert_eq!(before, tx.sqlite_change_observation().await.unwrap());
        tx.insert("items")
            .value("id", 3)
            .execute(tx.as_ref())
            .await
            .unwrap();
        let written = tx.sqlite_change_observation().await.unwrap();
        assert_eq!(written.total_changes, before.total_changes + 1);
        assert_eq!(written.data_version, before.data_version);
        if commit {
            tx.commit().await.unwrap();
        } else {
            tx.rollback().await.unwrap();
        }
        assert_eq!(written, db.sqlite_change_observation().await.unwrap());
    }
    db.close().await.unwrap();
    assert!(matches!(
        db.sqlite_change_observation().await,
        Err(DatabaseError::ConnectionClosed)
    ));

    let pool = RusqliteDatabase::new(
        (0..2)
            .map(|_| {
                Arc::new(switchy_async::sync::Mutex::new(
                    rusqlite::Connection::open(&path).unwrap(),
                ))
            })
            .collect(),
    );
    let tx = pool.begin_transaction().await.unwrap();
    // Even with only one available connection, this is still a multi-connection handle.
    assert!(matches!(
        pool.sqlite_change_observation().await,
        Err(DatabaseError::UnsupportedOperation(_))
    ));
    tx.rollback().await.unwrap();
}