Skip to content

Commit 26a02df

Browse files
BtXinBtXin
committed
Fixed integration tests
1 parent 774570a commit 26a02df

4 files changed

Lines changed: 144 additions & 84 deletions

File tree

‎crates/sources/src/providers/mongo.rs‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1558,7 +1558,11 @@ mod tests {
15581558
"SELECT product_id, name, category, price, in_stock FROM products ORDER BY product_id",
15591559
)
15601560
.await;
1561-
assert_eq!(total_rows(&batches), 5);
1561+
assert!(
1562+
total_rows(&batches) >= 5,
1563+
"expected at least 5 seeded rows, got {}",
1564+
total_rows(&batches)
1565+
);
15621566
}
15631567

15641568
#[tokio::test]
@@ -1568,7 +1572,11 @@ mod tests {
15681572
register_ci_collection(&mut ctx, "products", "product_id").await;
15691573

15701574
let batches = query_all(&ctx, "SELECT name FROM products ORDER BY product_id").await;
1571-
assert_eq!(total_rows(&batches), 5);
1575+
assert!(
1576+
total_rows(&batches) >= 5,
1577+
"expected at least 5 seeded rows, got {}",
1578+
total_rows(&batches)
1579+
);
15721580
assert_eq!(batches[0].num_columns(), 1);
15731581
}
15741582

‎crates/sources/src/providers/mysql.rs‎

Lines changed: 34 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -759,7 +759,7 @@ mod tests {
759759
register_ci_table(&mut ctx, "users").await;
760760

761761
let batches = query_all(&ctx, "SELECT id, name, email FROM users ORDER BY id").await;
762-
assert_eq!(total_rows(&batches), 3);
762+
assert!(total_rows(&batches) >= 3);
763763
}
764764

