From 9196c61a12e8d7a0084317f6bf6db0aeb1b5dcca Mon Sep 17 00:00:00 2001
From: Dmitry <b4tm4n@mail.ru>
Date: Tue, 15 Aug 2023 10:45:08 +0300
Subject: [PATCH] add test for writer.sink()

---
 lib/src/async_db.rs | 36 ++++++++++++++++++++++++++++++++----
 1 file changed, 32 insertions(+), 4 deletions(-)

diff --git a/lib/src/async_db.rs b/lib/src/async_db.rs
index 4990eab..991e825 100644
--- a/lib/src/async_db.rs
+++ b/lib/src/async_db.rs
@@ -439,6 +439,34 @@ mod test {
         }
     }
 
+    #[tokio::test]
+    async fn test_write_sink_read() {
+        let dir = tempdir().expect("tempdir");
+        let tmpfile = dir.path().join("test.tmp");
+        let opts = WriterOpts {
+            data_buf_size: 10 * 1024 * 1024,
+            out_buf_size: 10 * 1024 * 1024,
+            ..Default::default()
+        };
+        let mut writer: Writer<TestData> = Writer::new(&tmpfile, opts).await.expect("new writer");
+
+        let items_iter = gen_data(5);
+        let items: Vec<TestData> = items_iter.collect();
+
+        let src = futures::stream::iter(items.clone()).map(Ok);
+        pin_mut!(src);
+        src.forward(writer.sink()).await.expect("forward");
+        writer.finish().await.expect("finish write");
+
+        let reader: Reader<TestData> = Reader::new(&tmpfile, 2048).await.expect("new reader");
+        assert_eq!(items.len(), reader.len());
+
+        for (idx, item) in items.iter().enumerate() {
+            let ritem = reader.get(idx).await.expect("get");
+            assert_eq!(*item, ritem);
+        }
+    }
+
     #[tokio::test]
     async fn test_write_read_get_with_buf() {
         let dir = tempdir().expect("tempdir");
@@ -482,9 +510,9 @@ mod test {
         let items_iter = gen_data(5);
         let items: Vec<TestData> = items_iter.collect();
 
-        let src = futures::stream::iter(items.clone()).map(Ok);
+        let src = futures::stream::iter(items.clone());
         pin_mut!(src);
-        src.forward(writer.sink()).await.expect("forward");
+        writer.load(src).await.expect("load");
         writer.finish().await.expect("finish write");
 
         let reader: Reader<TestData> = Reader::new(&tmpfile, 2048).await.expect("new reader");
@@ -519,9 +547,9 @@ mod test {
         let items_iter = gen_data(5);
         let items: Vec<TestData> = items_iter.collect();
 
-        let src = futures::stream::iter(items.clone()).map(Ok);
+        let src = futures::stream::iter(items.clone());
         pin_mut!(src);
-        src.forward(writer.sink()).await.expect("forward");
+        writer.load(src).await.expect("load");
         writer.finish().await.expect("finish write");
 
         let reader: Reader<TestData> = Reader::new(&tmpfile, 2048).await.expect("new reader");