1551 lines
42 KiB
Rust
1551 lines
42 KiB
Rust
// RUST_LOG=warn cargo make ci-ws-integration
|
|
mod common;
|
|
|
|
mod ws_integration {
|
|
use std::time::Duration;
|
|
|
|
use serde_json::json;
|
|
use test_log::test;
|
|
use ulid::Ulid;
|
|
|
|
use super::common::{self, PASS, USER};
|
|
use crate::common::error::TestError;
|
|
|
|
#[test(tokio::test)]
|
|
async fn ping() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
// Send command
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "ping",
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res.is_object(), "result: {:?}", res);
|
|
let res = res.as_object().unwrap();
|
|
assert!(res.keys().all(|k| ["id", "result"].contains(&k.as_str())), "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn info() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
let ns = Ulid::new().to_string();
|
|
let db = Ulid::new().to_string();
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = common::ws_use(socket, Some(&ns), Some(&db)).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Setup operations
|
|
//
|
|
let res = common::ws_query(socket, "DEFINE TABLE user PERMISSIONS FULL").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = common::ws_query(
|
|
socket,
|
|
r#"
|
|
DEFINE SCOPE scope SESSION 24h
|
|
SIGNUP ( CREATE user SET user = $user, pass = crypto::argon2::generate($pass) )
|
|
SIGNIN ( SELECT * FROM user WHERE user = $user AND crypto::argon2::compare(pass, $pass) )
|
|
;
|
|
"#,
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = common::ws_query(
|
|
socket,
|
|
r#"
|
|
CREATE user CONTENT {
|
|
user: 'user',
|
|
pass: crypto::argon2::generate('pass')
|
|
};
|
|
"#,
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Sign in
|
|
let res =
|
|
common::ws_signin(socket, "user", "pass", Some(&ns), Some(&db), Some("scope")).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Send the info command
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "info",
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Verify the response contains the expected info
|
|
let res = res.unwrap();
|
|
assert!(res["result"].is_object(), "result: {:?}", res);
|
|
let res = res["result"].as_object().unwrap();
|
|
assert_eq!(res["user"], "user", "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn signup() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
let ns = Ulid::new().to_string();
|
|
let db = Ulid::new().to_string();
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = common::ws_use(socket, Some(&ns), Some(&db)).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Setup scope
|
|
let res = common::ws_query(
|
|
socket,
|
|
r#"
|
|
DEFINE SCOPE scope SESSION 24h
|
|
SIGNUP ( CREATE user SET email = $email, pass = crypto::argon2::generate($pass) )
|
|
SIGNIN ( SELECT * FROM user WHERE email = $email AND crypto::argon2::compare(pass, $pass) )
|
|
;"#,
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Signup
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "signup",
|
|
"params": [{
|
|
"ns": ns,
|
|
"db": db,
|
|
"sc": "scope",
|
|
"email": "email@email.com",
|
|
"pass": "pass",
|
|
}],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res.is_object(), "result: {:?}", res);
|
|
let res = res.as_object().unwrap();
|
|
|
|
// Verify response contains no error
|
|
assert!(res.keys().all(|k| ["id", "result"].contains(&k.as_str())), "result: {:?}", res);
|
|
|
|
// Verify it returns a token
|
|
assert!(res["result"].is_string(), "result: {:?}", res);
|
|
|
|
let res = res["result"].as_str().unwrap();
|
|
assert!(res.starts_with("eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzUxMiJ9"), "result: {}", res);
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn signin() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
let ns = Ulid::new().to_string();
|
|
let db = Ulid::new().to_string();
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = common::ws_use(socket, Some(&ns), Some(&db)).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Setup scope
|
|
let res = common::ws_query(
|
|
socket,
|
|
r#"
|
|
DEFINE SCOPE scope SESSION 24h
|
|
SIGNUP ( CREATE user SET email = $email, pass = crypto::argon2::generate($pass) )
|
|
SIGNIN ( SELECT * FROM user WHERE email = $email AND crypto::argon2::compare(pass, $pass) )
|
|
;"#,
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Signup
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "signup",
|
|
"params": [{
|
|
"ns": ns,
|
|
"db": db,
|
|
"sc": "scope",
|
|
"email": "email@email.com",
|
|
"pass": "pass",
|
|
}],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Sign in
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "signin",
|
|
"params": [{
|
|
"ns": ns,
|
|
"db": db,
|
|
"sc": "scope",
|
|
"email": "email@email.com",
|
|
"pass": "pass",
|
|
}],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res.is_object(), "result: {:?}", res);
|
|
let res = res.as_object().unwrap();
|
|
|
|
// Verify response contains no error
|
|
assert!(res.keys().all(|k| ["id", "result"].contains(&k.as_str())), "result: {:?}", res);
|
|
|
|
// Verify it returns a token
|
|
assert!(res["result"].is_string(), "result: {:?}", res);
|
|
let res = res["result"].as_str().unwrap();
|
|
assert!(res.starts_with("eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzUxMiJ9"), "result: {}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn variable_auth_live_query() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
let ns = Ulid::new().to_string();
|
|
let db = Ulid::new().to_string();
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = common::ws_use(socket, Some(&ns), Some(&db)).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Setup scope
|
|
let res = common::ws_query(socket, r#"
|
|
DEFINE SCOPE scope SESSION 2s
|
|
SIGNUP ( CREATE user SET user = $user, pass = crypto::argon2::generate($pass) )
|
|
SIGNIN ( SELECT * FROM user WHERE user = $user AND crypto::argon2::compare(pass, $pass) )
|
|
;"#).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Signup
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "signup",
|
|
"params": [{
|
|
"ns": ns,
|
|
"db": db,
|
|
"sc": "scope",
|
|
"user": "user",
|
|
"pass": "pass",
|
|
}],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Sign in
|
|
let res =
|
|
common::ws_signin(socket, "user", "pass", Some(&ns), Some(&db), Some("scope")).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
|
|
// Start Live Query
|
|
let table_name = "test_tableBB4B0A788C7E46E798720AEF938CBCF6";
|
|
let _live_query_response = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "66BB05C8-EF4B-4338-BCCD-8F8A19223CB1",
|
|
"method": "live",
|
|
"params": [
|
|
table_name
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap_or_else(|e| panic!("Error sending message: {}", e))
|
|
.as_object()
|
|
.unwrap_or_else(|| panic!("Expected object, got {:?}", res));
|
|
|
|
// Wait 2 seconds for auth to expire
|
|
tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
|
|
|
|
// Start second connection
|
|
let socket2 = &mut common::connect_ws(&addr).await?;
|
|
|
|
// Signin
|
|
let res =
|
|
common::ws_signin(socket2, "user", "pass", Some(&ns), Some(&db), Some("scope")).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Insert
|
|
let id = "A23A05ABC15C420E9A7E13D2C8657890";
|
|
let query = format!(r#"INSERT INTO {} {{"id": "{}", "name": "ok"}};"#, table_name, id);
|
|
let created = common::ws_query(socket2, query.as_str()).await.unwrap();
|
|
assert_eq!(created.len(), 1);
|
|
|
|
// Validate live query from first session didnt produce a result
|
|
let res = common::ws_recv_msg(socket).await;
|
|
match &res {
|
|
Err(e) => {
|
|
if let Some(TestError::NetworkError {
|
|
..
|
|
}) = e.downcast_ref::<TestError>()
|
|
{
|
|
} else {
|
|
panic!("Expected a network error, but got: {:?}", e)
|
|
}
|
|
}
|
|
Ok(v) => {
|
|
panic!("Expected a network error, but got: {:?}", v)
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn invalidate() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Verify we have a ROOT session
|
|
let res = common::ws_query(
|
|
socket,
|
|
&format!("DEFINE NAMESPACE {throwaway}", throwaway = Ulid::new()),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Invalidate session
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "invalidate",
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Verify we invalidated the root session
|
|
let res = common::ws_query(
|
|
socket,
|
|
&format!("DEFINE NAMESPACE {throwaway}", throwaway = Ulid::new()),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
|
|
assert_eq!(res[0]["status"], "ERR", "result: {:?}", res);
|
|
assert_eq!(
|
|
res[0]["result"], "IAM error: Not enough permissions to perform this action",
|
|
"result: {:?}",
|
|
res
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn authenticate() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let token = res.unwrap();
|
|
|
|
// Reconnect so we start with an empty session
|
|
socket.close(None).await?;
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Authenticate with the token
|
|
//
|
|
|
|
// Send command
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "authenticate",
|
|
"params": [
|
|
token,
|
|
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Verify we have a ROOT session
|
|
let res = common::ws_query(
|
|
socket,
|
|
&format!("DEFINE NAMESPACE {throwaway}", throwaway = Ulid::new()),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert_eq!(res[0]["status"], "OK", "result: {:?}", res);
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn kill_kill_endpoint() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let table_name = "table_D250F804BC244558982DB7D8712F6BE3".to_string();
|
|
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
let _ = common::ws_signin(socket, USER, PASS, None, None, None).await?;
|
|
let _ =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await?;
|
|
|
|
// LIVE query via live endpoint
|
|
let live_res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "live",
|
|
"params": [
|
|
table_name
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await?;
|
|
let live_id = live_res["result"].as_str().unwrap();
|
|
|
|
// KILL query via kill endpoint
|
|
common::ws_send_msg(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "kill",
|
|
"params": [
|
|
live_id
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await?;
|
|
|
|
// Verify we killed the query
|
|
let msgs = common::ws_recv_all_msgs(socket, 1, Duration::from_millis(1000)).await?;
|
|
assert!(
|
|
msgs.iter().all(|v| v["error"].is_null()),
|
|
"Unexpected error received: {:#?}",
|
|
msgs
|
|
);
|
|
let msg = msgs.get(0).unwrap();
|
|
assert!(msg["status"].is_null(), "unexpected status: {:?}", msg);
|
|
|
|
// Create some data for notification
|
|
let id = "an-id-goes-here";
|
|
let query = format!(r#"INSERT INTO {} {{"id": "{}", "name": "ok"}};"#, table_name, id);
|
|
let _ = common::ws_query(socket, query.as_str()).await.unwrap();
|
|
let json = json!({
|
|
"id": "1",
|
|
"method": "query",
|
|
"params": [query],
|
|
});
|
|
|
|
common::ws_send_msg(socket, serde_json::to_string(&json).unwrap()).await?;
|
|
|
|
// Wait some time for all messages to arrive, and then verify we didn't get any notification
|
|
let msgs = common::ws_recv_all_msgs(socket, 1, Duration::from_millis(500)).await?;
|
|
assert!(
|
|
msgs.iter().all(|v| v["error"].is_null()),
|
|
"Unexpected error received: {:#?}",
|
|
msgs
|
|
);
|
|
let lq_notif = msgs.iter().find(|v| common::ws_msg_is_notification_from_lq(v, live_id));
|
|
assert!(lq_notif.is_none(), "Expected to find no notifications, found 1: {:#?}", msgs);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn kill_query_endpoint() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let table_name = "table_8B5E5635869E4FF2A35C94E8FC2CAE9A".to_string();
|
|
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
let _ = common::ws_signin(socket, USER, PASS, None, None, None).await?;
|
|
let _ =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await?;
|
|
|
|
// LIVE query via live endpoint
|
|
let live_res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "live",
|
|
"params": [
|
|
table_name
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await?;
|
|
let live_id = live_res["result"].as_str().unwrap();
|
|
|
|
// KILL query via kill endpoint
|
|
let kill_query = format!("KILL '{live_id}'");
|
|
common::ws_send_msg(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "query",
|
|
"params": [
|
|
kill_query
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await?;
|
|
|
|
// Verify we killed the query
|
|
let msgs = common::ws_recv_all_msgs(socket, 1, Duration::from_millis(1000)).await?;
|
|
assert!(
|
|
msgs.iter().all(|v| v["error"].is_null()),
|
|
"Unexpected error received: {:#?}",
|
|
msgs
|
|
);
|
|
let msg = msgs.get(0).unwrap();
|
|
assert!(msg["status"].is_null(), "unexpected status: {:?}", msg);
|
|
|
|
// Create some data for notification
|
|
let id = "an-id-goes-here";
|
|
let query = format!(r#"INSERT INTO {} {{"id": "{}", "name": "ok"}};"#, table_name, id);
|
|
let _ = common::ws_query(socket, query.as_str()).await.unwrap();
|
|
let json = json!({
|
|
"id": "1",
|
|
"method": "query",
|
|
"params": [query],
|
|
});
|
|
|
|
common::ws_send_msg(socket, serde_json::to_string(&json).unwrap()).await?;
|
|
|
|
// Wait some time for all messages to arrive, and then verify we didn't get any notification
|
|
let msgs = common::ws_recv_all_msgs(socket, 1, Duration::from_millis(500)).await?;
|
|
assert!(
|
|
msgs.iter().all(|v| v["error"].is_null()),
|
|
"Unexpected error received: {:#?}",
|
|
msgs
|
|
);
|
|
let lq_notif = msgs.iter().find(|v| common::ws_msg_is_notification_from_lq(v, live_id));
|
|
assert!(lq_notif.is_none(), "Expected to find no notifications, found 1: {:#?}", msgs);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn live_live_endpoint() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_without_auth().await.unwrap();
|
|
let table_name = "table_FD40A9A361884C56B5908A934164884A".to_string();
|
|
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
let _ = common::ws_signin(socket, USER, PASS, None, None, None).await?;
|
|
let _ =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await?;
|
|
|
|
// LIVE query via live endpoint
|
|
let live_res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "live",
|
|
"params": [
|
|
table_name
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await?;
|
|
let live_id = live_res["result"].as_str().unwrap();
|
|
|
|
// Create some data for notification
|
|
// Manually send the query and wait for multiple messages. Ordering of the messages is not guaranteed, so we could receive the notification before the query result.
|
|
let id = "an-id-goes-here";
|
|
let query = format!(r#"INSERT INTO {} {{"id": "{}", "name": "ok"}};"#, table_name, id);
|
|
let json = json!({
|
|
"id": "1",
|
|
"method": "query",
|
|
"params": [query],
|
|
});
|
|
common::ws_send_msg(socket, serde_json::to_string(&json).unwrap()).await?;
|
|
|
|
// Wait some time for all messages to arrive, and then search for the notification message
|
|
let msgs = common::ws_recv_all_msgs(socket, 2, Duration::from_millis(500)).await;
|
|
assert!(msgs.is_ok(), "Error waiting for messages: {:?}", msgs.err());
|
|
let msgs = msgs.unwrap();
|
|
assert!(
|
|
msgs.iter().all(|v| v["error"].is_null()),
|
|
"Unexpected error received: {:#?}",
|
|
msgs
|
|
);
|
|
|
|
let lq_notif = msgs.iter().find(|v| common::ws_msg_is_notification_from_lq(v, live_id));
|
|
assert!(
|
|
lq_notif.is_some(),
|
|
"Expected to find a notification for LQ id {}: {:#?}",
|
|
live_id,
|
|
msgs
|
|
);
|
|
// Extract the notification object
|
|
let lq_notif = lq_notif.unwrap();
|
|
let lq_notif = lq_notif["result"].as_object().unwrap();
|
|
|
|
// Verify message on individual keys since the notification ID is random
|
|
let action = lq_notif["action"].as_str().unwrap();
|
|
let result = lq_notif["result"].as_object().unwrap();
|
|
assert_eq!(action, "CREATE", "expected notification to be `CREATE`: {:?}", lq_notif);
|
|
let expected_id = format!("{}:⟨{}⟩", table_name, id);
|
|
assert_eq!(
|
|
result["id"].as_str(),
|
|
Some(expected_id.as_str()),
|
|
"expected notification to have id {:?}: {:?}",
|
|
expected_id,
|
|
lq_notif
|
|
);
|
|
assert_eq!(
|
|
result["name"].as_str(),
|
|
Some("ok"),
|
|
"expected notification to have name `ok`: {:?}",
|
|
lq_notif
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn live_query_endpoint() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_without_auth().await.unwrap();
|
|
let table_name = "table_FD40A9A361884C56B5908A934164884A".to_string();
|
|
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
let _ = common::ws_signin(socket, USER, PASS, None, None, None).await?;
|
|
let _ =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await?;
|
|
|
|
// LIVE query via query endpoint
|
|
let lq_res =
|
|
common::ws_query(socket, format!("LIVE SELECT * FROM {};", table_name).as_str())
|
|
.await?;
|
|
assert_eq!(lq_res.len(), 1, "Expected 1 result got: {:?}", lq_res);
|
|
let live_id = lq_res[0]["result"].as_str().unwrap();
|
|
|
|
// Create some data for notification
|
|
// Manually send the query and wait for multiple messages. Ordering of the messages is not guaranteed, so we could receive the notification before the query result.
|
|
let id = "an-id-goes-here";
|
|
let query = format!(r#"INSERT INTO {} {{"id": "{}", "name": "ok"}};"#, table_name, id);
|
|
let json = json!({
|
|
"id": "1",
|
|
"method": "query",
|
|
"params": [query],
|
|
});
|
|
|
|
common::ws_send_msg(socket, serde_json::to_string(&json).unwrap()).await?;
|
|
|
|
// Wait some time for all messages to arrive, and then search for the notification message
|
|
let msgs = common::ws_recv_all_msgs(socket, 2, Duration::from_millis(500)).await;
|
|
assert!(msgs.is_ok(), "Error waiting for messages: {:?}", msgs.err());
|
|
let msgs = msgs.unwrap();
|
|
assert!(
|
|
msgs.iter().all(|v| v["error"].is_null()),
|
|
"Unexpected error received: {:#?}",
|
|
msgs
|
|
);
|
|
|
|
let lq_notif = msgs.iter().find(|v| common::ws_msg_is_notification_from_lq(v, live_id));
|
|
assert!(
|
|
lq_notif.is_some(),
|
|
"Expected to find a notification for LQ id {}: {:#?}",
|
|
live_id,
|
|
msgs
|
|
);
|
|
|
|
// Extract the notification object
|
|
let lq_notif = lq_notif.unwrap();
|
|
let lq_notif = lq_notif["result"].as_object().unwrap();
|
|
|
|
// Verify message on individual keys since the notification ID is random
|
|
let action = lq_notif["action"].as_str().unwrap();
|
|
let result = lq_notif["result"].as_object().unwrap();
|
|
assert_eq!(action, "CREATE", "expected notification to be `CREATE`: {:?}", lq_notif);
|
|
let expected_id = format!("{}:⟨{}⟩", table_name, id);
|
|
assert_eq!(
|
|
result["id"].as_str(),
|
|
Some(expected_id.as_str()),
|
|
"expected notification to have id {:?}: {:?}",
|
|
expected_id,
|
|
lq_notif
|
|
);
|
|
assert_eq!(
|
|
result["name"].as_str(),
|
|
Some("ok"),
|
|
"expected notification to have name `ok`: {:?}",
|
|
lq_notif
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn let_and_set() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Define variable using let
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "let",
|
|
"params": [
|
|
"let_var", "let_value",
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Define variable using set
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "set",
|
|
"params": [
|
|
"set_var", "set_value",
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Verify the variables are set
|
|
let res = common::ws_query(socket, "SELECT * FROM $let_var, $set_var").await?;
|
|
assert_eq!(
|
|
res[0]["result"],
|
|
serde_json::to_value(["let_value", "set_value"]).unwrap(),
|
|
"result: {:?}",
|
|
res
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn unset() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Define variable
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "let",
|
|
"params": [
|
|
"let_var", "let_value",
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Verify the variable is set
|
|
let res = common::ws_query(socket, "SELECT * FROM $let_var").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = res[0]["result"].as_array().unwrap();
|
|
|
|
assert_eq!(res[0], "let_value", "result: {:?}", res);
|
|
|
|
// Unset variable
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "unset",
|
|
"params": [
|
|
"let_var",
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Verify the variable is unset
|
|
let res = common::ws_query(socket, "SELECT * FROM $let_var").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = res[0]["result"].as_array().unwrap();
|
|
|
|
assert!(res[0].is_null(), "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn select() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Setup the database
|
|
//
|
|
let res = common::ws_query(socket, "CREATE foo").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Select data
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "select",
|
|
"params": [
|
|
"foo",
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res["result"].is_array(), "result: {:?}", res);
|
|
let res = res["result"].as_array().unwrap();
|
|
|
|
// Verify the response contains the output of the select
|
|
assert_eq!(res.len(), 1, "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn insert() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Insert data
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "insert",
|
|
"params": [
|
|
"table",
|
|
{
|
|
"name": "foo",
|
|
"value": "bar",
|
|
}
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Verify the data was inserted and can be queried
|
|
let res = common::ws_query(socket, "SELECT * FROM table").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = res[0]["result"].as_array().unwrap();
|
|
|
|
assert_eq!(res[0]["name"], "foo", "result: {:?}", res);
|
|
assert_eq!(res[0]["value"], "bar", "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn create() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Insert data
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "create",
|
|
"params": [
|
|
"table",
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Verify the data was created and can be queried
|
|
let res = common::ws_query(socket, "SELECT * FROM table").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = res[0]["result"].as_array().unwrap();
|
|
assert_eq!(res.len(), 1, "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn update() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Setup the database
|
|
//
|
|
let res = common::ws_query(socket, r#"CREATE table SET name = "foo""#).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Insert data
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "update",
|
|
"params": [
|
|
"table",
|
|
{
|
|
"value": "bar",
|
|
}
|
|
],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Verify the data was updated
|
|
let res = common::ws_query(socket, "SELECT * FROM table").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = &res[0]["result"].as_array().unwrap()[0];
|
|
assert!(res["name"].is_null(), "result: {:?}", res);
|
|
assert_eq!(res["value"], "bar", "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn merge() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Setup the database
|
|
//
|
|
let res = common::ws_query(socket, "CREATE foo").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = &res[0]["result"].as_array().unwrap()[0];
|
|
|
|
assert!(res["id"].is_string(), "result: {:?}", res);
|
|
let what = res["id"].as_str().unwrap();
|
|
|
|
//
|
|
// Merge data
|
|
//
|
|
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "merge",
|
|
"params": [
|
|
what, {
|
|
"name_for_merge": "foo",
|
|
"value_for_merge": "bar",
|
|
}
|
|
]
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Get the data and verify the output
|
|
//
|
|
let res = common::ws_query(socket, &format!("SELECT * FROM foo WHERE id = {what}")).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = &res[0]["result"].as_array().unwrap()[0];
|
|
|
|
assert_eq!(res["id"], what);
|
|
assert_eq!(res["name_for_merge"], "foo");
|
|
assert_eq!(res["value_for_merge"], "bar");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn patch() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Setup the database
|
|
//
|
|
let res =
|
|
common::ws_query(socket, r#"CREATE table SET original_name = "original_value""#).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = &res[0]["result"].as_array().unwrap()[0];
|
|
|
|
let what = res["id"].as_str().unwrap();
|
|
|
|
//
|
|
// Patch data
|
|
//
|
|
|
|
let ops = json!([
|
|
{
|
|
"op": "add",
|
|
"path": "patch_name",
|
|
"value": "patch_value"
|
|
},
|
|
{
|
|
"op": "remove",
|
|
"path": "original_name",
|
|
}
|
|
]);
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "patch",
|
|
"params": [
|
|
what, ops
|
|
]
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res["result"].is_object(), "result: {:?}", res);
|
|
let res = res["result"].as_object().unwrap();
|
|
assert_eq!(res.get("patch_name"), Some(json!("patch_value")).as_ref(), "result: {:?}", res);
|
|
|
|
//
|
|
// Get the data and verify the output
|
|
//
|
|
let res = common::ws_query(socket, &format!("SELECT * FROM table WHERE id = {what}")).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = &res[0]["result"].as_array().unwrap()[0];
|
|
|
|
assert_eq!(res["id"], what);
|
|
assert!(res["original_name"].is_null());
|
|
assert_eq!(res["patch_name"], "patch_value");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn delete() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Setup the database
|
|
//
|
|
let res = common::ws_query(socket, "CREATE table:id").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Verify the data was created and can be queried
|
|
//
|
|
let res = common::ws_query(socket, "SELECT * FROM table:id").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = res[0]["result"].as_array().unwrap();
|
|
|
|
assert_eq!(res.len(), 1, "result: {:?}", res);
|
|
|
|
//
|
|
// Delete data
|
|
//
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "delete",
|
|
"params": [
|
|
"table:id"
|
|
]
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Verify the data was deleted
|
|
//
|
|
let res = common::ws_query(socket, "SELECT * FROM table:id").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = res[0]["result"].as_array().unwrap();
|
|
|
|
assert_eq!(res.len(), 0, "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn format_json() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Setup the database
|
|
//
|
|
let res = common::ws_query(socket, "CREATE table:id").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Test JSON format
|
|
//
|
|
|
|
// Change format
|
|
let msg = json!({
|
|
"id": "1",
|
|
"method": "format",
|
|
"params": [
|
|
"json"
|
|
]
|
|
});
|
|
let res =
|
|
common::ws_send_msg_and_wait_response(socket, serde_json::to_string(&msg).unwrap())
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Query data
|
|
let res = common::ws_query(socket, "SELECT * FROM table:id").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = res[0]["result"].as_array().unwrap();
|
|
assert_eq!(res.len(), 1, "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn format_cbor() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Setup the database
|
|
//
|
|
let res = common::ws_query(socket, "CREATE table:id").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Test CBOR format
|
|
//
|
|
|
|
// Change format
|
|
let msg = serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "format",
|
|
"params": [
|
|
"cbor"
|
|
]
|
|
}))
|
|
.unwrap();
|
|
let res = common::ws_send_msg_and_wait_response(socket, msg).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Query data
|
|
let msg = serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "query",
|
|
"params": [
|
|
"SELECT * FROM table:id"
|
|
]
|
|
}))
|
|
.unwrap();
|
|
|
|
let res = common::ws_send_msg(socket, msg).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
let res = common::ws_recv_msg_with_fmt(socket, common::Format::Cbor).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res["result"].is_array(), "result: {:?}", res);
|
|
let res = res["result"].as_array().unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = res[0]["result"].as_array().unwrap();
|
|
assert_eq!(res.len(), 1, "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn format_pack() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Setup the database
|
|
//
|
|
let res = common::ws_query(socket, "CREATE table:id").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Test PACK format
|
|
//
|
|
|
|
// Change format
|
|
let msg = serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "format",
|
|
"params": [
|
|
"pack"
|
|
]
|
|
}))
|
|
.unwrap();
|
|
let res = common::ws_send_msg_and_wait_response(socket, msg).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
// Query data
|
|
let msg = serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "query",
|
|
"params": [
|
|
"SELECT * FROM table:id"
|
|
]
|
|
}))
|
|
.unwrap();
|
|
|
|
let res = common::ws_send_msg(socket, msg).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = common::ws_recv_msg_with_fmt(socket, common::Format::Pack).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res["result"].is_array(), "result: {:?}", res);
|
|
let res = res["result"].as_array().unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = res[0]["result"].as_array().unwrap();
|
|
assert_eq!(res.len(), 1, "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn query() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Run a CREATE query
|
|
//
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "query",
|
|
"params": [
|
|
"CREATE foo",
|
|
]
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Verify the data was created and can be queried
|
|
//
|
|
let res = common::ws_query(socket, "SELECT * FROM foo").await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res[0]["result"].is_array(), "result: {:?}", res);
|
|
let res = res[0]["result"].as_array().unwrap();
|
|
assert_eq!(res.len(), 1, "result: {:?}", res);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[test(tokio::test)]
|
|
async fn version() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
// Send command
|
|
let res = common::ws_send_msg_and_wait_response(
|
|
socket,
|
|
serde_json::to_string(&json!({
|
|
"id": "1",
|
|
"method": "version",
|
|
"params": [],
|
|
}))
|
|
.unwrap(),
|
|
)
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res = res.unwrap();
|
|
assert!(res["result"].is_string(), "result: {:?}", res);
|
|
let res = res["result"].as_str().unwrap();
|
|
|
|
// Verify response
|
|
assert!(res.starts_with("surrealdb-"), "result: {}", res);
|
|
Ok(())
|
|
}
|
|
|
|
// Validate that the WebSocket is able to process multiple queries concurrently
|
|
#[test(tokio::test)]
|
|
async fn concurrency() -> Result<(), Box<dyn std::error::Error>> {
|
|
let (addr, _server) = common::start_server_with_defaults().await.unwrap();
|
|
let socket = &mut common::connect_ws(&addr).await?;
|
|
|
|
//
|
|
// Prepare the connection
|
|
//
|
|
let res = common::ws_signin(socket, USER, PASS, None, None, None).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
let res =
|
|
common::ws_use(socket, Some(&Ulid::new().to_string()), Some(&Ulid::new().to_string()))
|
|
.await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
|
|
//
|
|
// Run 5 queries that do a SLEEP and verify they all complete concurrently
|
|
//
|
|
|
|
// Send queries
|
|
for i in 0..5 {
|
|
let query = format!("SLEEP 1s; RETURN 'done-{}';", i);
|
|
let query_msg = json!({
|
|
"id": "1",
|
|
"method": "query",
|
|
"params": [query],
|
|
});
|
|
let res = common::ws_send_msg(socket, serde_json::to_string(&query_msg).unwrap()).await;
|
|
assert!(res.is_ok(), "result: {:?}", res);
|
|
}
|
|
|
|
// Wait for queries to complete and verify they all completed within 2 seconds (assume they are executed concurrently)
|
|
let msgs = common::ws_recv_all_msgs(socket, 5, Duration::from_secs(2)).await;
|
|
assert!(msgs.is_ok(), "Error waiting for messages: {:?}", msgs.err());
|
|
|
|
let msgs = msgs.unwrap();
|
|
assert!(
|
|
msgs.iter().all(|v| v["error"].is_null()),
|
|
"Unexpected error received: {:#?}",
|
|
msgs
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
}
|