765765
#[tokio::test]
@@ -769,7 +769,7 @@ mod tests {
769769
register_ci_table(&mut ctx, "users").await;
770770

771771
let batches = query_all(&ctx, "SELECT name FROM users ORDER BY id").await;
772-
assert_eq!(total_rows(&batches), 3);
772+
assert!(total_rows(&batches) >= 3);
773773
assert_eq!(batches[0].num_columns(), 1);
774774

775775
let names = batches[0]
@@ -842,8 +842,8 @@ mod tests {
842842
.await
843843
.unwrap();
844844

845-
let before = query_all(&ctx, "SELECT id FROM users").await;
846-
let before_count = total_rows(&before);
845+
let before = query_all(&ctx, "SELECT id FROM users WHERE name = 'DeleteMe'").await;
846+
assert_eq!(total_rows(&before), 1);
847847

848848
ctx.sql("DELETE FROM users WHERE name = 'DeleteMe'")
849849
.await
@@ -852,8 +852,8 @@ mod tests {
852852
.await
853853
.expect("execute delete");
854854

855-
let after = query_all(&ctx, "SELECT id FROM users").await;
856-
assert_eq!(total_rows(&after), before_count - 1);
855+
let after = query_all(&ctx, "SELECT id FROM users WHERE name = 'DeleteMe'").await;
856+
assert_eq!(total_rows(&after), 0);
857857
}
858858

859859
#[tokio::test]
@@ -862,8 +862,8 @@ mod tests {
862862
let mut ctx = SessionContext::new();
863863
register_ci_table(&mut ctx, "users").await;
864864

865-
let before = query_all(&ctx, "SELECT id FROM users").await;
866-
let before_count = total_rows(&before);
865+
let before = query_all(&ctx, "SELECT id FROM users WHERE id = 1").await;
866+
assert_eq!(total_rows(&before), 1);
867867

868868
ctx.sql("DELETE FROM users WHERE id = 99999")
869869
.await
@@ -872,8 +872,8 @@ mod tests {
872872
.await
873873
.expect("execute delete");
874874

875-
let after = query_all(&ctx, "SELECT id FROM users").await;
876-
assert_eq!(total_rows(&after), before_count);
875+
let after = query_all(&ctx, "SELECT id FROM users WHERE id = 1").await;
876+
assert_eq!(total_rows(&after), 1);
877877
}
878878

879879
// ─── Update tests (integration) ─────────────────────────────────────
@@ -908,8 +908,15 @@ mod tests {
908908
let mut ctx = SessionContext::new();
909909
register_ci_table(&mut ctx, "users").await;
910910

911-
let before = query_all(&ctx, "SELECT id, name, email FROM users ORDER BY id").await;
912-
let before_count = total_rows(&before);
911+
let before = query_all(&ctx, "SELECT email FROM users WHERE id = 1").await;
912+
assert_eq!(total_rows(&before), 1);
913+
let before_email = before[0]
914+
.column(0)
915+
.as_any()
916+
.downcast_ref::<arrow::array::StringArray>()
917+
.unwrap()
918+
.value(0)
919+
.to_string();
913920

914921
ctx.sql("UPDATE users SET email = 'nobody@example.com' WHERE id = 99999")
915922
.await
@@ -918,8 +925,15 @@ mod tests {
918925
.await
919926
.expect("execute update");
920927

921-
let after = query_all(&ctx, "SELECT id FROM users").await;
922-
assert_eq!(total_rows(&after), before_count);
928+
let after = query_all(&ctx, "SELECT email FROM users WHERE id = 1").await;
929+
assert_eq!(total_rows(&after), 1);
930+
let after_email = after[0]
931+
.column(0)
932+
.as_any()
933+
.downcast_ref::<arrow::array::StringArray>()
934+
.unwrap()
935+
.value(0);
936+
assert_eq!(after_email, before_email);
923937
}
924938

925939
// ─── Combined DML test (integration) ────────────────────────────────
@@ -930,18 +944,15 @@ mod tests {
930944
let mut ctx = SessionContext::new();
931945
register_ci_table(&mut ctx, "users").await;
932946

933-
let before = query_all(&ctx, "SELECT id FROM users").await;
934-
let before_count = total_rows(&before);
935-
936947
// 1. Insert
937948
ctx.sql("INSERT INTO users (name, email) VALUES ('RoundTrip', 'roundtrip@example.com')")
938949
.await
939950
.unwrap()
940951
.collect()
941952
.await
942953
.unwrap();
943-
let after_insert = query_all(&ctx, "SELECT id FROM users").await;
944-
assert_eq!(total_rows(&after_insert), before_count + 1);
954+
let after_insert = query_all(&ctx, "SELECT id FROM users WHERE name = 'RoundTrip'").await;
955+
assert_eq!(total_rows(&after_insert), 1);
945956

946957
// 2. Update
947958
ctx.sql(
@@ -967,8 +978,8 @@ mod tests {
967978
.collect()
968979
.await
969980
.unwrap();
970-
let after_delete = query_all(&ctx, "SELECT id FROM users").await;
971-
assert_eq!(total_rows(&after_delete), before_count);
981+
let after_delete = query_all(&ctx, "SELECT id FROM users WHERE name = 'RoundTrip'").await;
982+
assert_eq!(total_rows(&after_delete), 0);
972983
}
973984

974985
// ─── Multi-table tests (integration) ────────────────────────────────
@@ -984,7 +995,7 @@ mod tests {
984995
"SELECT id, user_id, product, amount FROM orders ORDER BY id",
985996
)
986997
.await;
987-
assert_eq!(total_rows(&batches), 3);
998+
assert!(total_rows(&batches) >= 3);
988999
}
9891000

9901001
#[tokio::test]
@@ -1002,7 +1013,7 @@ mod tests {
10021013
ORDER BY o.id",
10031014
)
10041015
.await;
1005-
assert_eq!(total_rows(&batches), 3);
1016+
assert!(total_rows(&batches) >= 3);
10061017
}
10071018

10081019
#[tokio::test]

‎crates/sources/src/providers/redis/datasource.rs‎

Lines changed: 66 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -1733,22 +1733,6 @@ mod tests {
17331733
.unwrap_or_else(|e| panic!("register {} failed: {}", table, e));
17341734
}
17351735

1736-
/// Register a Redis table with declared column schema (for empty tables).
1737-
fn register_ci_table_with_columns(
1738-
ctx: &mut SessionContext,
1739-
table: &str,
1740-
key_column: &str,
1741-
columns: &str,
1742-
) {
1743-
let mut options = HashMap::new();
1744-
options.insert("key_space".to_string(), "mydb".to_string());
1745-
options.insert("table".to_string(), table.to_string());
1746-
options.insert("key_column".to_string(), key_column.to_string());
1747-
options.insert("columns".to_string(), columns.to_string());
1748-
register_redis_tables(ctx, table, "redis://127.0.0.1:6379", Some(&options))
1749-
.unwrap_or_else(|e| panic!("register {} failed: {}", table, e));
1750-
}
1751-
17521736
async fn ci_query_all(ctx: &SessionContext, sql: &str) -> Vec<RecordBatch> {
17531737
let df = ctx.sql(sql).await.expect("parse sql");
17541738
df.collect().await.expect("collect results")
@@ -1771,7 +1755,7 @@ mod tests {
17711755
"SELECT product_id, name, category, price, in_stock FROM products ORDER BY product_id",
17721756
)
17731757
.await;
1774-
assert_eq!(ci_total_rows(&batches), 5);
1758+
assert!(ci_total_rows(&batches) >= 5);
17751759
}
17761760

17771761
#[tokio::test]
@@ -1781,7 +1765,7 @@ mod tests {
17811765
register_ci_table(&mut ctx, "products");
17821766

17831767
let batches = ci_query_all(&ctx, "SELECT name FROM products ORDER BY product_id").await;
1784-
assert_eq!(ci_total_rows(&batches), 5);
1768+
assert!(ci_total_rows(&batches) >= 5);
17851769
assert_eq!(batches[0].num_columns(), 1);
17861770
}
17871771

@@ -1861,8 +1845,12 @@ mod tests {
18611845
.await
18621846
.unwrap();
18631847

1864-
let before = ci_query_all(&ctx, "SELECT product_id FROM products").await;
1865-
let before_count = ci_total_rows(&before);
1848+
let before = ci_query_all(
1849+
&ctx,
1850+
"SELECT product_id FROM products WHERE product_id = 'PROD_DEL_TEST'",
1851+
)
1852+
.await;
1853+
assert_eq!(ci_total_rows(&before), 1);
18661854

18671855
ctx.sql("DELETE FROM products WHERE product_id = 'PROD_DEL_TEST'")
18681856
.await
@@ -1871,8 +1859,12 @@ mod tests {
18711859
.await
18721860
.expect("execute delete");
18731861

1874-
let after = ci_query_all(&ctx, "SELECT product_id FROM products").await;
1875-
assert_eq!(ci_total_rows(&after), before_count - 1);
1862+
let after = ci_query_all(
1863+
&ctx,
1864+
"SELECT product_id FROM products WHERE product_id = 'PROD_DEL_TEST'",
1865+
)
1866+
.await;
1867+
assert_eq!(ci_total_rows(&after), 0);
18761868
}
18771869

18781870
#[tokio::test]
@@ -1881,8 +1873,12 @@ mod tests {
18811873
let mut ctx = SessionContext::new();
18821874
register_ci_table(&mut ctx, "products");
18831875

1884-
let before = ci_query_all(&ctx, "SELECT product_id FROM products").await;
1885-
let before_count = ci_total_rows(&before);
1876+
let before = ci_query_all(
1877+
&ctx,
1878+
"SELECT product_id FROM products WHERE product_id = 'PROD001'",
1879+
)
1880+
.await;
1881+
assert_eq!(ci_total_rows(&before), 1);
18861882

18871883
ctx.sql("DELETE FROM products WHERE product_id = 'NONEXISTENT'")
18881884
.await
@@ -1891,8 +1887,12 @@ mod tests {
18911887
.await
18921888
.expect("execute delete");
18931889

1894-
let after = ci_query_all(&ctx, "SELECT product_id FROM products").await;
1895-
assert_eq!(ci_total_rows(&after), before_count);
1890+
let after = ci_query_all(
1891+
&ctx,
1892+
"SELECT product_id FROM products WHERE product_id = 'PROD001'",
1893+
)
1894+
.await;
1895+
assert_eq!(ci_total_rows(&after), 1);
18961896
}
18971897

18981898
// ─── Update tests (integration) ─────────────────────────────────────
@@ -1931,8 +1931,19 @@ mod tests {
19311931
let mut ctx = SessionContext::new();
19321932
register_ci_table(&mut ctx, "products");
19331933

1934-
let before = ci_query_all(&ctx, "SELECT product_id FROM products").await;
1935-
let before_count = ci_total_rows(&before);
1934+
let before = ci_query_all(
1935+
&ctx,
1936+
"SELECT name FROM products WHERE product_id = 'PROD001'",
1937+
)
1938+
.await;
1939+
assert_eq!(ci_total_rows(&before), 1);
1940+
let before_name = before[0]
1941+
.column(0)
1942+
.as_any()
1943+
.downcast_ref::<StringArray>()
1944+
.unwrap()
1945+
.value(0)
1946+
.to_string();
19361947

19371948
ctx.sql("UPDATE products SET price = '0.0' WHERE product_id = 'NONEXISTENT'")
19381949
.await
@@ -1941,8 +1952,20 @@ mod tests {
19411952
.await
19421953
.expect("execute update");
19431954

1944-
let after = ci_query_all(&ctx, "SELECT product_id FROM products").await;
1945-
assert_eq!(ci_total_rows(&after), before_count);
1955+
let after = ci_query_all(
1956+
&ctx,
1957+
"SELECT name FROM products WHERE product_id = 'PROD001'",
1958+
)
1959+
.await;
1960+
assert_eq!(ci_total_rows(&after), 1);
1961+
let after_name = after[0]
1962+
.column(0)
1963+
.as_any()
1964+
.downcast_ref::<StringArray>()
1965+
.unwrap()
1966+
.value(0)
1967+
.to_string();
1968+
assert_eq!(before_name, after_name);
19461969
}
19471970

19481971
// ─── Combined DML test (integration) ────────────────────────────────
@@ -1953,9 +1976,6 @@ mod tests {
19531976
let mut ctx = SessionContext::new();
19541977
register_ci_table(&mut ctx, "products");
19551978

1956-
let before = ci_query_all(&ctx, "SELECT product_id FROM products").await;
1957-
let before_count = ci_total_rows(&before);
1958-
19591979
// 1. Insert
19601980
ctx.sql(
19611981
"INSERT INTO products (product_id, name, category, price, in_stock)
@@ -1966,8 +1986,12 @@ mod tests {
19661986
.collect()
19671987
.await
19681988
.unwrap();
1969-
let after_insert = ci_query_all(&ctx, "SELECT product_id FROM products").await;
1970-
assert_eq!(ci_total_rows(&after_insert), before_count + 1);
1989+
let after_insert = ci_query_all(
1990+
&ctx,
1991+
"SELECT product_id FROM products WHERE product_id = 'PROD_RT_TEST'",
1992+
)
1993+
.await;
1994+
assert_eq!(ci_total_rows(&after_insert), 1);
19711995

19721996
// 2. Update
19731997
ctx.sql("UPDATE products SET name = 'RoundTripUpdated', price = '20.0' WHERE product_id = 'PROD_RT_TEST'")
@@ -1981,6 +2005,7 @@ mod tests {
19812005
"SELECT name, price FROM products WHERE product_id = 'PROD_RT_TEST'",
19822006
)
19832007
.await;
2008+
assert_eq!(ci_total_rows(&batches), 1);
19842009
let names = batches[0]
19852010
.column(0)
19862011
.as_any()
@@ -1995,8 +2020,12 @@ mod tests {
19952020
.collect()
19962021
.await
19972022
.unwrap();
1998-
let after_delete = ci_query_all(&ctx, "SELECT product_id FROM products").await;
1999-
assert_eq!(ci_total_rows(&after_delete), before_count);
2023+
let after_delete = ci_query_all(
2024+
&ctx,
2025+
"SELECT product_id FROM products WHERE product_id = 'PROD_RT_TEST'",
2026+
)
2027+
.await;
2028+
assert_eq!(ci_total_rows(&after_delete), 0);
20002029
}
20012030

20022031
// ─── Non-key column filter test (integration) ───────────────────────

0 commit comments

Comments
 (0)