diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index 3982f75a9..1aadb48a8 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -1047,4 +1047,164 @@ mod tests { ); done.expect("served"); } + + #[tokio::test] + async fn a_line_that_is_not_a_request_gets_an_error_and_the_server_goes_on() { + let (resp, done) = exchange(b"not json\n".to_vec(), REQUEST_DEADLINES).await; + assert!( + matches!(&resp, Response::Err { message } if message.starts_with("bad request:")), + "{resp:?}" + ); + done.expect("served"); + } + + #[tokio::test] + async fn a_sender_that_closes_without_a_line_gets_no_reply() { + let (client, server) = UnixStream::pair().expect("socketpair"); + drop(client); + serve_one(server, REQUEST_DEADLINES, async |_| { + panic!("no request was sent") + }) + .await + .expect("an empty connection is not an error"); + } + + /// Drop `table` from the database `file` of a test's stores through a + /// second connection, so every later call on that store fails. + fn break_table(scratch: &Scratch, file: &str, table: &str) { + rusqlite::Connection::open(scratch.0.path().join(file)) + .expect("a second connection") + .execute(&format!("DROP TABLE {table}"), []) + .expect("the table drops"); + } + + #[test] + fn a_store_failure_is_an_error_response_and_never_a_wake() { + let (scratch, todos, reminders) = stores(); + break_table(&scratch, "todos.sqlite", "todos"); + break_table(&scratch, "reminders.sqlite", "reminders"); + let wake = Notify::new(); + let bus = Bus::new(); + let requests = [ + upsert("s", Some("k"), "x"), + Request::ClearTodo { + subsystem: "s".to_owned(), + key: Some("k".to_owned()), + all: false, + }, + Request::ClearTodo { + subsystem: "s".to_owned(), + key: None, + all: true, + }, + Request::ListTodos { subsystem: None }, + Request::MarkTodoDone { id: 1 }, + Request::MarkTodosDone { ids: vec![1, 2] }, + Request::StoreReminder { + message: "later".to_owned(), + timing: hive_sh4re::inbox::ReminderTiming::InSeconds { seconds: 60 }, + file_path: None, + }, + Request::ListReminders, + Request::CancelReminder { id: 1 }, + Request::CountPendingReminders, + Request::ReminderRollup { since_secs: 0 }, + ]; + for req in requests { + let label = format!("{req:?}"); + let resp = dispatch(req, &todos, &wake, Some(&reminders), &bus); + assert!(matches!(resp, Response::Err { .. }), "{label}: {resp:?}"); + } + assert!(!woke(&wake), "a failed upsert wakes nothing"); + } + + #[test] + fn a_transcribe_request_is_refused_by_the_store_dispatch() { + let (_scratch, todos, reminders) = stores(); + let resp = dispatch( + Request::Transcribe { + path: "/tmp/memo.ogg".to_owned(), + }, + &todos, + &Notify::new(), + Some(&reminders), + &Bus::new(), + ); + assert!( + matches!(&resp, Response::Err { message } if message.contains("connection handler")), + "{resp:?}" + ); + } + + #[tokio::test] + async fn binding_makes_the_directory_replaces_a_stale_socket_and_opens_the_mode() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("run/nested/agent.sock"); + let first = bind(&path).expect("the parent directories are created"); + drop(first); + assert!(path.exists(), "the dropped listener leaves its file behind"); + let _second = bind(&path).expect("a stale socket does not block the bind"); + let mode = std::fs::metadata(&path) + .expect("metadata") + .permissions() + .mode(); + assert_eq!(mode & 0o777, 0o666, "any uid in the container can dial it"); + } + + #[test] + fn binding_fails_with_the_path_when_the_directory_cannot_be_made() { + let dir = tempfile::tempdir().expect("tempdir"); + let blocker = dir.path().join("file"); + std::fs::write(&blocker, b"").expect("a plain file"); + let e = bind(&blocker.join("agent.sock")).expect_err("a file is not a directory"); + assert!(format!("{e:#}").contains("in-agent socket dir"), "{e:#}"); + } + + /// The real path a producer takes: dial the bound socket, send a request + /// line, read the answer, with the stores and the wake behind the server. + #[tokio::test] + async fn requests_over_the_bound_socket_reach_the_stores_and_the_wake() { + let (scratch, todos, reminders) = stores(); + let path = scratch.0.path().join("agent.sock"); + let listener = bind(&path).expect("bind"); + let (todos, reminders) = (Arc::new(todos), Arc::new(reminders)); + let wake = Arc::new(Notify::new()); + let bus = Bus::new(); + let server = tokio::spawn({ + let (todos, reminders, wake) = (todos.clone(), reminders.clone(), wake.clone()); + accept_loop(listener, 4, move |stream| { + let (todos, reminders, wake, bus) = + (todos.clone(), reminders.clone(), wake.clone(), bus.clone()); + async move { + handle_conn(stream, &todos, &wake, Some(&reminders), &bus) + .await + .expect("served"); + } + }) + }); + let ask = |req: Request| { + let path = path.clone(); + async move { + hive_sock_client::request::<_, Response>(&path, &req, hive_sock_client::Retry::None) + .await + .expect("an answer") + } + }; + + let resp = ask(upsert("forge", Some("pr-1"), "review the PR")).await; + assert!(matches!(resp, Response::Ok), "{resp:?}"); + assert!(woke(&wake), "a new todo wakes the serve loop"); + let resp = ask(Request::ListTodos { subsystem: None }).await; + assert!( + matches!(&resp, Response::LooseEnds { loose_ends } + if matches!(loose_ends.as_slice(), [LooseEnd::Todo { summary, .. }] if summary == "review the PR")), + "{resp:?}" + ); + let resp = ask(Request::CountPendingReminders).await; + assert!( + matches!(resp, Response::PendingRemindersCount { count: 0 }), + "{resp:?}" + ); + server.abort(); + } }