feat(rust): add shared Signal client and workspace configuration

rust/signal/tests/send.rs

@@ -0,0 +1,147 @@
+#![cfg(test)]
+#![cfg(target_os = "linux")]
+
+use std::{
+    cell::RefCell,
+    io::{self, BufRead, BufReader, Write},
+    os::{fd::AsRawFd, unix::net::UnixStream},
+    thread,
+    time::{Duration, Instant},
+};
+
+use serde_json::{Value, json};
+use signal::Signal;
+
+thread_local! {
+    static CONNECTION: RefCell<Option<UnixStream>> = const { RefCell::new(None) };
+}
+
+// Patch libc's connect only in this integration-test executable. Each test
+// supplies its own socket pair; no call can reach the real Signal daemon.
+// SAFETY: This symbol uses libc's connect ABI and preserves its return/errno
+// convention. It never delegates to a real connection attempt.
+#[unsafe(no_mangle)]
+unsafe extern "C" fn connect(
+    socket: libc::c_int,
+    address: *const libc::sockaddr,
+    length: libc::socklen_t,
+) -> libc::c_int {
+    let path_offset = std::mem::offset_of!(libc::sockaddr_un, sun_path);
+    let expected = b"/run/signal-cli/socket\0";
+    if !address.is_null()
+        && let Some(path_end) = path_offset.checked_add(expected.len())
+        && usize::try_from(length).is_ok_and(|length| length >= path_end)
+    {
+        // SAFETY: connect's caller supplies an address valid for length bytes.
+        // The bounds check above covers the entire pathname slice.
+        let path = unsafe {
+            std::slice::from_raw_parts(address.cast::<u8>().add(path_offset), expected.len())
+        };
+        if path == expected
+            && let Some(stream) = CONNECTION.with_borrow_mut(Option::take)
+        {
+            // SAFETY: Both descriptors are live and owned by this thread.
+            // dup2 replaces the unconnected socket with the test endpoint.
+            return if unsafe { libc::dup2(stream.as_raw_fd(), socket) } < 0 {
+                -1
+            } else {
+                0
+            };
+        }
+    }
+    // SAFETY: __errno_location returns this thread's valid errno pointer.
+    unsafe { *libc::__errno_location() = libc::ECONNREFUSED };
+    -1
+}
+
+fn test_connection() -> UnixStream {
+    let (client, server) = UnixStream::pair().unwrap();
+    CONNECTION.with_borrow_mut(|connection| assert!(connection.replace(client).is_none()));
+    server
+}
+
+fn signal() -> Signal {
+    Signal {
+        from_phone: "1234567890".into(),
+    }
+}
+
+#[test]
+fn reuses_sender_for_different_recipients_and_ignores_unrelated_responses() {
+    let client = signal();
+    for (to_phone, body) in [
+        ("0987654321", "failure\nwith details"),
+        ("+15555550102", "another failure"),
+    ] {
+        let server = test_connection();
+        let task = thread::spawn(move || {
+            let mut reader = BufReader::new(server);
+            let mut line = String::new();
+            reader.read_line(&mut line).unwrap();
+            let request: Value = serde_json::from_str(&line).unwrap();
+            assert_eq!(
+                request,
+                json!({
+                    "jsonrpc": "2.0", "method": "send", "id": "signal-alert",
+                    "params": { "account": "1234567890", "recipient": [to_phone], "message": body },
+                })
+            );
+            reader
+                .get_mut()
+                .write_all(b"{\"method\":\"receive\"}\n{\"id\":\"other\"}\n{\"id\":\"signal-")
+                .unwrap();
+            reader
+                .get_mut()
+                .write_all(b"alert\",\"result\":{}}\n")
+                .unwrap();
+        });
+        client.send(to_phone, body).unwrap();
+        task.join().unwrap();
+    }
+}
+
+#[test]
+fn rejects_rpc_errors_invalid_json_and_premature_close() {
+    for (reply, expected) in [
+        (
+            "{\"id\":\"signal-alert\",\"error\":{\"code\":-1}}\n",
+            "Signal RPC returned an error",
+        ),
+        ("invalid\n", "Invalid Signal RPC response"),
+        ("", "socket closed"),
+        ("{\"id\":\"signal-alert\"}", "socket closed"),
+    ] {
+        let server = test_connection();
+        let task = thread::spawn(move || {
+            let mut reader = BufReader::new(server);
+            reader.read_line(&mut String::new()).unwrap();
+            reader.get_mut().write_all(reply.as_bytes()).unwrap();
+        });
+        let error = signal().send("0987654321", "failure").unwrap_err();
+        assert!(format!("{error:#}").contains(expected), "{error:#}");
+        task.join().unwrap();
+    }
+}
+
+#[test]
+fn reports_connection_failure() {
+    let error = signal().send("0987654321", "failure").unwrap_err();
+    assert_eq!(error.to_string(), "Failed to connect to Signal RPC socket");
+    assert_eq!(
+        error.downcast_ref::<io::Error>().unwrap().kind(),
+        io::ErrorKind::ConnectionRefused
+    );
+}
+
+#[test]
+fn unresponsive_server_times_out() {
+    let _server = test_connection();
+    let start = Instant::now();
+    let error = signal().send("0987654321", "failure").unwrap_err();
+    assert_eq!(error.to_string(), "Failed to read Signal RPC response");
+    assert!(matches!(
+        error.downcast_ref::<io::Error>().unwrap().kind(),
+        io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
+    ));
+    assert!(start.elapsed() >= Duration::from_secs(4));
+}