diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4fb67f2..133c3f9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -108,6 +108,14 @@ jobs: SOULSEEK_E2E_REQUIRED: "1" run: cargo test -p soulseek-rs --test cli_e2e -- --nocapture + # The control protocol as a third party meets it: the TCP transport, the + # token, and the gate. None of it is reachable through the command + # surface, so the CLI suite cannot cover it in either mode. + - name: Run daemon protocol e2e tests (required against soulfind) + env: + SOULSEEK_E2E_REQUIRED: "1" + run: cargo test -p soulseek-rs --test daemon_protocol_e2e -- --nocapture --test-threads=1 + # The same suite again, with every session served by a daemon instead of # the command's own login. Identical records and exit codes both ways is # what makes daemon mode a routing decision rather than a second diff --git a/soulseek-rs/src/daemon/server/transport.rs b/soulseek-rs/src/daemon/server/transport.rs index 649ffb8..84169f1 100644 --- a/soulseek-rs/src/daemon/server/transport.rs +++ b/soulseek-rs/src/daemon/server/transport.rs @@ -181,9 +181,13 @@ fn handle(stream: Stream, daemon: &Daemon) -> std::io::Result<()> { let writer = std::thread::spawn(move || write_loop(stream, &outbox)); if !authenticate(&mut reader, daemon, trusted, &out) { - closer.shutdown(); + // Let the writer flush the refusal before closing: shutting the socket + // first discards the reply, and the caller learns only that the + // connection vanished — which is indistinguishable from the daemon + // having crashed. drop(out); let _ = writer.join(); + closer.shutdown(); return Ok(()); } // An established client is expected to idle, holding the connection open @@ -237,10 +241,10 @@ fn handle(stream: Stream, daemon: &Daemon) -> std::io::Result<()> { } daemon.hub.unsubscribe(subscription); - closer.shutdown(); drop(out); let _ = pump.join(); let _ = writer.join(); + closer.shutdown(); Ok(()) } diff --git a/soulseek-rs/src/ui/main_tui/mod.rs b/soulseek-rs/src/ui/main_tui/mod.rs index 66e036d..386c5a4 100644 --- a/soulseek-rs/src/ui/main_tui/mod.rs +++ b/soulseek-rs/src/ui/main_tui/mod.rs @@ -264,3 +264,227 @@ pub fn launch_main_tui( ); tui.run(terminal) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::daemon::proto::ChatMessageDto; + use crate::models::MessageDirection; + + /// A session that knows a conversation and nothing else — what attaching + /// to a daemon looks like from the TUI's side. + struct TalkativeSession(Vec); + + impl SessionApi for TalkativeSession { + fn message_history(&self) -> Vec { + self.0.clone() + } + fn username(&self) -> String { + "tester".to_string() + } + fn listen_port(&self) -> Option { + None + } + fn session_loss(&self) -> Option { + None + } + fn search_with_cancel( + &self, + _query: &str, + _timeout: Duration, + _cancel: Option>, + ) -> soulseek_rs::Result> { + Ok(Vec::new()) + } + fn get_search_results( + &self, + _key: &str, + ) -> Vec { + Vec::new() + } + fn get_search_results_count(&self, _key: &str) -> usize { + 0 + } + fn try_get_search_results( + &self, + _key: &str, + ) -> Option> { + None + } + fn start_wishlist_search( + &self, + _query: &str, + ) -> soulseek_rs::Result<()> { + Ok(()) + } + fn wishlist_interval(&self) -> Duration { + Duration::from_mins(12) + } + fn download( + &self, + _filename: String, + _username: String, + _size: u64, + _directory: String, + ) -> soulseek_rs::Result<( + soulseek_rs::types::Download, + std::sync::mpsc::Receiver, + )> { + Err(soulseek_rs::SoulseekRs::NotConnected) + } + fn download_with_metadata( + &self, + _filename: String, + _username: String, + _size: u64, + _directory: String, + _metadata: soulseek_rs::types::DownloadMetadata, + ) -> soulseek_rs::Result<( + soulseek_rs::types::Download, + std::sync::mpsc::Receiver, + )> { + Err(soulseek_rs::SoulseekRs::NotConnected) + } + fn get_all_downloads(&self) -> Vec { + Vec::new() + } + fn pause_download(&self, _u: &str, _f: &str) -> bool { + false + } + fn resume_download(&self, _u: &str, _f: &str) -> bool { + false + } + fn remove_queued_download(&self, _u: &str, _f: &str) -> bool { + false + } + fn remove_download(&self, _u: &str, _f: &str) -> bool { + false + } + fn uploads(&self) -> Vec { + Vec::new() + } + fn take_upload_events(&self) -> Vec { + Vec::new() + } + fn cancel_upload(&self, _u: &str, _f: &str) -> bool { + false + } + fn set_upload_slots(&self, _slots: usize) {} + fn check_privileges(&self) -> soulseek_rs::Result<()> { + Ok(()) + } + fn own_privilege_seconds(&self) -> Option { + None + } + fn request_room_list(&self) -> soulseek_rs::Result<()> { + Ok(()) + } + fn join_room(&self, _room: &str) -> soulseek_rs::Result<()> { + Ok(()) + } + fn leave_room(&self, _room: &str) -> soulseek_rs::Result<()> { + Ok(()) + } + fn say_in_room( + &self, + _room: &str, + _message: &str, + ) -> soulseek_rs::Result<()> { + Ok(()) + } + fn room_members(&self, _room: &str) -> Vec { + Vec::new() + } + fn take_room_events(&self) -> Vec { + Vec::new() + } + fn send_private_message( + &self, + _u: &str, + _m: &str, + ) -> soulseek_rs::Result<()> { + Ok(()) + } + fn take_private_messages(&self) -> Vec { + Vec::new() + } + fn browse_user(&self, _username: &str) -> soulseek_rs::Result<()> { + Ok(()) + } + fn take_browse_result( + &self, + _username: &str, + ) -> Option> { + None + } + fn request_user_info( + &self, + _username: &str, + ) -> soulseek_rs::Result<()> { + Ok(()) + } + fn user_info(&self, _username: &str) -> Option { + None + } + fn shared_counts(&self) -> (u32, u32) { + (0, 0) + } + fn shared_directories(&self) -> Vec { + Vec::new() + } + fn set_shared_directories( + &self, + _directories: Vec, + ) -> soulseek_rs::Result<()> { + Ok(()) + } + } + + fn tui(history: Vec) -> MainTui { + MainTui::new( + Arc::new(TalkativeSession(history)), + "/tmp".to_string(), + 1, + Duration::from_secs(1), + None, + ) + } + + #[test] + fn an_attached_session_opens_with_the_conversation_already_in_it() { + // A TUI attached to a daemon has no state file of its own; the + // conversation has to come from the session, or it starts blank while + // the daemon has been collecting messages for hours. + let tui = tui(vec![ + ChatMessageDto { + peer: "bob".into(), + outgoing: false, + text: "hi".into(), + at: 1, + }, + ChatMessageDto { + peer: "bob".into(), + outgoing: true, + text: "hello".into(), + at: 2, + }, + ]); + + assert_eq!(tui.state.messages.len(), 2); + assert_eq!(tui.state.messages[0].peer, "bob"); + assert_eq!(tui.state.messages[0].direction, MessageDirection::Incoming); + assert_eq!( + tui.state.messages[1].direction, + MessageDirection::Outgoing, + "both halves of the conversation, not just what arrived" + ); + assert_eq!(tui.state.chat_peers(), ["bob"]); + } + + #[test] + fn a_local_session_starts_from_its_own_snapshot_instead() { + // Locally the session knows no history and the state file is the whole + // record, so nothing is invented. + assert!(tui(Vec::new()).state.messages.is_empty()); + } +} diff --git a/soulseek-rs/tests/cli_e2e.rs b/soulseek-rs/tests/cli_e2e.rs index 191a258..904285d 100644 --- a/soulseek-rs/tests/cli_e2e.rs +++ b/soulseek-rs/tests/cli_e2e.rs @@ -701,13 +701,34 @@ impl TestServer { format!("{}:{}", self.host, self.port) } - /// Arguments that point the binary at this server as `user`. + /// Arguments that point the binary at this server as `user` — or, in + /// daemon mode, at a daemon logged in as `user`. + /// + /// Every helper that builds a command goes through here, so routing at + /// this one point is what puts the streaming and wishlist tests through a + /// daemon too, rather than only the ones that happen to use [`cli`]. /// /// The listener stays on: a Soulseek transfer is delivered over a /// connection the *uploader* opens back to us, so a downloader without a /// listener never receives its file. fn args(&self, user: &str) -> Vec { - self.args_on(user, free_port().expect("listener port")) + self.args_with(user, &[]) + } + + /// The same, with flags that configure the *session* rather than the + /// command — shares, the listener, where downloads land. Locally they go + /// to the run itself; in daemon mode they configure the daemon, because + /// that is whose session they now describe. + fn args_with(&self, user: &str, session_flags: &[String]) -> Vec { + match mode() { + Mode::Local => { + let mut args = + self.args_on(user, free_port().expect("listener port")); + args.extend(session_flags.iter().cloned()); + args + } + Mode::Daemon => self.daemon_args(user, session_flags), + } } /// The same arguments on a chosen listener port, for the tests about what @@ -746,6 +767,19 @@ impl TestServer { ] } + /// The state directory of the daemon serving `user`, where its socket and + /// token live. + fn daemon_state(&self, user: &str) -> PathBuf { + self.daemon_for(user, &[]); + let key = format!("{}|{user}", self.address()); + let daemons = self.daemons.lock().expect("daemon registry"); + daemons + .get(&key) + .expect("the daemon was just started") + .state + .clone() + } + fn daemon_for(&self, user: &str, session_flags: &[String]) -> PathBuf { let key = format!("{}|{user}", self.address()); let mut daemons = self.daemons.lock().expect("daemon registry"); @@ -893,17 +927,14 @@ fn probe_bytes() -> Vec { /// the whole point: if the records and exit code come out the same either way, /// daemon mode is equivalent. fn cli(server: &TestServer, user: &str, args: &[&str]) -> Output { - let mut all = match mode() { - Mode::Local => server.args(user), - Mode::Daemon => { - let (session_flags, _) = split_session_flags(args); - server.daemon_args(user, &session_flags) + let (session_flags, command_args) = match mode() { + // Locally a flag is a flag: the run configures its own session. + Mode::Local => { + (Vec::new(), args.iter().map(|a| (*a).to_string()).collect()) } + Mode::Daemon => split_session_flags(args), }; - let command_args: Vec = match mode() { - Mode::Local => args.iter().map(|a| (*a).to_string()).collect(), - Mode::Daemon => split_session_flags(args).1, - }; + let mut all = server.args_with(user, &session_flags); all.extend(command_args); let refs: Vec<&str> = all.iter().map(String::as_str).collect(); run(&refs) @@ -3221,3 +3252,146 @@ fn concurrent_downloads_sharing_a_listener_port_all_land() { assert_eq!(written, probe_bytes(), "run {i} wrote the wrong bytes"); } } + +// --------------------------------------------------------------------------- +// Daemon mode +// +// Everything above proves a command behaves the same whichever kind of session +// serves it. These prove the things that only exist *because* there is a +// daemon: one session shared by several clients at once, the remote transport +// and its token, and the gate that decides who may talk to it at all. +// --------------------------------------------------------------------------- + +/// Two clients, one session, at the same time. +/// +/// This is the whole reason the daemon exists: the server allows one login per +/// account, so without it a second command would displace the first and both +/// would start reporting a lost session. +#[test] +fn two_clients_share_one_daemon_session_at_once() { + if mode() != Mode::Daemon { + println!("skipped: this is a daemon-mode property"); + return; + } + let server = server_or_skip!(); + let share = Scratch::new("share"); + std::fs::write(share.path().join("cli_probe_shared.bin"), probe_bytes()) + .expect("share file"); + let _sharer = + server.client("cli_e2e_two_client_peer", vec![share.display()]); + settle(); + + // One daemon, because both runs use the same account. + let args = server.args("cli_e2e_two_clients"); + let spawn = |extra: &[&str]| { + let mut all = args.clone(); + all.extend(extra.iter().map(|a| (*a).to_string())); + let refs: Vec<&str> = all.iter().map(String::as_str).collect(); + command(&refs) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("the binary should start") + }; + + let first = spawn(&["--json", "--search-timeout", "8", "search", "shared"]); + let second = spawn(&["--json", "browse", "cli_e2e_two_client_peer"]); + + let first = first.wait_with_output().expect("wait"); + let second = second.wait_with_output().expect("wait"); + + assert_eq!( + code(&first), + EXIT_OK, + "the search failed while another client was attached: {}", + stderr(&first) + ); + assert_eq!( + code(&second), + EXIT_OK, + "the browse failed while another client was attached: {}", + stderr(&second) + ); + assert!( + !records(&first).is_empty() && !records(&second).is_empty(), + "both clients should have got answers" + ); + // Neither may report the session being taken over: that is exactly what a + // second *login* would have caused, and what borrowing one avoids. + for output in [&first, &second] { + assert_ne!(code(output), EXIT_SESSION_LOST, "{}", stderr(output)); + } +} + +/// A daemon reports itself, and stops when asked. +#[test] +fn a_daemon_reports_its_session_and_stops_on_request() { + if mode() != Mode::Daemon { + println!("skipped: there is no daemon to ask in local mode"); + return; + } + let server = server_or_skip!(); + // A daemon of its own: this test stops it, and the shared fixtures are + // still serving the rest of the suite. + let user = "cli_e2e_daemon_status"; + let state = server.daemon_state(user); + let socket = state.join("daemon.sock"); + let at = socket.display().to_string(); + + let status = + run(&["--no-config", "--json", "--daemon", &at, "daemon", "status"]); + assert_eq!(code(&status), EXIT_OK, "stderr: {}", stderr(&status)); + let record: serde_json::Value = + serde_json::from_str(records(&status).first().expect("one record")) + .expect("valid JSON"); + assert_eq!(record["user"], user); + assert_eq!(record["server"], server.address()); + assert!( + record["clients"].as_u64().is_some_and(|n| n >= 1), + "the asking client counts as attached: {record}" + ); + + let stopped = run(&["--no-config", "--daemon", &at, "daemon", "stop"]); + assert_eq!(code(&stopped), EXIT_OK, "stderr: {}", stderr(&stopped)); + + // The socket goes away, and a command that finds nothing there says so + // rather than hanging. + let deadline = Instant::now() + Duration::from_secs(20); + while socket.exists() && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(200)); + } + let after = run(&["--no-config", "--daemon", &at, "daemon", "status"]); + assert_eq!( + code(&after), + EXIT_CONNECTION, + "a stopped daemon is a connection error, not a hang: {}", + stderr(&after) + ); +} + +/// `daemon token` prints a secret, and the same one every time. +#[test] +fn the_daemon_token_is_stable_across_reads() { + let state = Scratch::new("token"); + let read = || { + let mut command = Command::new(BIN); + for name in CLI_ENV_VARS { + command.env_remove(name); + } + command.env("SOULSEEK_STATE_DIR", state.path()); + command.current_dir(std::env::temp_dir()); + command.args(["--no-config", "daemon", "token"]); + command.output().expect("the binary should run") + }; + + let first = read(); + assert_eq!(code(&first), EXIT_OK, "stderr: {}", stderr(&first)); + let token = records(&first).first().expect("a token").clone(); + assert_eq!(token.len(), 64, "256 bits, hex encoded"); + assert!(token.chars().all(|c| c.is_ascii_hexdigit())); + assert_eq!( + records(&read()).first().expect("a token"), + &token, + "a restart must not lock existing clients out" + ); +} diff --git a/soulseek-rs/tests/daemon_protocol_e2e.rs b/soulseek-rs/tests/daemon_protocol_e2e.rs new file mode 100644 index 0000000..06ea7f1 --- /dev/null +++ b/soulseek-rs/tests/daemon_protocol_e2e.rs @@ -0,0 +1,494 @@ +//! The control protocol as a third party meets it: a socket, JSON-RPC lines, +//! and the gate that decides who gets past `auth`. +//! +//! The CLI suite drives the daemon through this binary, which means it only +//! ever sends well-formed requests over a Unix socket. Everything here is the +//! other half — the TCP transport and its token, and what happens to a caller +//! who is hostile, broken, or simply not authenticated. None of it is +//! reachable through the command surface, and all of it is load-bearing: on +//! the socket, connecting *is* authenticating. +//! +//! Like the rest of the e2e suite this needs a soulfind server and skips +//! without one; `SOULSEEK_E2E_REQUIRED=1` turns that skip into a failure. + +use std::io::{BufRead, BufReader, Write}; +use std::net::TcpStream; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +const BIN: &str = env!("CARGO_BIN_EXE_soulseek-rs"); + +/// Every environment variable the binary reads, cleared so a developer's shell +/// cannot change what a test observes — and so no run finds a real daemon. +const CLI_ENV_VARS: [&str; 15] = [ + "SOULSEEK_CONFIG_DIR", + "SOULSEEK_STATE_DIR", + "SOULSEEK_DAEMON", + "SOULSEEK_DAEMON_TOKEN", + "SOULSEEK_USERNAME", + "SOULSEEK_PASSWORD", + "SOULSEEK_PASSWORD_CMD", + "SOULSEEK_SERVER", + "SOULSEEK_NO_LISTENER", + "SOULSEEK_LISTENER_PORT", + "SOULSEEK_DOWNLOAD_DIR", + "SOULSEEK_SHARED_DIR", + "SOULSEEK_MAX_CONCURRENT_DOWNLOADS", + "SOULSEEK_SEARCH_TIMEOUT", + "SOULSEEK_CONFIG", +]; + +/// A daemon bound to loopback TCP as well as its socket, so both transports +/// can be exercised against one session. +struct Daemon { + child: Child, + state: PathBuf, + port: u16, + _server: Soulfind, +} + +impl Daemon { + fn start() -> Option { + let server = Soulfind::start()?; + let state = scratch("daemon-proto"); + let port = free_port()?; + + let mut command = Command::new(BIN); + for name in CLI_ENV_VARS { + command.env_remove(name); + } + command.env("SOULSEEK_STATE_DIR", &state); + command.current_dir(std::env::temp_dir()); + command.args([ + "--no-config", + "--server", + &server.address(), + "--username", + "cli_e2e_proto", + "--password", + "pw", + "--listener-port", + &free_port()?.to_string(), + "daemon", + "--bind", + &format!("127.0.0.1:{port}"), + ]); + command.stdout(Stdio::null()).stderr(Stdio::null()); + let child = command.spawn().ok()?; + + let daemon = Self { + child, + state, + port, + _server: server, + }; + daemon.wait_until_ready()?; + Some(daemon) + } + + /// Ready means logged in, not merely bound: the daemon binds before it + /// logs in, and a client that connects between the two would be asking + /// about a session that is not there yet. + fn wait_until_ready(&self) -> Option<()> { + let deadline = Instant::now() + Duration::from_mins(1); + while Instant::now() < deadline { + if let Some(mut connection) = self.connect_tcp() + && let Some(reply) = connection.call(&format!( + r#"{{"jsonrpc":"2.0","id":1,"method":"auth","params":{{"protocol":1,"token":"{}"}}}}"#, + self.token() + )) + && reply.contains("\"username\"") + { + return Some(()); + } + std::thread::sleep(Duration::from_millis(250)); + } + None + } + + fn token(&self) -> String { + std::fs::read_to_string(self.state.join("daemon.token")) + .map(|token| token.trim().to_string()) + .unwrap_or_default() + } + + fn socket_path(&self) -> PathBuf { + self.state.join("daemon.sock") + } + + fn connect_tcp(&self) -> Option { + TcpStream::connect(("127.0.0.1", self.port)) + .ok() + .map(|stream| { + let read = stream.try_clone().expect("a second handle"); + Connection::from_parts(Box::new(read), Box::new(stream)) + }) + } + + #[cfg(unix)] + fn connect_socket(&self) -> Option { + let stream = + std::os::unix::net::UnixStream::connect(self.socket_path()).ok()?; + let read = stream.try_clone().expect("a second handle"); + Some(Connection::from_parts(Box::new(read), Box::new(stream))) + } +} + +impl Drop for Daemon { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + let _ = std::fs::remove_dir_all(&self.state); + } +} + +/// One control connection, spoken to by hand. +struct Connection { + reader: BufReader>, + writer: Box, +} + +impl Connection { + /// Boxed halves, so one type covers both transports. + fn from_parts( + read: Box, + write: Box, + ) -> Self { + Self { + reader: BufReader::new(read), + writer: write, + } + } + + /// Send one line and read the reply, `None` if the daemon hung up first. + fn call(&mut self, line: &str) -> Option { + self.send(line)?; + self.read() + } + + fn send(&mut self, line: &str) -> Option<()> { + writeln!(self.writer, "{line}").ok()?; + self.writer.flush().ok() + } + + fn read(&mut self) -> Option { + let mut reply = String::new(); + match self.reader.read_line(&mut reply) { + Ok(0) | Err(_) => None, + Ok(_) => Some(reply), + } + } + + fn authenticate(&mut self, token: Option<&str>) -> Option { + let params = match token { + Some(token) => format!(r#"{{"protocol":1,"token":"{token}"}}"#), + None => r#"{"protocol":1}"#.to_string(), + }; + self.call(&format!( + r#"{{"jsonrpc":"2.0","id":1,"method":"auth","params":{params}}}"# + )) + } +} + +macro_rules! daemon_or_skip { + () => { + match Daemon::start() { + Some(daemon) => daemon, + None => { + let required = std::env::var("SOULSEEK_E2E_REQUIRED") + .is_ok_and(|v| v != "0" && !v.is_empty()); + assert!( + !required, + "SOULSEEK_E2E_REQUIRED is set but no daemon could be \ + started (set SOULFIND_BIN)" + ); + println!( + "daemon protocol e2e skipped: no soulfind server (set \ + SOULFIND_BIN to run)" + ); + return; + } + } + }; +} + +#[test] +fn tcp_needs_the_token_and_the_socket_does_not() { + let daemon = daemon_or_skip!(); + + let mut anonymous = daemon.connect_tcp().expect("connect"); + let refused = anonymous.authenticate(None).expect("an answer"); + assert!( + refused.contains("\"error\""), + "TCP without a token must be refused: {refused}" + ); + + let mut wrong = daemon.connect_tcp().expect("connect"); + let bad = wrong + .authenticate(Some(&"0".repeat(64))) + .expect("an answer"); + assert!( + bad.contains("\"error\""), + "a wrong token must be refused: {bad}" + ); + + let mut good = daemon.connect_tcp().expect("connect"); + let accepted = good.authenticate(Some(&daemon.token())).expect("an answer"); + assert!( + accepted.contains("\"result\"") && accepted.contains("cli_e2e_proto"), + "the right token must be accepted: {accepted}" + ); + + #[cfg(unix)] + { + // The socket's permissions have already answered the question. + let mut local = daemon.connect_socket().expect("connect"); + let reply = local.authenticate(None).expect("an answer"); + assert!( + reply.contains("\"result\""), + "a Unix socket needs no token: {reply}" + ); + } +} + +#[test] +fn nothing_can_be_asked_before_authenticating() { + let daemon = daemon_or_skip!(); + let mut connection = daemon.connect_tcp().expect("connect"); + + let reply = connection + .call(r#"{"jsonrpc":"2.0","id":1,"method":"daemon.status"}"#) + .expect("an answer"); + assert!( + reply.contains("\"error\""), + "a request before auth must be refused: {reply}" + ); + assert!( + !reply.contains("cli_e2e_proto"), + "an unauthenticated caller must learn nothing about the session: \ + {reply}" + ); + assert!( + connection + .call(r#"{"jsonrpc":"2.0","id":2,"method":"daemon.status"}"#) + .is_none(), + "the connection must be dropped, not left open to try again" + ); +} + +#[test] +fn an_error_carries_an_id_so_a_client_can_match_it() { + let daemon = daemon_or_skip!(); + let mut connection = daemon.connect_tcp().expect("connect"); + connection + .authenticate(Some(&daemon.token())) + .expect("authenticated"); + + let reply = connection + .call(r#"{"jsonrpc":"2.0","id":7,"method":"no.such.method"}"#) + .expect("an answer"); + assert!(reply.contains("\"id\":7"), "{reply}"); + assert!(reply.contains("-32601"), "{reply}"); + + // A malformed line still gets an id — null — rather than looking like a + // notification the client should ignore. + let malformed = connection.call("{ this is not json").expect("an answer"); + assert!(malformed.contains("\"id\":null"), "{malformed}"); + assert!(malformed.contains("-32700"), "{malformed}"); +} + +#[test] +fn a_notification_is_answered_with_silence() { + let daemon = daemon_or_skip!(); + let mut connection = daemon.connect_tcp().expect("connect"); + connection + .authenticate(Some(&daemon.token())) + .expect("authenticated"); + + // No id, so no reply — the next line read must be the answer to the + // request that follows it, not a reply to this one. + connection + .send(r#"{"jsonrpc":"2.0","method":"no.such.method"}"#) + .expect("sent"); + let reply = connection + .call(r#"{"jsonrpc":"2.0","id":9,"method":"daemon.status"}"#) + .expect("an answer"); + assert!( + reply.contains("\"id\":9"), + "a notification must not produce a reply of its own: {reply}" + ); +} + +#[test] +fn an_oversized_line_cannot_make_the_daemon_allocate() { + let daemon = daemon_or_skip!(); + let mut connection = daemon.connect_tcp().expect("connect"); + + // Over the 1 MiB cap, and never terminated: an unauthenticated peer must + // not be able to make the daemon buffer without bound. + let flood = "x".repeat(2 * 1024 * 1024); + let _ = connection.send(&flood); + assert!( + connection.read().is_none(), + "an oversized request must end the connection" + ); + + // The daemon is still serving everyone else. + let mut healthy = daemon.connect_tcp().expect("connect"); + let reply = healthy + .authenticate(Some(&daemon.token())) + .expect("an answer"); + assert!(reply.contains("\"result\""), "{reply}"); +} + +#[test] +fn a_silent_connection_is_closed_rather_than_held_open() { + let daemon = daemon_or_skip!(); + let mut connection = daemon.connect_tcp().expect("connect"); + + // Says nothing at all. The handshake timeout is 30s; well inside a minute. + let deadline = Instant::now() + Duration::from_mins(1); + let mut closed = false; + while Instant::now() < deadline { + if connection.read().is_none() { + closed = true; + break; + } + } + assert!( + closed, + "a peer that connects and never authenticates must not hold a thread \ + forever" + ); +} + +#[test] +fn discover_serves_a_contract_whose_parameters_match_what_is_accepted() { + let daemon = daemon_or_skip!(); + let mut connection = daemon.connect_tcp().expect("connect"); + connection + .authenticate(Some(&daemon.token())) + .expect("authenticated"); + + let reply = connection + .call(r#"{"jsonrpc":"2.0","id":1,"method":"rpc.discover"}"#) + .expect("an answer"); + let document: serde_json::Value = + serde_json::from_str(&reply).expect("valid JSON"); + let methods = document["result"]["methods"] + .as_array() + .expect("the contract lists methods"); + + // The published parameter names have to be the keys the daemon reads. + // Published as one descriptor named `params`, a generated client would + // send {"params":{"params":{…}}} and every call would fail -32602 — which + // is exactly what this asserts cannot happen. + let join = methods + .iter() + .find(|m| m["name"] == "room.join") + .expect("room.join is published"); + let names: Vec<&str> = join["params"] + .as_array() + .expect("params is an array") + .iter() + .map(|p| p["name"].as_str().expect("a name")) + .collect(); + assert_eq!(names, ["room"], "{join}"); + + // And a call built from that description is accepted. + let accepted = connection + .call( + r#"{"jsonrpc":"2.0","id":2,"method":"room.join","params":{"room":"cli_e2e_proto_room"}}"#, + ) + .expect("an answer"); + assert!( + accepted.contains("\"result\""), + "a call shaped by the published contract must be accepted: {accepted}" + ); +} + +// --------------------------------------------------------------------------- +// Fixtures +// --------------------------------------------------------------------------- + +struct Soulfind { + child: Option, + db: PathBuf, + port: u16, +} + +impl Soulfind { + fn start() -> Option { + let bin = soulfind_binary()?; + let port = free_port()?; + let db = std::env::temp_dir().join(format!("soulfind-proto-{port}.db")); + let _ = std::fs::remove_file(&db); + + let child = Command::new(bin) + .args(["-d", db.to_str()?, "-p", &port.to_string()]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .ok()?; + + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if TcpStream::connect(("127.0.0.1", port)).is_ok() { + return Some(Self { + child: Some(child), + db, + port, + }); + } + std::thread::sleep(Duration::from_millis(100)); + } + let mut child = child; + let _ = child.kill(); + let _ = child.wait(); + None + } + + fn address(&self) -> String { + format!("127.0.0.1:{}", self.port) + } +} + +impl Drop for Soulfind { + fn drop(&mut self) { + if let Some(child) = self.child.as_mut() { + let _ = child.kill(); + let _ = child.wait(); + } + let _ = std::fs::remove_file(&self.db); + } +} + +fn soulfind_binary() -> Option { + if let Ok(path) = std::env::var("SOULFIND_BIN") { + let path = PathBuf::from(path); + return path.exists().then_some(path); + } + Path::new(env!("CARGO_MANIFEST_DIR")) + .ancestors() + .map(|dir| dir.join("soulfind/bin/soulfind")) + .find(|candidate| candidate.exists()) +} + +fn free_port() -> Option { + std::net::TcpListener::bind("127.0.0.1:0") + .ok()? + .local_addr() + .ok() + .map(|addr| addr.port()) +} + +fn scratch(label: &str) -> PathBuf { + static COUNTER: std::sync::atomic::AtomicU32 = + std::sync::atomic::AtomicU32::new(0); + let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let path = std::env::temp_dir() + .join(format!("soulseek-{label}-{}-{n}", std::process::id())); + std::fs::create_dir_all(&path).expect("scratch directory"); + path +}