diff --git a/docs/sdk-python.md b/docs/sdk-python.md index da6fcef9..35f42bda 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -367,6 +367,12 @@ path on the no-network lane. ```python matches = sb.find("/work", r"TODO|FIXME", glob="*.py") +from contextlib import closing + +with closing(sb.find_iter("/work", r"TODO")) as stream: # streamed; closing ends the walk in the guest + for m in stream: + if m["line"] > 100: + break # [{"file", "line", "content"}]; glob is anchored *? wildcards on the file name results = sb.replace(["/work/main.py"], r"foo", "bar") @@ -394,7 +400,7 @@ when the relay drops, which `w.error` tells apart from a clean close (`None`). ```python sb.git_clone(url, "/work/repo", branch="main", depth=1, auth=token) # egress lane only -st = sb.git_status("/work/repo") # {"branch", "ahead", "behind", "files"} +st = sb.git_status("/work/repo") # {"branch", "ahead", "behind", "files", "truncated"?} sb.git_add("/work/repo", ["a.txt"]) sha = sb.git_commit("/work/repo", "message", "Dev ") sb.git_push("/work/repo", auth=token) # egress lane only diff --git a/docs/sdk.md b/docs/sdk.md index 37629df0..339048b9 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -441,6 +441,14 @@ err = sb.Pull(ctx, "/work", tarWriter) // stream /work back as a tar ```go matches, err := sb.Find(ctx, "/work", `TODO|FIXME`, "*.go") // []wire.Match{File, Line, Content}; glob is anchored *? wildcards on the file name +for m, err := range sb.FindSeq(ctx, "/work", `TODO`, "") { // streamed; a break ends the walk in the guest + if err != nil { + return err + } + if m.Line > 100 { + break + } +} results, err := sb.Replace(ctx, []string{"/work/main.go"}, `foo`, "bar") // []wire.Replaced{File, Replacements}; per-file atomic @@ -468,7 +476,7 @@ overflow instead of the stream silently dropping events. ```go err = sb.GitClone(ctx, url, "/work/repo", "main", 0, token) // egress lane only; depth > 0 = shallow -st, err := sb.GitStatus(ctx, "/work/repo") // Branch, Ahead, Behind, Files[] +st, err := sb.GitStatus(ctx, "/work/repo") // Branch, Ahead, Behind, Files[]; Truncated when the list was cut at ~1 MiB err = sb.GitAdd(ctx, "/work/repo", "a.txt") hash, err := sb.GitCommit(ctx, "/work/repo", "message", "Dev ") err = sb.GitPush(ctx, "/work/repo", token) // egress lane only diff --git a/docs/silkd.md b/docs/silkd.md index 4f096a58..20e2d0e2 100644 --- a/docs/silkd.md +++ b/docs/silkd.md @@ -38,7 +38,7 @@ a frame only one side can parse fails CI. | search | `fs_find {path, pattern, glob?}` → `match{file, line, content}`… → `done` / `fs_replace {files, pattern, replacement}` → `replaced{file, replacements}`… → `done` | regex as data, no shell quoting; `glob` is anchored `*`/`?` wildcards over file names; find skips binary and >8 MiB files, and replace skips the same 8 MiB bound with a zero count | | watch | `fs_watch {path, recursive?}` | `ready` once armed (events after it are guaranteed captured), then `event{kind, path}` until the client disconnects; watcher or delivery-queue overflow errors arrive as a terminal `error` instead of silently losing events | | pty | `pty_open {cols, rows, cwd?, env?, user?}` / `pty_resize {pid, cols, rows}` | a shell under a pseudo-terminal; output as `stdout` frames, input as `stdin` frames, exit terminal. PTYs register in the proc table like any exec | -| git | `git_clone {url, path, branch?, depth?, auth?}` / `git_status {path}` / `git_add {path, files}` / `git_commit {path, message, author}` / `git_push {path, auth?}` / `git_pull {path, auth?}` / `git_branch {path, action, name?}` | structured results (porcelain-v2 status, commit hash, branch list). `auth` is injected as an in-memory header, never written to guest disk | +| git | `git_clone {url, path, branch?, depth?, auth?}` / `git_status {path}` / `git_add {path, files}` / `git_commit {path, message, author}` / `git_push {path, auth?}` / `git_pull {path, auth?}` / `git_branch {path, action, name?}` | structured results (porcelain-v2 status, commit hash, branch list). A status past about 1 MiB of entries carries `truncated: true` with the head of the list, so one frame never exceeds the cap. `auth` is injected as an in-memory header, never written to guest disk | | port | `port_forward {port}` | relays guest TCP 127.0.0.1:port over this connection: `ready` once connected, then `data` both ways (`data_end` half-closes the guest socket); the server closing ends the stream with `done`. Works on both lanes — the no-network lane's only way in | | lsp | `lsp_start {language, root?}` / `lsp_request {server_id}` / `lsp_stop {server_id}` | a broker for the language server a flavor image ships: `lsp_start` spawns the argv named in `/etc/silkd/lsp.d/` (absent on the base image → `not_found` naming the flavor) and answers `lsp_started{server_id}`; `lsp_request` attaches the JSON-RPC byte stream (`ready`, then `data` both ways — silkd pipes bytes, it never parses LSP; `data_end` half-closes the server's stdin) and the stream ending reaps the server (v1 single-shot); `lsp_stop` kills it early, and a server nothing attaches within 5 minutes is reaped | | misc | `info` | `{version, proto, uptime_secs, procs, sessions}` — the in-guest readiness probe (sandboxd consumes it; no SDK surface). Distinct from the control plane's `GET /v1/info`, which reports node pools and claims | diff --git a/protocol/wire/fixtures/v1/resp_git_status_result_truncated.json b/protocol/wire/fixtures/v1/resp_git_status_result_truncated.json new file mode 100644 index 00000000..be64dcb7 --- /dev/null +++ b/protocol/wire/fixtures/v1/resp_git_status_result_truncated.json @@ -0,0 +1 @@ +{"type":"git_status_result","branch":"main","ahead":0,"behind":0,"files":[{"path":"a.txt","staged":"?","unstaged":"?"}],"truncated":true} diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index 24dcd9e7..499809f2 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -594,10 +594,11 @@ func (Event) RespType() string { return "event" } // GitStatusResult answers GitStatus. type GitStatusResult struct { - Branch string `json:"branch"` - Ahead uint32 `json:"ahead"` - Behind uint32 `json:"behind"` - Files []GitFileStatus `json:"files"` + Branch string `json:"branch"` + Ahead uint32 `json:"ahead"` + Behind uint32 `json:"behind"` + Files []GitFileStatus `json:"files"` + Truncated bool `json:"truncated,omitempty"` } func (GitStatusResult) RespType() string { return "git_status_result" } diff --git a/protocol/wire/frame_test.go b/protocol/wire/frame_test.go index 50eefd1e..4efef372 100644 --- a/protocol/wire/frame_test.go +++ b/protocol/wire/frame_test.go @@ -58,8 +58,8 @@ func TestFixtureCorpusRoundTrips(t *testing.T) { seen++ } - if seen != 60 { - t.Fatalf("fixture corpus: %d frames, want exactly 60", seen) + if seen != 61 { + t.Fatalf("fixture corpus: %d frames, want exactly 61", seen) } } diff --git a/sdk/go/find.go b/sdk/go/find.go index 9fd2c4bf..69b0c5b3 100644 --- a/sdk/go/find.go +++ b/sdk/go/find.go @@ -2,6 +2,7 @@ package sandbox import ( "context" + "iter" "github.com/cocoonstack/sandbox/protocol/wire" ) @@ -13,6 +14,12 @@ func (s *Sandbox) Find(ctx context.Context, path, pattern, glob string) ([]wire. return collectRPC[wire.Match](ctx, s, &wire.FsFind{Path: path, Pattern: pattern, Glob: glob}) } +// FindSeq streams the matches Find would collect; breaking out of the loop +// ends the walk in the guest, so the caller bounds the result. +func (s *Sandbox) FindSeq(ctx context.Context, path, pattern, glob string) iter.Seq2[wire.Match, error] { + return streamRPC[wire.Match](ctx, s, &wire.FsFind{Path: path, Pattern: pattern, Glob: glob}) +} + // Replace rewrites pattern (a regular expression) to replacement in each of // files, returning one result per file with its replacement count. func (s *Sandbox) Replace(ctx context.Context, files []string, pattern, replacement string) ([]wire.Replaced, error) { diff --git a/sdk/go/find_test.go b/sdk/go/find_test.go index 2b5861a0..bf681b5f 100644 --- a/sdk/go/find_test.go +++ b/sdk/go/find_test.go @@ -33,6 +33,37 @@ func TestFindStreamsMatches(t *testing.T) { } } +func TestFindSeqStopsAfterTheCallerBreaks(t *testing.T) { + sb := fakeSandbox(t) + ctx := t.Context() + if err := sb.Mkdir(ctx, "/work", true); err != nil { + t.Fatalf("mkdir: %v", err) + } + if err := sb.WriteFile(ctx, "/work/many.txt", []byte(strings.Repeat("TODO\n", 500)), nil); err != nil { + t.Fatalf("write: %v", err) + } + + var seen []wire.Match + for m, err := range sb.FindSeq(ctx, "/work", "TODO", "") { + if err != nil { + t.Fatalf("FindSeq: %v", err) + } + if seen = append(seen, m); len(seen) == 3 { + break + } + } + if len(seen) != 3 || seen[2].Line != 3 { + t.Fatalf("seen %+v, want the first three lines", seen) + } + all, err := sb.Find(ctx, "/work", "TODO", "") + if err != nil { + t.Fatalf("Find after an abandoned stream: %v", err) + } + if len(all) != 500 { + t.Errorf("Find returned %d matches, want 500", len(all)) + } +} + func TestFindBadPatternIsTypedError(t *testing.T) { sb := fakeSandbox(t) _, err := sb.Find(t.Context(), "/", "(", "") diff --git a/sdk/go/utils.go b/sdk/go/utils.go index c334e78d..00b943d0 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "io" + "iter" "github.com/cocoonstack/sandbox/protocol/wire" "github.com/cocoonstack/sandbox/sdk/go/silkd" @@ -63,28 +64,46 @@ func oneShotRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire. // collectRPC sends req and gathers every streamed frame of type T until Done. func collectRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire.Request) ([]T, error) { - conn, done, err := s.call(ctx, req) - if err != nil { - return nil, err - } - defer done() var out []T - for { - resp, err := recv(ctx, conn) + for v, err := range streamRPC[T, PT](ctx, s, req) { if err != nil { return nil, err } - if v, ok := resp.(PT); ok { - out = append(out, *v) - continue + out = append(out, v) + } + return out, nil +} + +// streamRPC sends req and yields each streamed frame of type T until Done; breaking out closes the connection, which ends the guest-side producer. +func streamRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire.Request) iter.Seq2[T, error] { + return func(yield func(T, error) bool) { + var zero T + conn, done, err := s.call(ctx, req) + if err != nil { + yield(zero, err) + return } - switch r := resp.(type) { - case *wire.Done: - return out, nil - case *wire.ErrorResp: - return nil, r - default: - return nil, unexpected(resp) + defer done() + for { + resp, err := recv(ctx, conn) + if err != nil { + yield(zero, err) + return + } + if v, ok := resp.(PT); ok { + if !yield(*v, nil) { + return + } + continue + } + switch r := resp.(type) { + case *wire.Done: + case *wire.ErrorResp: + yield(zero, r) + default: + yield(zero, unexpected(resp)) + } + return } } } diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index d536dde4..99f7a472 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -157,9 +157,17 @@ def pull(self, path: str) -> bytes: return _drain_data(conn) def find(self, path: str, pattern: str, glob: str = "") -> list[dict]: + return list(self.find_iter(path, pattern, glob)) + + def find_iter(self, path: str, pattern: str, glob: str = "") -> Iterator[dict]: + """Yields matches as they stream. Closing the generator closes the + connection and ends the walk in the guest; wrap it in contextlib.closing + for deterministic cleanup, since only CPython finalizes on refcount.""" with self._dial() as conn: conn.send("fs_find", path=path, pattern=pattern, glob=glob or None) - return [f for f in conn.recv_until("done") if f["type"] == "match"] + for f in conn.recv_until("done"): + if f["type"] == "match": + yield f def replace(self, files: list[str], pattern: str, replacement: str) -> list[dict]: with self._dial() as conn: diff --git a/silkd/src/find.rs b/silkd/src/find.rs index 635b8f9d..d8cc679b 100644 --- a/silkd/src/find.rs +++ b/silkd/src/find.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use regex::Regex; use tokio::fs; -use tokio::io::AsyncWrite; +use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite}; use tokio::runtime::Handle; use tokio::sync::{Semaphore, SemaphorePermit, mpsc}; @@ -67,9 +67,10 @@ struct Walk<'a> { } impl Walk<'_> { - /// Walks `root` depth-first, sending a `match` frame per matching line. A - /// failure on the root propagates; deeper ones skip only that directory. - fn run(&self, root: PathBuf) -> std::io::Result<()> { + /// Walks `root` depth-first, sending a `match` frame per matching line; false + /// means the receiver went away. A failure on the root propagates; deeper + /// ones skip only that directory. + fn run(&self, root: PathBuf) -> std::io::Result { let mut stack = vec![root]; let mut root = true; while let Some(dir) = stack.pop() { @@ -79,6 +80,9 @@ impl Walk<'_> { Err(_) => continue, }; for ent in rd { + if self.tx.is_closed() { + return Ok(false); + } let ent = match ent { Ok(ent) => ent, Err(e) if root => return Err(e), @@ -91,12 +95,12 @@ impl Walk<'_> { if ft.is_dir() { stack.push(p); } else if ft.is_file() && name_matches(&p, self.name_re) && !self.scan_file(&p) { - return Ok(()); + return Ok(false); } } root = false; } - Ok(()) + Ok(true) } /// Scans one file, reporting whether the receiver is still listening. The size @@ -133,13 +137,18 @@ impl Walk<'_> { /// Streams `match` frames for every line under `path` matching `pattern`, /// terminated by `done`. `glob` narrows the walk to file names matching it /// (`*` and `?` wildcards); an invalid pattern is a bad-request error. -pub async fn find( +pub async fn find( + reader: &mut R, w: &mut W, path: String, pattern: String, glob: Option, -) -> std::io::Result<()> { - find_bounded(w, path, pattern, glob, MATCH_QUEUE_BYTES).await +) -> std::io::Result<()> +where + R: AsyncBufRead + Unpin, + W: AsyncWrite + Unpin, +{ + find_bounded(reader, w, path, pattern, glob, MATCH_QUEUE_BYTES).await } /// Rewrites every `pattern` match to `replacement` in each of `files`, @@ -191,13 +200,18 @@ pub async fn replace( proto::write_frame(w, &Response::Done).await } -async fn find_bounded( +async fn find_bounded( + reader: &mut R, w: &mut W, path: String, pattern: String, glob: Option, budget_bytes: usize, -) -> std::io::Result<()> { +) -> std::io::Result<()> +where + R: AsyncBufRead + Unpin, + W: AsyncWrite + Unpin, +{ let re = match Regex::new(&pattern) { Ok(re) => re, Err(e) => return proto::error_frame(w, ErrorKind::BadRequest, e.to_string()).await, @@ -223,18 +237,35 @@ async fn find_bounded( let mut batch = Vec::new(); let mut buf = Vec::new(); let mut failed = None; - while rx.recv_many(&mut batch, MATCH_QUEUE).await > 0 { - let sent = proto::write_frames(w, &mut buf, &batch).await; - budget.release(batch.iter().map(match_cost).sum()); - batch.clear(); - if let Err(e) = sent { - failed = Some(e); - break; + let mut gone = false; + loop { + tokio::select! { + biased; + // The client sends nothing during a find, so any readable state ends the walk. + _ = reader.fill_buf() => { + gone = true; + break; + } + n = rx.recv_many(&mut batch, MATCH_QUEUE) => { + if n == 0 { + break; + } + let sent = proto::write_frames(w, &mut buf, &batch).await; + budget.release(batch.iter().map(match_cost).sum()); + batch.clear(); + if let Err(e) = sent { + failed = Some(e); + break; + } + } } } drop(rx); budget.close(); let walked = walk.await.map_err(std::io::Error::other)?; + if gone { + return Ok(()); + } if let Some(e) = failed { if e.kind() == std::io::ErrorKind::InvalidData { return err_frame(w, &e, "find").await; @@ -242,7 +273,7 @@ async fn find_bounded( return Err(e); } match walked { - Ok(()) => proto::write_frame(w, &Response::Done).await, + Ok(_) => proto::write_frame(w, &Response::Done).await, Err(e) => err_frame(w, &e, "read_dir").await, } } @@ -278,7 +309,7 @@ fn glob_regex(glob: &str) -> Result { mod tests { use std::time::Duration; - use tokio::io::{AsyncReadExt, DuplexStream}; + use tokio::io::{AsyncReadExt, BufReader, DuplexStream}; use tokio::task::JoinHandle; use super::*; @@ -304,8 +335,22 @@ mod tests { (w, reader) } + fn silent_client() -> (DuplexStream, BufReader) { + let (peer, r) = tokio::io::duplex(64); + (peer, BufReader::new(r)) + } + async fn run(w: &mut DuplexStream, dir: &Path, budget: usize) -> std::io::Result<()> { - find_bounded(w, dir.display().to_string(), "needle".into(), None, budget).await + let (_peer, mut reader) = silent_client(); + find_bounded( + &mut reader, + w, + dir.display().to_string(), + "needle".into(), + None, + budget, + ) + .await } #[tokio::test(flavor = "multi_thread")] @@ -330,6 +375,49 @@ mod tests { assert!(found.is_err()); } + #[tokio::test(flavor = "multi_thread")] + async fn find_stops_without_done_when_the_client_disconnects() { + let dir = tree(2000, 1).await; + let (mut w, reader) = sink(); + let (peer, mut client) = silent_client(); + drop(peer); + let found = tokio::time::timeout( + Duration::from_secs(10), + find_bounded( + &mut client, + &mut w, + dir.path().display().to_string(), + "needle".into(), + None, + MATCH_QUEUE_BYTES, + ), + ) + .await + .expect("walk must end once the client is gone"); + found.expect("find"); + drop(w); + let out = reader.await.expect("join"); + assert!(!out.contains("\"type\":\"done\""), "{out}"); + } + + #[tokio::test(flavor = "multi_thread")] + async fn walk_stops_at_the_next_entry_once_the_receiver_is_gone() { + let dir = tree(50, 1).await; + let re = Regex::new("needle").expect("regex"); + let (tx, rx) = mpsc::channel::(MATCH_QUEUE); + drop(rx); + let budget = MatchBudget::new(MATCH_QUEUE_BYTES, Handle::current()); + let walk = Walk { + re: &re, + name_re: None, + tx: &tx, + budget: &budget, + }; + let completed = + tokio::task::block_in_place(|| walk.run(dir.path().to_path_buf())).expect("walk"); + assert!(!completed); + } + #[tokio::test(flavor = "multi_thread")] async fn find_reports_an_oversized_match_as_an_error_frame() { let dir = tempfile::tempdir().expect("tempdir"); diff --git a/silkd/src/git.rs b/silkd/src/git.rs index effd09a8..a0e32250 100644 --- a/silkd/src/git.rs +++ b/silkd/src/git.rs @@ -12,6 +12,10 @@ use tokio::process::Command; use crate::proto::{self, ErrorKind, GitBranchOp, GitFileStatus, Response}; use crate::sysutil; +/// Estimated JSON bytes of file entries one status frame carries before it truncates; an eighth of the frame cap leaves room for escaping. +const STATUS_FILES_BYTES: usize = proto::MAX_FRAME / 8; +const STATUS_ENTRY_OVERHEAD: usize = 40; + /// Clones `url` into `path` (optionally a branch, shallow depth), then Done. pub async fn clone( w: &mut W, @@ -220,12 +224,19 @@ fn parse_status(text: &str) -> Response { let mut branch = String::new(); let (mut ahead, mut behind) = (0, 0); let mut files = Vec::new(); + let mut bytes = 0; + let mut truncated = false; for line in text.lines() { if let Some(rest) = line.strip_prefix("# branch.head ") { branch = rest.to_string(); } else if let Some(rest) = line.strip_prefix("# branch.ab ") { (ahead, behind) = parse_ahead_behind(rest); } else if let Some(f) = parse_file_line(line) { + bytes += f.path.len() + STATUS_ENTRY_OVERHEAD; + if bytes > STATUS_FILES_BYTES { + truncated = true; + break; + } files.push(f); } } @@ -234,6 +245,7 @@ fn parse_status(text: &str) -> Response { ahead, behind, files, + truncated, } } @@ -309,6 +321,25 @@ fn split_author(author: &str) -> (&str, &str) { mod tests { use super::*; + #[test] + fn status_truncates_past_the_frame_budget() { + let text: String = (0..200_000).map(|i| format!("? f{i}\n")).collect(); + let Response::GitStatusResult { + files, truncated, .. + } = parse_status(&text) + else { + panic!("status frame") + }; + assert!(truncated); + assert!(!files.is_empty() && files.len() < 200_000); + assert!(serde_json::to_vec(&parse_status(&text)).unwrap().len() < proto::MAX_FRAME); + + let Response::GitStatusResult { truncated, .. } = parse_status("? one\n") else { + panic!("status frame") + }; + assert!(!truncated); + } + #[test] fn parses_ordinary_rename_untracked_and_conflict() { let ordinary = parse_file_line("1 .M N... 100644 100644 100644 h1 h2 src/main.rs").unwrap(); diff --git a/silkd/src/proto.rs b/silkd/src/proto.rs index f7218e9c..d462235d 100644 --- a/silkd/src/proto.rs +++ b/silkd/src/proto.rs @@ -279,6 +279,8 @@ pub enum Response { ahead: u32, behind: u32, files: Vec, + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + truncated: bool, }, GitCommitResult { hash: String, @@ -820,7 +822,7 @@ mod tests { seen += 1; } assert_eq!( - seen, 60, + seen, 61, "fixture corpus: adding a verb means adding its fixture" ); } diff --git a/silkd/src/server.rs b/silkd/src/server.rs index b37236f5..563160b0 100644 --- a/silkd/src/server.rs +++ b/silkd/src/server.rs @@ -136,7 +136,7 @@ impl State { path, pattern, glob, - } => find::find(&mut writer, path, pattern, glob).await, + } => find::find(&mut reader, &mut writer, path, pattern, glob).await, Request::FsReplace { files, pattern, diff --git a/silkd/tests/find_e2e.rs b/silkd/tests/find_e2e.rs index 57f581ee..5ddd4857 100644 --- a/silkd/tests/find_e2e.rs +++ b/silkd/tests/find_e2e.rs @@ -3,13 +3,35 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; +use std::sync::Arc; + +use serde_json::{Value, json}; +use silkd::server::State; +use tokio::io::AsyncWriteExt; + use common::{exchange, type_of}; -use serde_json::json; async fn write_file(dir: &std::path::Path, name: &str, body: &str) { tokio::fs::write(dir.join(name), body).await.unwrap(); } +async fn find_frames(req: Value) -> Vec { + let (mut cw, mut out, handle) = common::connect(&Arc::new(State::new())); + cw.write_all(format!("{req}\n").as_bytes()).await.unwrap(); + let mut frames = Vec::new(); + while let Some(l) = out.next_line().await.unwrap() { + let frame: Value = serde_json::from_str(&l).unwrap(); + let terminal = matches!(type_of(&frame), "done" | "error"); + frames.push(frame); + if terminal { + break; + } + } + drop(cw); + handle.await.unwrap().unwrap(); + frames +} + #[tokio::test] async fn find_streams_line_matches() { let dir = tempfile::tempdir().unwrap(); @@ -18,13 +40,12 @@ async fn find_streams_line_matches() { tokio::fs::create_dir(dir.path().join("sub")).await.unwrap(); write_file(&dir.path().join("sub"), "c.rs", "// TODO nested\n").await; - let frames = exchange(&[json!({ + let frames = find_frames(json!({ "op": "fs_find", "path": dir.path().to_str().unwrap(), "pattern": "TODO", "glob": "*.rs" - }) - .to_string()]) + })) .await; let matches: Vec<_> = frames.iter().filter(|f| type_of(f) == "match").collect(); @@ -45,13 +66,12 @@ async fn find_glob_is_a_real_glob_not_a_substring() { write_file(dir.path(), "a.rs", "TODO\n").await; write_file(dir.path(), "a.rs.bak", "TODO\n").await; - let frames = exchange(&[json!({ + let frames = find_frames(json!({ "op": "fs_find", "path": dir.path().to_str().unwrap(), "pattern": "TODO", "glob": "*.rs" - }) - .to_string()]) + })) .await; let files: Vec<_> = frames .iter() @@ -61,13 +81,12 @@ async fn find_glob_is_a_real_glob_not_a_substring() { assert_eq!(files.len(), 1, "{files:?}"); assert!(files[0].ends_with("a.rs")); - let frames = exchange(&[json!({ + let frames = find_frames(json!({ "op": "fs_find", "path": dir.path().to_str().unwrap(), "pattern": "TODO", "glob": "?.rs" - }) - .to_string()]) + })) .await; assert_eq!( frames.iter().filter(|f| type_of(f) == "match").count(), @@ -79,12 +98,11 @@ async fn find_glob_is_a_real_glob_not_a_substring() { #[tokio::test] async fn find_rejects_bad_pattern() { let dir = tempfile::tempdir().unwrap(); - let frames = exchange(&[json!({ + let frames = find_frames(json!({ "op": "fs_find", "path": dir.path().to_str().unwrap(), "pattern": "(" - }) - .to_string()]) + })) .await; assert_eq!(type_of(&frames[0]), "error"); assert_eq!(frames[0]["kind"], "bad_request"); @@ -159,12 +177,11 @@ async fn find_skips_a_file_over_the_size_bound() { .await .unwrap(); - let frames = exchange(&[json!({ + let frames = find_frames(json!({ "op": "fs_find", "path": dir.path().to_str().unwrap(), "pattern": "foo" - }) - .to_string()]) + })) .await; assert_eq!(frames.iter().filter(|f| type_of(f) == "match").count(), 0);