diff --git a/Cargo.lock b/Cargo.lock index 27dfd5fef9..2b20b22c74 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1682,6 +1682,7 @@ dependencies = [ "reqwest 0.12.28", "serde", "serde_json", + "tempfile", "tokio", "uuid", ] diff --git a/crates/dbx-cli/Cargo.toml b/crates/dbx-cli/Cargo.toml index d272aaa4ba..a81c00834b 100644 --- a/crates/dbx-cli/Cargo.toml +++ b/crates/dbx-cli/Cargo.toml @@ -15,3 +15,6 @@ serde_json = "1.0" tokio = { version = "1", features = ["full"] } reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } uuid = { version = "1", features = ["v4", "serde"] } + +[dev-dependencies] +tempfile = "3" diff --git a/crates/dbx-cli/src/commands.rs b/crates/dbx-cli/src/commands.rs index 9ab5ab8d04..e97350dd6d 100644 --- a/crates/dbx-cli/src/commands.rs +++ b/crates/dbx-cli/src/commands.rs @@ -1,26 +1,219 @@ use dbx_core::cli::{fail, ok, CliEnvelope, CliErrorCode, CliSource}; pub(crate) async fn run(args: Vec) -> Result<(), CliEnvelope<()>> { - let output = match args.as_slice() { - [cmd, ..] if cmd == "context" => context().await, - [cmd, ..] if cmd == "selection" => runtime_required("dbx selection requires DBX GUI runtime."), - [cmd, sub, ..] if cmd == "result" && sub == "current" => { - runtime_required("dbx result current requires DBX GUI runtime.") - } - _ => ok(CliSource::Headless, serde_json::json!({ "runtime": "headless" })), - }; - + let output = dispatch(args).await; println!("{}", serde_json::to_string_pretty(&output).unwrap()); + + if matches!(output, CliEnvelope::Failure { .. }) { + std::process::exit(1); + } + Ok(()) } -async fn context() -> CliEnvelope { +pub(crate) async fn dispatch(args: Vec) -> CliEnvelope { + let parsed = match parse_args(args) { + Ok(parsed) => parsed, + Err(err) => return err, + }; + + match parsed.positionals.as_slice() { + [cmd, rest @ ..] if cmd == "context" => context(rest).await, + [cmd, sub, rest @ ..] if cmd == "conn" && sub == "list" => conn_list(rest).await, + [cmd, sub, name, rest @ ..] if cmd == "conn" && sub == "show" => conn_show(name, rest).await, + [cmd, sub, rest @ ..] if cmd == "schema" && sub == "snapshot" => schema_snapshot(rest).await, + [cmd, rest @ ..] if cmd == "safe-query" => safe_query(rest).await, + [cmd, rest @ ..] if cmd == "handoff" => handoff(rest).await, + [cmd, rest @ ..] if cmd == "selection" => selection(rest).await, + [cmd, sub, rest @ ..] if cmd == "result" && sub == "current" => result_current(rest).await, + _ => fail(CliSource::Headless, CliErrorCode::InternalError, "Unknown command", false), + } +} + +struct ParsedArgs { + positionals: Vec, +} + +fn parse_args(args: Vec) -> Result> { + let mut positionals = Vec::new(); + let mut index = 0; + + while index < args.len() { + if args[index] == "--format" { + let Some(format) = args.get(index + 1) else { + return Err(fail(CliSource::Headless, CliErrorCode::InternalError, "--format requires a value", true)); + }; + if format != "json" { + return Err(fail(CliSource::Headless, CliErrorCode::InternalError, "Only --format json is supported", true)); + } + index += 2; + } else { + positionals.push(args[index].clone()); + index += 1; + } + } + + Ok(ParsedArgs { positionals }) +} + +async fn context(_args: &[String]) -> CliEnvelope { match crate::runtime_client::get_json("/context").await { Ok(data) => ok(CliSource::GuiRuntime, data), Err(_) => ok(CliSource::Headless, serde_json::json!({ "runtime": "headless" })), } } +async fn conn_list(_args: &[String]) -> CliEnvelope { + ok(CliSource::Headless, serde_json::json!({ "connections": [] })) +} + +async fn conn_show(_name: &str, _args: &[String]) -> CliEnvelope { + ok(CliSource::Headless, serde_json::json!({})) +} + +async fn schema_snapshot(args: &[String]) -> CliEnvelope { + let Some(_conn) = option_value(args, "--conn") else { + return fail(CliSource::Headless, CliErrorCode::ConnectionNotFound, "--conn is required", true); + }; + + fail(CliSource::Headless, CliErrorCode::InternalError, "Schema snapshot headless execution is not implemented", true) +} + +async fn safe_query(args: &[String]) -> CliEnvelope { + let Some(_conn) = option_value(args, "--conn") else { + return fail(CliSource::Headless, CliErrorCode::ConnectionNotFound, "--conn is required", true); + }; + let Some(_sql) = option_value(args, "--sql") else { + return fail(CliSource::Headless, CliErrorCode::QueryClassificationFailed, "--sql is required", true); + }; + + fail(CliSource::Headless, CliErrorCode::InternalError, "Safe query headless execution is not implemented", true) +} + +async fn handoff(args: &[String]) -> CliEnvelope { + let body = serde_json::json!({ + "conn": option_value(args, "--conn"), + "title": option_value(args, "--title"), + "sql": option_value(args, "--sql"), + }); + + match crate::runtime_client::post_json("/handoff", body).await { + Ok(data) => ok(CliSource::GuiRuntime, data), + Err(_) => fail(CliSource::Headless, CliErrorCode::InternalError, "Handoff payload is required", true), + } +} + +async fn selection(_args: &[String]) -> CliEnvelope { + match crate::runtime_client::get_json("/selection").await { + Ok(data) => ok(CliSource::GuiRuntime, data), + Err(_) => runtime_required("dbx selection requires DBX GUI runtime."), + } +} + +async fn result_current(args: &[String]) -> CliEnvelope { + let limit = option_value(args, "--limit").unwrap_or("50"); + + match crate::runtime_client::get_json(&format!("/result/current?limit={limit}")).await { + Ok(data) => ok(CliSource::GuiRuntime, data), + Err(_) => runtime_required("dbx result current requires DBX GUI runtime."), + } +} + fn runtime_required(message: &str) -> CliEnvelope { fail(CliSource::Headless, CliErrorCode::GuiRuntimeRequired, message, true) } + +fn option_value<'a>(args: &'a [String], key: &str) -> Option<&'a str> { + args.windows(2) + .find(|pair| pair[0] == key) + .map(|pair| pair[1].as_str()) +} + +#[cfg(test)] +mod tests { + use super::*; + use dbx_core::cli::CliErrorCode; + use std::sync::Mutex; + + static ENV_LOCK: Mutex<()> = Mutex::new(()); + + fn assert_failure_code(env: CliEnvelope, expected: CliErrorCode) { + match env { + CliEnvelope::Failure { error, .. } => assert_eq!(error.code, expected), + CliEnvelope::Success { .. } => panic!("expected failure envelope"), + } + } + + #[tokio::test] + async fn gui_only_commands_return_runtime_required_without_runtime() { + let _guard = ENV_LOCK.lock().unwrap(); + let dir = tempfile::tempdir().unwrap(); + std::env::set_var("DBX_APP_DATA_DIR", dir.path()); + + assert_failure_code( + dispatch(vec!["selection".into(), "--format".into(), "json".into()]).await, + CliErrorCode::GuiRuntimeRequired, + ); + assert_failure_code( + dispatch(vec![ + "result".into(), + "current".into(), + "--limit".into(), + "25".into(), + "--format".into(), + "json".into(), + ]) + .await, + CliErrorCode::GuiRuntimeRequired, + ); + + std::env::remove_var("DBX_APP_DATA_DIR"); + } + + #[tokio::test] + async fn recognizes_all_eight_cli_commands_with_json_format() { + let _guard = ENV_LOCK.lock().unwrap(); + let dir = tempfile::tempdir().unwrap(); + std::env::set_var("DBX_APP_DATA_DIR", dir.path()); + + let cases = [ + vec!["context", "--format", "json"], + vec!["conn", "list", "--format", "json"], + vec!["conn", "show", "__missing__", "--redacted", "--format", "json"], + vec!["schema", "snapshot", "--format", "json"], + vec!["safe-query", "--format", "json"], + vec!["handoff", "--format", "json"], + vec!["selection", "--format", "json"], + vec!["result", "current", "--limit", "50", "--format", "json"], + ]; + + for args in cases { + let env = dispatch(args.iter().map(|value| value.to_string()).collect()).await; + let json = serde_json::to_value(&env).unwrap(); + assert!(json.get("ok").is_some(), "missing ok for args: {args:?}"); + assert!(json.get("source").is_some(), "missing source for args: {args:?}"); + assert!( + json.get("data").is_some() || json.get("error").is_some(), + "missing data/error for args: {args:?}" + ); + } + + std::env::remove_var("DBX_APP_DATA_DIR"); + } + + #[tokio::test] + async fn unknown_command_returns_internal_error_envelope() { + assert_failure_code( + dispatch(vec!["not-a-command".into(), "--format".into(), "json".into()]).await, + CliErrorCode::InternalError, + ); + } + + #[tokio::test] + async fn rejects_non_json_format() { + assert_failure_code( + dispatch(vec!["context".into(), "--format".into(), "text".into()]).await, + CliErrorCode::InternalError, + ); + } +} diff --git a/crates/dbx-cli/src/runtime_client.rs b/crates/dbx-cli/src/runtime_client.rs index 940fd53996..45cb1898b8 100644 --- a/crates/dbx-cli/src/runtime_client.rs +++ b/crates/dbx-cli/src/runtime_client.rs @@ -9,6 +9,10 @@ pub struct RuntimeDiscovery { } pub fn app_data_dir() -> PathBuf { + if let Ok(path) = std::env::var("DBX_APP_DATA_DIR") { + return PathBuf::from(path); + } + let home = std::env::var(if cfg!(windows) { "APPDATA" } else { "HOME" }) .unwrap_or_else(|_| ".".to_string()); @@ -31,12 +35,42 @@ pub async fn get_json(path: &str) -> Result { let runtime = load_runtime().ok_or_else(|| "runtime unavailable".to_string())?; let url = format!("http://127.0.0.1:{}{}", runtime.port, path); - reqwest::Client::new() + let response = reqwest::Client::new() .get(url) .bearer_auth(runtime.token) .send() .await - .map_err(|err| err.to_string())? + .map_err(|err| err.to_string())?; + + let status = response.status(); + if !status.is_success() { + return Err(format!("runtime request failed with status {status}")); + } + + response + .json() + .await + .map_err(|err| err.to_string()) +} + +pub async fn post_json(path: &str, body: serde_json::Value) -> Result { + let runtime = load_runtime().ok_or_else(|| "runtime unavailable".to_string())?; + let url = format!("http://127.0.0.1:{}{}", runtime.port, path); + + let response = reqwest::Client::new() + .post(url) + .bearer_auth(runtime.token) + .json(&body) + .send() + .await + .map_err(|err| err.to_string())?; + + let status = response.status(); + if !status.is_success() { + return Err(format!("runtime request failed with status {status}")); + } + + response .json() .await .map_err(|err| err.to_string()) diff --git a/crates/dbx-core/src/cli.rs b/crates/dbx-core/src/cli.rs index 26af58de2c..dec7434300 100644 --- a/crates/dbx-core/src/cli.rs +++ b/crates/dbx-core/src/cli.rs @@ -116,12 +116,8 @@ where let source = source.ok_or_else(|| de::Error::missing_field("source"))?; match (ok, has_data, has_error) { - (true, true, false) => { - Ok(CliEnvelope::Success { ok, source, data: data.expect("data presence was checked") }) - } - (false, false, true) => { - Ok(CliEnvelope::Failure { ok, source, error: error.expect("error presence was checked") }) - } + (true, true, false) => Ok(CliEnvelope::Success { ok, source, data: data.expect("data presence was checked") }), + (false, false, true) => Ok(CliEnvelope::Failure { ok, source, error: error.expect("error presence was checked") }), (true, _, _) => Err(de::Error::custom("ok=true requires data without error")), (false, _, _) => Err(de::Error::custom("ok=false requires error without data")), } diff --git a/crates/dbx-core/src/storage.rs b/crates/dbx-core/src/storage.rs index 4447929499..347caec01e 100644 --- a/crates/dbx-core/src/storage.rs +++ b/crates/dbx-core/src/storage.rs @@ -144,11 +144,10 @@ async fn ensure_history_columns(pool: &SqlitePool) -> Result<(), String> { } async fn ensure_handoffs_sequence(pool: &SqlitePool) -> Result<(), String> { - let (seq_columns,): (i64,) = - sqlx::query_as("SELECT COUNT(*) FROM pragma_table_info('handoffs') WHERE name = 'seq'") - .fetch_one(pool) - .await - .map_err(|e| e.to_string())?; + let (seq_columns,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM pragma_table_info('handoffs') WHERE name = 'seq'") + .fetch_one(pool) + .await + .map_err(|e| e.to_string())?; if seq_columns > 0 { return Ok(()); @@ -1058,8 +1057,10 @@ mod handoff_tests { async fn handoffs_table_has_stable_autoincrement_sequence() { let storage = open_temp_storage().await; - let columns: Vec<(String,)> = - sqlx::query_as("SELECT name FROM pragma_table_info('handoffs')").fetch_all(&storage.db).await.unwrap(); + let columns: Vec<(String,)> = sqlx::query_as("SELECT name FROM pragma_table_info('handoffs')") + .fetch_all(&storage.db) + .await + .unwrap(); assert!(columns.iter().any(|(name,)| name == "seq")); }