|
| 1 | +use std::sync::mpsc::channel; |
| 2 | +use std::time::Duration; |
| 3 | + |
| 4 | +use rusqlite::Connection; |
| 5 | +use sqlite_watcher::queue::ChangeOperation; |
| 6 | +use sqlite_watcher::wal::{start_wal_watcher, WalWatcherConfig}; |
| 7 | +use tempfile::tempdir; |
| 8 | + |
| 9 | +#[test] |
| 10 | +fn integration_watcher_emits_changes() { |
| 11 | + let dir = tempdir().unwrap(); |
| 12 | + let db_path = dir.path().join("tailer.db"); |
| 13 | + let writer = Connection::open(&db_path).unwrap(); |
| 14 | + writer |
| 15 | + .execute_batch( |
| 16 | + r#" |
| 17 | + PRAGMA journal_mode=WAL; |
| 18 | + CREATE TABLE posts(id INTEGER PRIMARY KEY, title TEXT); |
| 19 | + "#, |
| 20 | + ) |
| 21 | + .unwrap(); |
| 22 | + |
| 23 | + let (tx, rx) = channel(); |
| 24 | + let _handle = start_wal_watcher( |
| 25 | + &db_path, |
| 26 | + WalWatcherConfig { |
| 27 | + poll_interval: Duration::from_millis(100), |
| 28 | + min_event_bytes: 0, |
| 29 | + }, |
| 30 | + tx, |
| 31 | + ) |
| 32 | + .unwrap(); |
| 33 | + |
| 34 | + std::thread::sleep(Duration::from_millis(200)); |
| 35 | + |
| 36 | + writer |
| 37 | + .execute("INSERT INTO posts(title) VALUES ('hello')", []) |
| 38 | + .unwrap(); |
| 39 | + let insert = rx.recv_timeout(Duration::from_secs(3)).unwrap(); |
| 40 | + assert_eq!(insert.table_name, "posts"); |
| 41 | + assert_eq!(insert.operation, ChangeOperation::Insert); |
| 42 | + |
| 43 | + writer |
| 44 | + .execute("UPDATE posts SET title='hi' WHERE id=1", []) |
| 45 | + .unwrap(); |
| 46 | + let update = rx.recv_timeout(Duration::from_secs(3)).unwrap(); |
| 47 | + assert_eq!(update.operation, ChangeOperation::Update); |
| 48 | + |
| 49 | + writer.execute("DELETE FROM posts WHERE id=1", []).unwrap(); |
| 50 | + let delete = rx.recv_timeout(Duration::from_secs(3)).unwrap(); |
| 51 | + assert_eq!(delete.operation, ChangeOperation::Delete); |
| 52 | +} |
0 commit comments