mirror of
https://github.com/mountain-loop/yaak.git
synced 2026-11-13 12:13:42 +01:00
CLI: handle gRPC and WebSocket requests, keep IDs on Yaak import (#816)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
c0bfe7a97a
commit
eba8dd5864
13 files changed
+1947
-197
No files matched your search
@@ -241,8 +241,12 @@ pub struct ExportArgs {
|
||||
#[arg(long, conflicts_with = "workspace_ids")]
|
||||
pub all: bool,
|
||||
|
||||
/// Include private environments in the export
|
||||
#[arg(long)]
|
||||
/// Include private environments in the export. Environments made in the app are private
|
||||
/// unless marked shareable, so they are left out by default, as the app's export does.
|
||||
#[arg(
|
||||
long = "include-private",
|
||||
visible_alias = "include-private-environments"
|
||||
)]
|
||||
pub include_private_environments: bool,
|
||||
}
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ use yaak::import;
|
||||
use yaak_models::export::{self, ExportDataParams};
|
||||
use yaak_models::util::{
|
||||
BatchUpsertResult, ImportDestination, ImportOrigin, ImportPlanAction, ImportPlanItem,
|
||||
ImportPlanWarning,
|
||||
};
|
||||
use yaak_plugins::events::{ImportRequest, ImportResources, PluginContext};
|
||||
|
||||
@@ -13,11 +14,14 @@ type CommandResult<T = ()> = std::result::Result<T, String>;
|
||||
|
||||
pub async fn run_import(ctx: &CliContext, args: ImportArgs) -> i32 {
|
||||
match import(ctx, args).await {
|
||||
Ok((result, items)) => {
|
||||
Ok((result, items, warnings)) => {
|
||||
println!("Imported {}", format_counts(&result));
|
||||
if let Some(skipped) = format_skipped(&items) {
|
||||
println!("Skipped {skipped}");
|
||||
}
|
||||
for warning in warnings {
|
||||
eprintln!("Warning: {}: {}", warning.title, warning.detail);
|
||||
}
|
||||
0
|
||||
}
|
||||
Err(error) => {
|
||||
@@ -43,7 +47,7 @@ pub fn run_export(ctx: &CliContext, args: ExportArgs) -> i32 {
|
||||
async fn import(
|
||||
ctx: &CliContext,
|
||||
args: ImportArgs,
|
||||
) -> CommandResult<(BatchUpsertResult, Vec<ImportPlanItem>)> {
|
||||
) -> CommandResult<(BatchUpsertResult, Vec<ImportPlanItem>, Vec<ImportPlanWarning>)> {
|
||||
if let Some(workspace_id) = args.workspace_id.as_deref() {
|
||||
ctx.db()
|
||||
.get_workspace(workspace_id)
|
||||
@@ -67,6 +71,15 @@ async fn import(
|
||||
.to_string(),
|
||||
);
|
||||
}
|
||||
// Everything would land in the one workspace, merging separate workspaces into it
|
||||
if workspace_id.is_some() && resources.workspaces.len() > 1 {
|
||||
return Err(format!(
|
||||
"{} contains {} workspaces, and --workspace-id imports into a single one. Run `yaak import {}` without --workspace-id to import each as its own workspace, or export one workspace at a time.",
|
||||
args.file.display(),
|
||||
resources.workspaces.len(),
|
||||
args.file.display(),
|
||||
));
|
||||
}
|
||||
let destination = match workspace_id {
|
||||
Some(workspace_id) => {
|
||||
ImportDestination::ExistingWorkspace { workspace_id, folder_id: None }
|
||||
@@ -83,9 +96,10 @@ async fn import(
|
||||
)
|
||||
.map_err(|e| format!("Failed to plan import: {e}"))?;
|
||||
let items = plan.items.clone();
|
||||
let warnings = plan.warnings.clone();
|
||||
let imported = import::commit_import_plan(ctx.query_manager(), plan)
|
||||
.map_err(|e| format!("Failed to import data: {e}"))?;
|
||||
Ok((imported, items))
|
||||
Ok((imported, items, warnings))
|
||||
}
|
||||
|
||||
fn file_origin(path: &std::path::Path) -> ImportOrigin {
|
||||
|
||||
@@ -646,7 +646,43 @@ mod tests {
|
||||
fx.ok("delete_http_request", json!({ "id": id })).await,
|
||||
format!("Deleted: Created ({id})")
|
||||
);
|
||||
assert_eq!(fx.json("get_http_request", json!({ "id": id })).await, Value::Null);
|
||||
assert_eq!(
|
||||
fx.err("get_http_request", json!({ "id": id })).await,
|
||||
format!("HTTP request with ID {id} not found")
|
||||
);
|
||||
fx.finish().await;
|
||||
}
|
||||
|
||||
/// QA sweep bug 6: `get_http_request` with a `wr_` ID or `rq_bogus` answered `"null"`
|
||||
/// with `isError: false`, and `delete_http_request` on a `wr_` ID failed obscurely.
|
||||
#[tokio::test]
|
||||
async fn http_request_tools_reject_unknown_and_non_http_ids() {
|
||||
let fx = Fixture::new(None).await;
|
||||
let websocket = fx
|
||||
.query_manager
|
||||
.with_tx(|db| {
|
||||
db.upsert_websocket_request(
|
||||
&yaak_models::models::WebsocketRequest {
|
||||
workspace_id: fx.workspace_id.clone(),
|
||||
name: "Socket".into(),
|
||||
..Default::default()
|
||||
},
|
||||
&UpdateSource::Background,
|
||||
)
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
fx.err("get_http_request", json!({ "id": "rq_bogus" })).await,
|
||||
"HTTP request with ID rq_bogus not found"
|
||||
);
|
||||
let expected = format!(
|
||||
"{} is a WebSocket request, and these tools only work with HTTP requests. Use the Yaak app or the `yaak request` CLI commands for it.",
|
||||
websocket.id
|
||||
);
|
||||
assert_eq!(fx.err("get_http_request", json!({ "id": websocket.id })).await, expected);
|
||||
assert_eq!(fx.err("delete_http_request", json!({ "id": websocket.id })).await, expected);
|
||||
assert!(fx.query_manager.connect().get_websocket_request(&websocket.id).is_ok());
|
||||
fx.finish().await;
|
||||
}
|
||||
|
||||
|
||||
@@ -67,18 +67,48 @@ pub async fn run(
|
||||
}
|
||||
}
|
||||
|
||||
/// Find a request of any type, with an error that names the ID rather than a table.
|
||||
pub(crate) fn get_any_request(ctx: &CliContext, request_id: &str) -> CommandResult<AnyRequest> {
|
||||
let db = ctx.db();
|
||||
if let Ok(request) = db.get_http_request(request_id) {
|
||||
return Ok(AnyRequest::HttpRequest(request));
|
||||
}
|
||||
if let Ok(request) = db.get_grpc_request(request_id) {
|
||||
return Ok(AnyRequest::GrpcRequest(request));
|
||||
}
|
||||
if let Ok(request) = db.get_websocket_request(request_id) {
|
||||
return Ok(AnyRequest::WebsocketRequest(request));
|
||||
}
|
||||
Err(format!("Request {request_id} not found"))
|
||||
}
|
||||
|
||||
fn list(ctx: &CliContext, workspace_id: Option<&str>) -> CommandResult {
|
||||
let workspace_id = resolve_workspace_id(ctx, workspace_id, "request list")?;
|
||||
let requests = ctx
|
||||
.db()
|
||||
let db = ctx.db();
|
||||
let http = db
|
||||
.list_http_requests(&workspace_id)
|
||||
.map_err(|e| format!("Failed to list requests: {e}"))?;
|
||||
if requests.is_empty() {
|
||||
.map_err(|e| format!("Failed to list HTTP requests: {e}"))?;
|
||||
let grpc = db
|
||||
.list_grpc_requests(&workspace_id)
|
||||
.map_err(|e| format!("Failed to list gRPC requests: {e}"))?;
|
||||
let websocket = db
|
||||
.list_websocket_requests(&workspace_id)
|
||||
.map_err(|e| format!("Failed to list WebSocket requests: {e}"))?;
|
||||
|
||||
if http.is_empty() && grpc.is_empty() && websocket.is_empty() {
|
||||
println!("No requests found in workspace {}", workspace_id);
|
||||
} else {
|
||||
for request in requests {
|
||||
println!("{} - {} {}", request.id, request.method, request.name);
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
// HTTP lines keep their `<id> - <method> <name>` shape; the other types put their type
|
||||
// where the method would be, since they have no method of that kind.
|
||||
for request in http {
|
||||
println!("{} - {} {}", request.id, request.method, request.name);
|
||||
}
|
||||
for request in grpc {
|
||||
println!("{} - gRPC {}", request.id, request.name);
|
||||
}
|
||||
for request in websocket {
|
||||
println!("{} - WebSocket {}", request.id, request.name);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -112,20 +142,65 @@ async fn schema(ctx: &CliContext, request_type: RequestSchemaType, pretty: bool)
|
||||
/// HttpRequest with its unknown fields dropped, quietly producing a broken request
|
||||
/// (the gRPC method name lands in `method`, and `service` disappears).
|
||||
fn reject_non_http_payload(payload: &Value, context: &str) -> CommandResult {
|
||||
let Some(object) = payload.as_object() else {
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
if let Some(model) = object.get("model").and_then(Value::as_str)
|
||||
if let Some(model) = payload.get("model").and_then(Value::as_str)
|
||||
&& !model.is_empty()
|
||||
&& model != "http_request"
|
||||
{
|
||||
return Err(format!(
|
||||
"{context} only supports HTTP requests, but the payload has \"model\": \"{model}\". The CLI cannot work with {model} requests yet; use the Yaak app instead."
|
||||
"{context} only supports HTTP requests, but the payload has \"model\": \"{model}\". The CLI cannot create {model} requests yet; use the Yaak app instead."
|
||||
));
|
||||
}
|
||||
|
||||
let known: std::collections::BTreeSet<String> = serde_json::to_value(schema_for!(HttpRequest))
|
||||
reject_unknown_fields::<HttpRequest>(payload, context, "an HTTP request", "http").map_err(
|
||||
|error| {
|
||||
format!("{error} If this is a gRPC or WebSocket request, the CLI cannot create it yet.")
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
/// An update is a merge patch onto the stored request, so it must use that request's
|
||||
/// model and fields. Anything else would be dropped silently on save.
|
||||
fn reject_mismatched_update(payload: &Value, request: &AnyRequest) -> CommandResult {
|
||||
let (model, noun, schema_type) = match request {
|
||||
AnyRequest::HttpRequest(_) => ("http_request", "an HTTP request", "http"),
|
||||
AnyRequest::GrpcRequest(_) => ("grpc_request", "a gRPC request", "grpc"),
|
||||
AnyRequest::WebsocketRequest(_) => {
|
||||
("websocket_request", "a WebSocket request", "websocket")
|
||||
}
|
||||
};
|
||||
let context = "request update";
|
||||
if let Some(given) = payload.get("model").and_then(Value::as_str)
|
||||
&& !given.is_empty()
|
||||
&& given != model
|
||||
{
|
||||
return Err(format!(
|
||||
"{context} cannot change a request's type: the stored request is {noun} (\"{model}\"), but the payload has \"model\": \"{given}\"."
|
||||
));
|
||||
}
|
||||
match request {
|
||||
AnyRequest::HttpRequest(_) => {
|
||||
reject_unknown_fields::<HttpRequest>(payload, context, noun, schema_type)
|
||||
}
|
||||
AnyRequest::GrpcRequest(_) => {
|
||||
reject_unknown_fields::<GrpcRequest>(payload, context, noun, schema_type)
|
||||
}
|
||||
AnyRequest::WebsocketRequest(_) => {
|
||||
reject_unknown_fields::<WebsocketRequest>(payload, context, noun, schema_type)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn reject_unknown_fields<T: schemars::JsonSchema>(
|
||||
payload: &Value,
|
||||
context: &str,
|
||||
noun: &str,
|
||||
schema_type: &str,
|
||||
) -> CommandResult {
|
||||
let Some(object) = payload.as_object() else {
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
let known: std::collections::BTreeSet<String> = serde_json::to_value(schema_for!(T))
|
||||
.ok()
|
||||
.and_then(|schema| schema.get("properties").and_then(Value::as_object).cloned())
|
||||
.map(|properties| properties.keys().cloned().collect())
|
||||
@@ -140,7 +215,7 @@ fn reject_non_http_payload(payload: &Value, context: &str) -> CommandResult {
|
||||
|
||||
if !unknown.is_empty() {
|
||||
return Err(format!(
|
||||
"{context} got fields that are not part of an HTTP request: {}. If this is a gRPC or WebSocket request, the CLI cannot work with it yet. Run `yaak request schema http` for the valid fields.",
|
||||
"{context} got fields that are not part of {noun}: {}. Run `yaak request schema {schema_type}` for the valid fields.",
|
||||
unknown.join(", ")
|
||||
));
|
||||
}
|
||||
@@ -454,43 +529,67 @@ fn create(
|
||||
fn update(ctx: &CliContext, json: Option<String>, json_input: Option<String>) -> CommandResult {
|
||||
let patch = parse_required_json(json, json_input, "request update")?;
|
||||
let id = require_id(&patch, "request update")?;
|
||||
reject_non_http_payload(&patch, "request update")?;
|
||||
|
||||
let existing = ctx
|
||||
.db()
|
||||
.get_http_request(&id)
|
||||
.map_err(|e| format!("Failed to get request for update: {e}"))?;
|
||||
let updated = apply_merge_patch(&existing, &patch, &id, "request update")?;
|
||||
let existing =
|
||||
get_any_request(ctx, &id).map_err(|e| format!("Failed to get request for update: {e}"))?;
|
||||
reject_mismatched_update(&patch, &existing)?;
|
||||
|
||||
let saved = ctx
|
||||
.query_manager()
|
||||
.with_tx(|tx| tx.upsert_http_request(&updated, &UpdateSource::Sync))
|
||||
.map_err(|e| format!("Failed to update request: {e}"))?;
|
||||
let context = "request update";
|
||||
let source = &UpdateSource::Sync;
|
||||
let saved_id = match existing {
|
||||
AnyRequest::HttpRequest(existing) => {
|
||||
let updated = apply_merge_patch(&existing, &patch, &id, context)?;
|
||||
ctx.query_manager().with_tx(|tx| tx.upsert_http_request(&updated, source)).map(|r| r.id)
|
||||
}
|
||||
AnyRequest::GrpcRequest(existing) => {
|
||||
let updated = apply_merge_patch(&existing, &patch, &id, context)?;
|
||||
ctx.query_manager().with_tx(|tx| tx.upsert_grpc_request(&updated, source)).map(|r| r.id)
|
||||
}
|
||||
AnyRequest::WebsocketRequest(existing) => {
|
||||
let updated = apply_merge_patch(&existing, &patch, &id, context)?;
|
||||
ctx.query_manager()
|
||||
.with_tx(|tx| tx.upsert_websocket_request(&updated, source))
|
||||
.map(|r| r.id)
|
||||
}
|
||||
}
|
||||
.map_err(|e| format!("Failed to update request: {e}"))?;
|
||||
|
||||
println!("Updated request: {}", saved.id);
|
||||
println!("Updated request: {saved_id}");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn show(ctx: &CliContext, request_id: &str) -> CommandResult {
|
||||
let request =
|
||||
ctx.db().get_http_request(request_id).map_err(|e| format!("Failed to get request: {e}"))?;
|
||||
let output = serde_json::to_string_pretty(&request)
|
||||
.map_err(|e| format!("Failed to serialize request: {e}"))?;
|
||||
let request = get_any_request(ctx, request_id)?;
|
||||
let output = match &request {
|
||||
AnyRequest::HttpRequest(r) => serde_json::to_string_pretty(r),
|
||||
AnyRequest::GrpcRequest(r) => serde_json::to_string_pretty(r),
|
||||
AnyRequest::WebsocketRequest(r) => serde_json::to_string_pretty(r),
|
||||
}
|
||||
.map_err(|e| format!("Failed to serialize request: {e}"))?;
|
||||
println!("{output}");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn delete(ctx: &CliContext, request_id: &str, yes: bool) -> CommandResult {
|
||||
// Resolved before asking, so an unknown ID fails without a prompt
|
||||
let request = get_any_request(ctx, request_id)?;
|
||||
if !yes && !confirm_delete("request", request_id) {
|
||||
println!("Aborted");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let deleted = ctx
|
||||
let source = &UpdateSource::Sync;
|
||||
let deleted_id = ctx
|
||||
.query_manager()
|
||||
.with_tx(|tx| tx.delete_http_request_by_id(request_id, &UpdateSource::Sync))
|
||||
.with_tx(|tx| match &request {
|
||||
AnyRequest::HttpRequest(r) => tx.delete_http_request_by_id(&r.id, source).map(|r| r.id),
|
||||
AnyRequest::GrpcRequest(r) => tx.delete_grpc_request_by_id(&r.id, source).map(|r| r.id),
|
||||
AnyRequest::WebsocketRequest(r) => {
|
||||
tx.delete_websocket_request_by_id(&r.id, source).map(|r| r.id)
|
||||
}
|
||||
})
|
||||
.map_err(|e| format!("Failed to delete request: {e}"))?;
|
||||
println!("Deleted request: {}", deleted.id);
|
||||
println!("Deleted request: {deleted_id}");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -503,7 +602,7 @@ pub async fn send_request_by_id(
|
||||
verbose: bool,
|
||||
) -> Result<(), String> {
|
||||
let request =
|
||||
ctx.db().get_any_request(request_id).map_err(|e| format!("Failed to get request: {e}"))?;
|
||||
get_any_request(ctx, request_id).map_err(|e| format!("Failed to get request: {e}"))?;
|
||||
match request {
|
||||
AnyRequest::HttpRequest(http_request) => {
|
||||
send_http_request_by_id(
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
use crate::cli::{ResponseArgs, ResponseCommands};
|
||||
use crate::commands::request::get_any_request;
|
||||
use crate::context::CliContext;
|
||||
use crate::utils::confirm::confirm_delete;
|
||||
use crate::utils::workspace::resolve_workspace_id;
|
||||
use std::io::Write;
|
||||
use yaak_models::models::HttpResponse;
|
||||
use yaak_models::models::{
|
||||
GrpcConnection, GrpcEventType, HttpResponse, WebsocketConnection, WebsocketEventType,
|
||||
};
|
||||
use yaak_models::queries::any_request::AnyRequest;
|
||||
use yaak_models::util::UpdateSource;
|
||||
|
||||
type CommandResult<T = ()> = std::result::Result<T, String>;
|
||||
@@ -25,37 +29,124 @@ pub fn run(ctx: &CliContext, args: ResponseArgs) -> i32 {
|
||||
}
|
||||
}
|
||||
|
||||
/// Accepts a response ID, or a request ID meaning "the latest response for that request".
|
||||
fn resolve_response(ctx: &CliContext, id: &str) -> CommandResult<HttpResponse> {
|
||||
if let Ok(response) = ctx.db().get_http_response(id) {
|
||||
return Ok(response);
|
||||
/// What a request leaves behind when sent: an HTTP response, or the connection a gRPC or
|
||||
/// WebSocket request opened, whose messages are its events.
|
||||
enum StoredResponse {
|
||||
Http(HttpResponse),
|
||||
Grpc(GrpcConnection),
|
||||
Websocket(WebsocketConnection),
|
||||
}
|
||||
|
||||
impl StoredResponse {
|
||||
fn id(&self) -> &str {
|
||||
match self {
|
||||
StoredResponse::Http(r) => &r.id,
|
||||
StoredResponse::Grpc(c) => &c.id,
|
||||
StoredResponse::Websocket(c) => &c.id,
|
||||
}
|
||||
}
|
||||
|
||||
if ctx.db().get_http_request(id).is_ok() {
|
||||
return ctx
|
||||
.db()
|
||||
.list_http_responses_for_request(id, Some(1))
|
||||
.map_err(|e| format!("Failed to list responses: {e}"))?
|
||||
fn created_at(&self) -> impl Ord + use<> {
|
||||
match self {
|
||||
StoredResponse::Http(r) => r.created_at,
|
||||
StoredResponse::Grpc(c) => c.created_at,
|
||||
StoredResponse::Websocket(c) => c.created_at,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Accepts a response or connection ID, or a request ID meaning "the latest response for that
|
||||
/// request".
|
||||
fn resolve_response(ctx: &CliContext, id: &str) -> CommandResult<StoredResponse> {
|
||||
let db = ctx.db();
|
||||
if let Ok(response) = db.get_http_response(id) {
|
||||
return Ok(StoredResponse::Http(response));
|
||||
}
|
||||
if let Ok(connection) = db.get_grpc_connection(id) {
|
||||
return Ok(StoredResponse::Grpc(connection));
|
||||
}
|
||||
if let Ok(connection) = db.get_websocket_connection(id) {
|
||||
return Ok(StoredResponse::Websocket(connection));
|
||||
}
|
||||
|
||||
let Ok(request) = get_any_request(ctx, id) else {
|
||||
return Err(format!("Could not resolve ID '{id}' as a response or request"));
|
||||
};
|
||||
let latest = list_for_request(ctx, &request, Some(1))?.into_iter().next();
|
||||
latest.ok_or_else(|| format!("Request {id} has no stored responses yet"))
|
||||
}
|
||||
|
||||
fn list_for_request(
|
||||
ctx: &CliContext,
|
||||
request: &AnyRequest,
|
||||
limit: Option<u64>,
|
||||
) -> CommandResult<Vec<StoredResponse>> {
|
||||
let db = ctx.db();
|
||||
let error = |e: yaak_models::error::Error| format!("Failed to list responses: {e}");
|
||||
Ok(match request {
|
||||
AnyRequest::HttpRequest(r) => db
|
||||
.list_http_responses_for_request(&r.id, limit)
|
||||
.map_err(error)?
|
||||
.into_iter()
|
||||
.next()
|
||||
.ok_or_else(|| format!("Request {id} has no stored responses yet"));
|
||||
}
|
||||
.map(StoredResponse::Http)
|
||||
.collect(),
|
||||
AnyRequest::GrpcRequest(r) => db
|
||||
.list_grpc_connections_for_request(&r.id, limit)
|
||||
.map_err(error)?
|
||||
.into_iter()
|
||||
.map(StoredResponse::Grpc)
|
||||
.collect(),
|
||||
AnyRequest::WebsocketRequest(r) => {
|
||||
let connections = db.list_websocket_connections_for_request(&r.id).map_err(error)?;
|
||||
let limit = limit.map_or(usize::MAX, |n| usize::try_from(n).unwrap_or(usize::MAX));
|
||||
connections.into_iter().take(limit).map(StoredResponse::Websocket).collect()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
Err(format!("Could not resolve ID '{id}' as a response or request"))
|
||||
/// Every kind of stored response in a workspace, newest first, with `limit` applied to the
|
||||
/// merged history rather than to each kind.
|
||||
fn list_for_workspace(
|
||||
ctx: &CliContext,
|
||||
workspace_id: &str,
|
||||
limit: Option<u64>,
|
||||
) -> CommandResult<Vec<StoredResponse>> {
|
||||
let db = ctx.db();
|
||||
let error = |e: yaak_models::error::Error| format!("Failed to list responses: {e}");
|
||||
// Each kind comes back newest first, so the newest `limit` overall are among the newest
|
||||
// `limit` of each kind.
|
||||
let mut responses = db
|
||||
.list_http_responses(workspace_id, limit)
|
||||
.map_err(error)?
|
||||
.into_iter()
|
||||
.map(StoredResponse::Http)
|
||||
.chain(
|
||||
db.list_grpc_connections(workspace_id)
|
||||
.map_err(error)?
|
||||
.into_iter()
|
||||
.map(StoredResponse::Grpc),
|
||||
)
|
||||
.chain(
|
||||
db.list_websocket_connections(workspace_id)
|
||||
.map_err(error)?
|
||||
.into_iter()
|
||||
.map(StoredResponse::Websocket),
|
||||
)
|
||||
.collect::<Vec<_>>();
|
||||
responses.sort_by_key(|r| std::cmp::Reverse(r.created_at()));
|
||||
if let Some(limit) = limit {
|
||||
responses.truncate(usize::try_from(limit).unwrap_or(usize::MAX));
|
||||
}
|
||||
Ok(responses)
|
||||
}
|
||||
|
||||
fn list(ctx: &CliContext, id: Option<&str>, limit: Option<u64>) -> CommandResult {
|
||||
// An explicit request ID scopes to that request; anything else lists the workspace.
|
||||
let responses = match id {
|
||||
Some(id) if ctx.db().get_http_request(id).is_ok() => ctx
|
||||
.db()
|
||||
.list_http_responses_for_request(id, limit)
|
||||
.map_err(|e| format!("Failed to list responses: {e}"))?,
|
||||
other => {
|
||||
let workspace_id = resolve_workspace_id(ctx, other, "response list")?;
|
||||
ctx.db()
|
||||
.list_http_responses(&workspace_id, limit)
|
||||
.map_err(|e| format!("Failed to list responses: {e}"))?
|
||||
let responses = match id.map(|id| get_any_request(ctx, id)) {
|
||||
Some(Ok(request)) => list_for_request(ctx, &request, limit)?,
|
||||
_ => {
|
||||
let workspace_id = resolve_workspace_id(ctx, id, "response list")?;
|
||||
list_for_workspace(ctx, &workspace_id, limit)?
|
||||
}
|
||||
};
|
||||
|
||||
@@ -65,7 +156,20 @@ fn list(ctx: &CliContext, id: Option<&str>, limit: Option<u64>) -> CommandResult
|
||||
}
|
||||
|
||||
for response in responses {
|
||||
let status = match response.error {
|
||||
let (error, status, url, elapsed) = match &response {
|
||||
StoredResponse::Http(r) => {
|
||||
let reason = r.status_reason.clone().unwrap_or_default();
|
||||
let status = format!("{} {}", r.status, reason).trim_end().to_string();
|
||||
(r.error.as_deref(), status, r.url.as_str(), r.elapsed)
|
||||
}
|
||||
StoredResponse::Grpc(c) => {
|
||||
(c.error.as_deref(), format!("gRPC status {}", c.status), c.url.as_str(), c.elapsed)
|
||||
}
|
||||
StoredResponse::Websocket(c) => {
|
||||
(c.error.as_deref(), c.status.to_string(), c.url.as_str(), c.elapsed)
|
||||
}
|
||||
};
|
||||
let status = match error {
|
||||
// Errors can be a paragraph of nested debug output; `show` has the full text.
|
||||
Some(error) => {
|
||||
let first_line = error.lines().next().unwrap_or_default();
|
||||
@@ -75,34 +179,70 @@ fn list(ctx: &CliContext, id: Option<&str>, limit: Option<u64>) -> CommandResult
|
||||
};
|
||||
format!("ERROR {summary}")
|
||||
}
|
||||
None => {
|
||||
let reason = response.status_reason.unwrap_or_default();
|
||||
format!("{} {}", response.status, reason).trim_end().to_string()
|
||||
}
|
||||
None => status,
|
||||
};
|
||||
println!("{} - {} {} ({}ms)", response.id, status, response.url, response.elapsed);
|
||||
println!("{} - {} {} ({}ms)", response.id(), status, url, elapsed);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn show(ctx: &CliContext, id: &str) -> CommandResult {
|
||||
let response = resolve_response(ctx, id)?;
|
||||
let output = serde_json::to_string_pretty(&response)
|
||||
.map_err(|e| format!("Failed to serialize response: {e}"))?;
|
||||
let output = match resolve_response(ctx, id)? {
|
||||
StoredResponse::Http(r) => serde_json::to_string_pretty(&r),
|
||||
StoredResponse::Grpc(c) => serde_json::to_string_pretty(&c),
|
||||
StoredResponse::Websocket(c) => serde_json::to_string_pretty(&c),
|
||||
}
|
||||
.map_err(|e| format!("Failed to serialize response: {e}"))?;
|
||||
println!("{output}");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// An HTTP body as stored; for a gRPC or WebSocket connection, each message the server sent,
|
||||
/// oldest first, one per line.
|
||||
fn body(ctx: &CliContext, id: &str) -> CommandResult {
|
||||
let response = resolve_response(ctx, id)?;
|
||||
let Some(body_path) = response.body_path else {
|
||||
return Err(format!("Response {} has no stored body", response.id));
|
||||
let bytes = match resolve_response(ctx, id)? {
|
||||
StoredResponse::Http(response) => {
|
||||
let Some(body_path) = response.body_path else {
|
||||
return Err(format!("Response {} has no stored body", response.id));
|
||||
};
|
||||
std::fs::read(&body_path)
|
||||
.map_err(|e| format!("Failed to read response body at {body_path}: {e}"))?
|
||||
}
|
||||
StoredResponse::Grpc(connection) => {
|
||||
let mut events = ctx
|
||||
.db()
|
||||
.list_grpc_events(&connection.id)
|
||||
.map_err(|e| format!("Failed to list connection events: {e}"))?;
|
||||
events.sort_by_key(|e| e.created_at);
|
||||
let mut bytes = Vec::new();
|
||||
for event in events.iter().filter(|e| e.event_type == GrpcEventType::ServerMessage) {
|
||||
bytes.extend_from_slice(event.content.as_bytes());
|
||||
bytes.push(b'\n');
|
||||
}
|
||||
bytes
|
||||
}
|
||||
StoredResponse::Websocket(connection) => {
|
||||
let mut events = ctx
|
||||
.db()
|
||||
.list_websocket_events(&connection.id)
|
||||
.map_err(|e| format!("Failed to list connection events: {e}"))?;
|
||||
events.sort_by_key(|e| e.created_at);
|
||||
let mut bytes = Vec::new();
|
||||
for event in events.iter().filter(|e| {
|
||||
e.is_server
|
||||
&& matches!(
|
||||
e.message_type,
|
||||
WebsocketEventType::Text | WebsocketEventType::Binary
|
||||
)
|
||||
}) {
|
||||
bytes.extend_from_slice(&event.message);
|
||||
bytes.push(b'\n');
|
||||
}
|
||||
bytes
|
||||
}
|
||||
};
|
||||
|
||||
let bytes = std::fs::read(&body_path)
|
||||
.map_err(|e| format!("Failed to read response body at {body_path}: {e}"))?;
|
||||
|
||||
let mut stdout = std::io::stdout();
|
||||
stdout.write_all(&bytes).map_err(|e| format!("Failed to write response body: {e}"))?;
|
||||
stdout.flush().map_err(|e| format!("Failed to flush response body: {e}"))?;
|
||||
@@ -112,14 +252,29 @@ fn body(ctx: &CliContext, id: &str) -> CommandResult {
|
||||
fn delete(ctx: &CliContext, id: &str, yes: bool) -> CommandResult {
|
||||
// Deleting is scoped to a whole request or workspace, since a single stored
|
||||
// response is rarely the thing someone wants to remove.
|
||||
if ctx.db().get_http_request(id).is_ok() {
|
||||
if let Ok(request) = get_any_request(ctx, id) {
|
||||
if !yes && !confirm_delete("responses for request", id) {
|
||||
println!("Aborted");
|
||||
return Ok(());
|
||||
}
|
||||
let source = &UpdateSource::Sync;
|
||||
let count = ctx
|
||||
.query_manager()
|
||||
.with_tx(|tx| tx.delete_all_http_responses_for_request(id, &UpdateSource::Sync))
|
||||
.with_tx(|tx| match &request {
|
||||
AnyRequest::HttpRequest(r) => {
|
||||
tx.delete_all_http_responses_for_request(&r.id, source)
|
||||
}
|
||||
AnyRequest::GrpcRequest(r) => {
|
||||
let count = tx.list_grpc_connections_for_request(&r.id, None)?.len();
|
||||
tx.delete_all_grpc_connections_for_request(&r.id, source)?;
|
||||
Ok(count)
|
||||
}
|
||||
AnyRequest::WebsocketRequest(r) => {
|
||||
let count = tx.list_websocket_connections_for_request(&r.id)?.len();
|
||||
tx.delete_all_websocket_connections_for_request(&r.id, source)?;
|
||||
Ok(count)
|
||||
}
|
||||
})
|
||||
.map_err(|e| format!("Failed to delete responses: {e}"))?;
|
||||
println!("Deleted {count} responses for request {id}");
|
||||
return Ok(());
|
||||
@@ -133,7 +288,13 @@ fn delete(ctx: &CliContext, id: &str, yes: bool) -> CommandResult {
|
||||
let count = ctx
|
||||
.query_manager()
|
||||
.with_tx(|tx| {
|
||||
tx.delete_all_http_responses_for_workspace(&workspace_id, &UpdateSource::Sync)
|
||||
let source = &UpdateSource::Sync;
|
||||
let grpc = tx.list_grpc_connections(&workspace_id)?.len();
|
||||
let websocket = tx.list_websocket_connections(&workspace_id)?.len();
|
||||
let http = tx.delete_all_http_responses_for_workspace(&workspace_id, source)?;
|
||||
tx.delete_all_grpc_connections_for_workspace(&workspace_id, source)?;
|
||||
tx.delete_all_websocket_connections_for_workspace(&workspace_id, source)?;
|
||||
Ok::<_, yaak_models::error::Error>(http + grpc + websocket)
|
||||
})
|
||||
.map_err(|e| format!("Failed to delete responses: {e}"))?;
|
||||
println!("Deleted {count} responses for workspace {workspace_id}");
|
||||
|
||||
@@ -179,6 +179,22 @@ async fn send_many(
|
||||
let mut success_count = 0usize;
|
||||
let mut failures: Vec<(String, String)> = Vec::new();
|
||||
|
||||
// The CLI sends HTTP only, so the rest of a folder is reported, not counted as failing.
|
||||
let mut skipped: Vec<String> = Vec::new();
|
||||
let mut request_ids = request_ids;
|
||||
request_ids.retain(|id| match request::get_any_request(ctx, id) {
|
||||
Ok(AnyRequest::GrpcRequest(r)) => {
|
||||
skipped.push(format!("{id} - gRPC {}", r.name));
|
||||
false
|
||||
}
|
||||
Ok(AnyRequest::WebsocketRequest(r)) => {
|
||||
skipped.push(format!("{id} - WebSocket {}", r.name));
|
||||
false
|
||||
}
|
||||
// A request that has gone missing since it was listed fails when sent, as before
|
||||
Ok(AnyRequest::HttpRequest(_)) | Err(_) => true,
|
||||
});
|
||||
|
||||
match mode {
|
||||
ExecutionMode::Sequential => {
|
||||
for request_id in request_ids {
|
||||
@@ -229,7 +245,18 @@ async fn send_many(
|
||||
}
|
||||
|
||||
let failure_count = failures.len();
|
||||
println!("Send summary: {success_count} succeeded, {failure_count} failed");
|
||||
if skipped.is_empty() {
|
||||
println!("Send summary: {success_count} succeeded, {failure_count} failed");
|
||||
} else {
|
||||
println!(
|
||||
"Send summary: {success_count} succeeded, {failure_count} failed, {} skipped",
|
||||
skipped.len()
|
||||
);
|
||||
println!("Skipped, since the CLI can only send HTTP requests so far:");
|
||||
for line in &skipped {
|
||||
println!(" {line}");
|
||||
}
|
||||
}
|
||||
|
||||
if failure_count == 0 {
|
||||
return Ok(());
|
||||
|
||||
Reference in new issue
Block a user