dumbgoy/soulseek-rs-modded
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge pull request #97 from michel/test/event-synced-e2e

Browse Source 
test: wait on events instead of timers in the flaky tests
This commit is contained in:
Michel de Graaf
2026-09-10 15:48:22 +02:00
committed by GitHub
parent 43d41ddb79 a6d4c8b3df
commit be92fd8b3a
6 changed files with 211 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
soulseek-rs-lib/src/client/rooms.rs
+4 -1
View File
@@ -150,8 +150,11 @@ impl Client {
/// Returns [`SoulseekRs::NotConnected`] if the client is not connected.
pub fn watch_user(&self, username: &str) -> Result<()> {
use crate::message::server::MessageFactory;
// Held across the send: the reply is applied under this lock, so it
// cannot land first and have its "does not exist" undone by the add.
let mut context = self.context.write_safe()?;
self.send_server_message(MessageFactory::build_watch_user(username))?;
self.context.write_safe()?.add_watched_user(username);
context.add_watched_user(username);
Ok(())
}
soulseek-rs-lib/tests/e2e.rs
+133 -64
View File
@@ -76,38 +76,51 @@ impl TestServer {
}
/// Spawn a local soulfind on an ephemeral port with a throwaway database.
///
/// soulfind exits if its port is taken between `free_port()` handing it
/// out and soulfind binding it, so a child that dies is retried on a
/// fresh port rather than read as a missing server.
fn spawn(gate: std::sync::MutexGuard<'static, ()>) -> Option<Self> {
let bin = soulfind_binary()?;
let port = free_port()?;
let db = std::env::temp_dir().join(format!("soulfind-e2e-{port}.db"));
let _ = std::fs::remove_file(&db);
for _ in 0..3 {
let port = free_port()?;
let db =
std::env::temp_dir().join(format!("soulfind-e2e-{port}.db"));
let _ = std::fs::remove_file(&db);
let mut child = Command::new(&bin)
.arg("-p")
.arg(port.to_string())
.arg("-d")
.arg(&db)
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.ok()?;
let mut child = Command::new(&bin)
.arg("-p")
.arg(port.to_string())
.arg("-d")
.arg(&db)
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.ok()?;
if wait_until_listening("127.0.0.1", port, Duration::from_secs(5))
.is_none()
{
// Server never came up (e.g. a toolchain/SQLite issue); skip.
// A child that has exited is never taken as up: soulfind exits
// when its port is gone, and whoever took it may answer instead.
let deadline = Instant::now() + Duration::from_secs(10);
while Instant::now() < deadline
&& child.try_wait().ok().flatten().is_none()
{
if TcpStream::connect(("127.0.0.1", port)).is_ok() {
return Some(Self {
host: "127.0.0.1".to_string(),
port,
child: Some(child),
db: Some(db),
_gate: gate,
});
}
std::thread::sleep(Duration::from_millis(100));
}
// Never came up (e.g. a toolchain/SQLite issue) or died.
let _ = child.kill();
let _ = child.wait();
return None;
let _ = std::fs::remove_file(&db);
}
Some(Self {
host: "127.0.0.1".to_string(),
port,
child: Some(child),
db: Some(db),
_gate: gate,
})
None
}
fn settings(&self, username: &str, password: &str) -> ClientSettings {
@@ -327,8 +340,15 @@ fn a_chat_room_message_is_delivered_between_users() {
alice.join_room(room).expect("alice joins room");
bob.join_room(room).expect("bob joins room");
// Give both joins time to register on the server before speaking.
std::thread::sleep(Duration::from_millis(500));
// Each is in once the server has sent them the member list; a line said
// before bob is in never reaches him.
let deadline = Instant::now() + Duration::from_secs(5);
while alice.room_members(room).is_empty()
|| bob.room_members(room).is_empty()
{
assert!(Instant::now() < deadline, "both should get into the room");
std::thread::sleep(Duration::from_millis(50));
}
let _ = alice.take_room_events();
let _ = bob.take_room_events();
@@ -526,7 +546,14 @@ fn open_control_and_await_queue(
let mut p = connect_retry(&cfg.listen_addr, Duration::from_secs(5))?;
p.set_read_timeout(Some(Duration::from_secs(10)))?;
p.write_all(&peer_init_bytes(&cfg.peer_username, "P", 0))?;
// The client answers a user-info request through the peer it registered
// for this connection, so the answer means a download queues on it
// rather than dialling a peer the server has never heard of.
p.write_all(
&soulseek_rs::message::peer::build_user_info_request().get_buffer(),
)?;
p.flush()?;
expect_code(&mut p, 16, Duration::from_secs(10))?;
let _ = cfg.ready.send(());
loop {
@@ -860,10 +887,6 @@ fn file_downloads_over_p_and_f(
.recv_timeout(Duration::from_secs(5))
.expect("mock uploader P connection");
// The listener registers an incoming peer under its plain username; give
// that registration a moment to complete before queuing the download.
std::thread::sleep(Duration::from_millis(1500));
let (_download, status_rx) = client
.download(
filename.to_string(),
@@ -943,7 +966,6 @@ fn cancel_mid_transfer(peer_username: &str, stall_after: Option<usize>) {
ready_rx
.recv_timeout(Duration::from_secs(5))
.expect("mock uploader P connection");
std::thread::sleep(Duration::from_millis(1500));
let (_download, status_rx) = client
.download(
@@ -1043,7 +1065,6 @@ fn a_cancelled_queued_download_declines_the_peers_offer() {
ready_rx
.recv_timeout(Duration::from_secs(5))
.expect("mock uploader P connection");
std::thread::sleep(Duration::from_millis(1500));
let download_dir = unique_download_dir();
let (_download, status_rx) = client
@@ -1117,7 +1138,6 @@ fn refusal_fails_the_download_quickly(
ready_rx
.recv_timeout(Duration::from_secs(5))
.expect("mock uploader P connection");
std::thread::sleep(Duration::from_millis(1500));
let download_dir = unique_download_dir();
let (_download, status_rx) = client
@@ -1217,7 +1237,6 @@ fn an_interrupted_download_resumes_from_its_partial_file() {
ready_rx
.recv_timeout(Duration::from_secs(5))
.expect("mock uploader P connection");
std::thread::sleep(Duration::from_millis(1500));
let (_download, status_rx) = client
.download(
@@ -2936,9 +2955,9 @@ fn a_joined_room_reports_its_members_statistics() {
assert!(sharer.login().expect("sharer login"));
sharer.join_room(room).expect("sharer joins room");
// Let the server register the sharer's membership and share counts, so
// the observer's own join reply describes a room that already has them.
std::thread::sleep(Duration::from_millis(750));
// The sharer's membership and share counts have to land first, so the
// observer's own join reply describes a room that already has them.
fence(&sharer, "e2e_stats_observer");
let mut observer =
Client::with_settings(server.settings("e2e_stats_observer", "pw"));
@@ -3019,9 +3038,9 @@ fn watching_a_user_returns_their_status_and_share_counts() {
watcher.connect().expect("watcher connect");
assert!(watcher.login().expect("watcher login"));
// Let the subject's share counts reach the server before watching, so
// the reply describes a user it already knows the statistics for.
std::thread::sleep(Duration::from_millis(750));
// The subject's share counts have to reach the server before watching,
// so the reply describes a user it already knows the statistics for.
fence(&subject, "e2e_watch_watcher");
watcher
.watch_user("e2e_watch_subject")
@@ -3352,7 +3371,6 @@ fn download_progress_reports_the_rate_actually_received() {
ready_rx
.recv_timeout(Duration::from_secs(5))
.expect("mock uploader P connection");
std::thread::sleep(Duration::from_millis(1500));
let (_download, status_rx) = client
.download(
@@ -3796,6 +3814,18 @@ fn await_room_event<T>(
None
}
/// Wait until the server has handled everything `client` has sent, including
/// requests it never answers: soulfind takes a connection's messages in order,
/// so the answer to a question about `other` asked now comes after them.
fn fence(client: &Client, other: &str) {
client.request_user_info(other).expect("ask about a user");
let deadline = Instant::now() + Duration::from_secs(5);
while client.user_info(other).is_none() {
assert!(Instant::now() < deadline, "the server should answer");
std::thread::sleep(Duration::from_millis(50));
}
}
#[test]
fn a_ticker_set_in_a_room_reaches_the_other_members() {
use soulseek_rs::types::RoomEvent;
@@ -3812,7 +3842,12 @@ fn a_ticker_set_in_a_room_reaches_the_other_members() {
alice.join_room(room).expect("alice joins");
bob.join_room(room).expect("bob joins");
std::thread::sleep(Duration::from_millis(500));
// A ticker reaches only who is in the room when it is set.
let deadline = Instant::now() + Duration::from_secs(5);
while bob.room_members(room).is_empty() {
assert!(Instant::now() < deadline, "bob should get into the room");
std::thread::sleep(Duration::from_millis(50));
}
let _ = bob.take_room_events();
alice
@@ -3856,11 +3891,15 @@ fn a_client_joining_later_receives_the_whole_ticker_board() {
alice.connect().expect("alice connect");
assert!(alice.login().expect("alice login"));
alice.join_room(room).expect("alice joins");
std::thread::sleep(Duration::from_millis(300));
alice
.set_room_ticker(room, "standing message")
.expect("set ticker");
std::thread::sleep(Duration::from_millis(500));
// The room, alice included, is told once the board holds it.
await_room_event(&alice, Duration::from_secs(5), |event| {
matches!(event, RoomEvent::TickerAdded { room: r, .. } if r == room)
.then_some(())
})
.expect("alice should see her ticker go up");
// Carol joins afterwards and must be handed the board that already exists.
let mut carol =
@@ -3901,7 +3940,7 @@ fn the_global_room_feed_carries_a_room_we_never_joined() {
watcher.join_global_room().expect("join global room");
talker.join_room(room).expect("talker joins a room");
std::thread::sleep(Duration::from_millis(500));
fence(&watcher, "e2e_global_talk");
let _ = watcher.take_room_events();
let body = "spoken where nobody is watching";
@@ -3925,7 +3964,7 @@ fn the_global_room_feed_carries_a_room_we_never_joined() {
// And leaving the feed stops it: the next message must not arrive.
watcher.leave_global_room().expect("leave global room");
std::thread::sleep(Duration::from_millis(500));
fence(&watcher, "e2e_global_talk");
let _ = watcher.take_room_events();
talker
.say_in_room(room, "after leaving")
@@ -3991,7 +4030,16 @@ fn a_room_search_is_answered_by_the_rooms_members() {
);
sharer.join_room(room).expect("sharer joins");
searcher.join_room(room).expect("searcher joins");
std::thread::sleep(Duration::from_millis(750));
// A room search goes to whoever is in the room when it is sent.
let deadline = Instant::now() + Duration::from_secs(5);
while !searcher
.room_members(room)
.iter()
.any(|member| member == "e2e_rs_sharer")
{
assert!(Instant::now() < deadline, "both should get into the room");
std::thread::sleep(Duration::from_millis(50));
}
let query = "relic";
searcher.search_room(room, query).expect("room search");
@@ -4028,7 +4076,7 @@ fn shared_interests_make_two_users_similar() {
alice
.add_dislike("e2e_interest_muzak")
.expect("alice dislikes something");
std::thread::sleep(Duration::from_millis(500));
fence(&bob, "e2e_like_alice");
// Who else likes this item (code 112).
let deadline = Instant::now() + Duration::from_secs(5);
@@ -4122,7 +4170,7 @@ fn recommendations_are_returned_for_our_interests() {
alice.add_interest(shared).expect("alice likes shared");
bob.add_interest(shared).expect("bob likes shared");
bob.add_interest(other).expect("bob likes other");
std::thread::sleep(Duration::from_millis(500));
fence(&bob, "e2e_rec_alice");
let deadline = Instant::now() + Duration::from_secs(8);
let mut recommended = Vec::new();
@@ -4256,7 +4304,23 @@ fn a_changed_password_is_what_the_next_login_needs() {
client
.change_password("second-pw")
.expect("change password");
std::thread::sleep(Duration::from_millis(500));
// soulfind stores the new password once it has hashed it, and forgets the
// change if our session closes first. The session stays until a login
// with the new password shows it landed; a refused one leaves it alone.
let deadline = Instant::now() + Duration::from_secs(10);
loop {
let mut fresh =
Client::with_settings(server.settings(user, "second-pw"));
fresh.connect().expect("connect with the new password");
if matches!(fresh.login(), Ok(true)) {
break;
}
assert!(
Instant::now() < deadline,
"the new password must be accepted"
);
std::thread::sleep(Duration::from_millis(100));
}
drop(client);
let mut stale = Client::with_settings(server.settings(user, "first-pw"));
@@ -4265,14 +4329,6 @@ fn a_changed_password_is_what_the_next_login_needs() {
!matches!(stale.login(), Ok(true)),
"the old password must stop working"
);
drop(stale);
let mut fresh = Client::with_settings(server.settings(user, "second-pw"));
fresh.connect().expect("connect with the new password");
assert!(
fresh.login().expect("login with the new password"),
"the new password must be accepted"
);
}
#[test]
@@ -4436,6 +4492,11 @@ fn phrases_the_server_excludes_are_kept_for_our_replies() {
// network (code 160). Nicotine+ applies them to the files it offers in a
// search reply, not to the searches it sends, and so do we — this pins
// that the announced list arrives and is kept for that use.
// It binds listeners, so it waits for the server tests: a port handed
// to a soulfind that is still starting is otherwise up for grabs.
let _gate = SERVER_GATE
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut phrases = Message::new();
phrases
.write_int32(160)
@@ -4604,7 +4665,8 @@ fn a_private_room_is_owned_granted_and_revoked() {
owner
.join_private_room(room)
.expect("owner creates the room");
std::thread::sleep(Duration::from_millis(750));
// An invite reaches only a user already accepting invitations.
fence(&guest, "e2e_priv_owner");
// The guest is invited, and hears about it (code 139).
owner
@@ -4723,7 +4785,7 @@ fn a_private_room_someone_else_owns_cannot_be_taken() {
owner
.join_private_room(room)
.expect("owner creates the room");
std::thread::sleep(Duration::from_millis(750));
fence(&owner, "e2e_taken_outsider");
// An outsider asking for the same name must be refused (code 1003), not
// handed the room.
@@ -4776,7 +4838,7 @@ fn an_acknowledged_offline_message_is_not_delivered_twice() {
sender
.send_private_message("e2e_ack_recipient", body)
.expect("send to an offline user");
std::thread::sleep(Duration::from_millis(500));
fence(&sender, "e2e_ack_recipient");
let received_once = |label: &str| -> bool {
let mut client =
@@ -4794,7 +4856,9 @@ fn an_acknowledged_offline_message_is_not_delivered_twice() {
std::thread::sleep(Duration::from_millis(100));
}
}
std::thread::sleep(Duration::from_millis(500));
// The acknowledgement goes out ahead of the message it answers, so
// once the server has handled what followed, it has the ack too.
fence(&client, "e2e_ack_sender");
got
};
@@ -5348,7 +5412,7 @@ fn a_member_can_give_up_a_private_room_and_an_owner_can_disband_it() {
owner
.join_private_room(room)
.expect("owner creates the room");
std::thread::sleep(Duration::from_millis(750));
fence(&guest, "e2e_leave_owner");
owner
.add_room_member(room, "e2e_leave_guest")
.expect("owner invites the guest");
@@ -5459,6 +5523,11 @@ fn a_disconnected_client_gives_its_listener_port_back() {
// next one has to be able to advertise the port it was configured with,
// not fall back to an ephemeral one nobody was told about. No server is
// needed — binding the listener happens in connect(), before the dial.
// It binds listeners, so it waits for the server tests: a port handed
// to a soulfind that is still starting is otherwise up for grabs.
let _gate = SERVER_GATE
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let port = free_port().expect("a port to hold");
let settings = |port: u16| ClientSettings {
username: "e2e_port_release".to_string(),
soulseek-rs-lib/tests/resume_e2e.rs
+10 -7
View File
@@ -379,18 +379,21 @@ fn a_live_download_pauses_and_resumes_on_the_same_connection() {
matches!(status, DownloadStatus::Paused { .. })
});
// Give a read already in flight one timeout cycle to observe the pause.
std::thread::sleep(Duration::from_millis(1_200));
// The pause is checked before every read, so once it lands only the read
// already in flight can finish: one 64 KiB buffer at most. Queue more
// than that, which a transfer ignoring the pause would take all of.
let stable_len = fs::metadata(&part_path).expect("paused partial").len();
let queued_while_paused = 8 * 1024;
let in_flight = 64 * 1024;
let queued_while_paused = in_flight + 8 * 1024;
peer.write_all(&content[first_chunk..first_chunk + queued_while_paused])
.expect("queue bytes while paused");
peer.flush().expect("flush bytes while paused");
std::thread::sleep(Duration::from_millis(300));
assert_eq!(
fs::metadata(&part_path).expect("paused partial").len(),
stable_len,
"a paused transfer must stop consuming its socket"
let paused_len = fs::metadata(&part_path).expect("paused partial").len();
assert!(
paused_len <= stable_len + in_flight as u64,
"a paused transfer must stop consuming its socket: {paused_len} \
bytes after pausing at {stable_len}"
);
assert!(subject.client.resume_download(&control.username, filename));
soulseek-rs/src/port_mapping/mod.rs
+25 -13
View File
@@ -418,7 +418,18 @@ mod tests {
thread::spawn(move || {
let mut buf = [0u8; 64];
// Answer many requests (initial map + cleanup unmap + any renews).
while let Ok((n, from)) = socket.recv_from(&mut buf) {
loop {
let (n, from) = match socket.recv_from(&mut buf) {
Ok(received) => received,
// Windows reports a reply that met an already-closed
// client socket on the next receive; the gateway is fine.
Err(e)
if e.kind() == std::io::ErrorKind::ConnectionReset =>
{
continue;
}
Err(_) => break,
};
let request = buf[..n].to_vec();
let internal = u16::from_be_bytes([request[4], request[5]]);
let lifetime = u32::from_be_bytes([
@@ -465,20 +476,21 @@ mod tests {
);
// Dropping the mapper must remove the mapping (a lifetime-0 request).
// A map the gateway was slow to answer is sent again, and that resend
// can arrive first.
drop(mapper);
let unmap_req = requests
.recv_timeout(StdDuration::from_secs(3))
.expect("dropping the mapper should send an unmap request");
assert_eq!(
let unmapped = std::iter::from_fn(|| {
requests.recv_timeout(StdDuration::from_secs(3)).ok()
})
.any(|request| {
u32::from_be_bytes([
unmap_req[8],
unmap_req[9],
unmap_req[10],
unmap_req[11]
]),
0,
"cleanup should request lifetime 0"
);
request[8],
request[9],
request[10],
request[11],
]) == 0
});
assert!(unmapped, "cleanup should request lifetime 0");
}
/// A minimal in-process UPnP IGD: an HTTP server that answers every SOAP
soulseek-rs/tests/cli_e2e.rs
+28 -5
View File
@@ -884,21 +884,28 @@ fn split_session_flags(args: &[&str]) -> (Vec<String>, Vec<String>) {
///
/// `--no-config` is dropped for the obvious reason, and `--quiet` with it:
/// these commands report the schedule the *server* chose on stderr, which is
/// the only place a test can see it.
/// the only place a test can see it. Session flags go to the daemon, as in
/// [`cli`].
fn cli_with_config(
server: &TestServer,
user: &str,
config: &Path,
args: &[&str],
) -> Output {
let (session_flags, command_args) = match mode() {
Mode::Local => {
(Vec::new(), args.iter().map(|a| (*a).to_string()).collect())
}
Mode::Daemon => split_session_flags(args),
};
let mut all: Vec<String> = server
.args(user)
.args_with(user, &session_flags)
.into_iter()
.filter(|arg| arg != "--no-config" && arg != "--quiet")
.collect();
all.push("--config".to_string());
all.push(config.to_str().expect("utf-8 path").to_string());
all.extend(args.iter().map(|a| (*a).to_string()));
all.extend(command_args);
let refs: Vec<&str> = all.iter().map(String::as_str).collect();
run(&refs)
}
@@ -1888,8 +1895,19 @@ fn room_listen_streams_what_is_said_in_the_room() {
.spawn()
.expect("the binary should run");
// Give the listener time to join before saying anything.
std::thread::sleep(Duration::from_secs(3));
// The room announces the listener once the server has it in; a line
// said before that never reaches it.
let deadline = Instant::now() + Duration::from_secs(30);
while !speaker.take_room_events().iter().any(|event| {
matches!(
event,
soulseek_rs::types::RoomEvent::UserJoined { room, username }
if room == "cli_e2e_lobby_b" && username == "cli_e2e_room_reader"
)
}) {
assert!(Instant::now() < deadline, "the listener never joined");
std::thread::sleep(Duration::from_millis(50));
}
speaker
.say_in_room("cli_e2e_lobby_b", "streamed line")
.expect("say");
@@ -1958,6 +1976,11 @@ fn a_private_message_reaches_its_recipient() {
fn message_read_streams_incoming_private_messages() {
let server = server_or_skip!();
let sender = server.client("cli_e2e_pm_sender", Vec::new());
// soulfind drops a message to a name it has never seen but keeps one for
// a known user until they log in, so however long the reader takes to
// get online, the line below reaches it. Registered before the reader's
// own session exists: a later login as them would take its place.
drop(server.client("cli_e2e_pm_reader", Vec::new()));
settle();
let mut args = server.args("cli_e2e_pm_reader");
soulseek-rs/tests/common/mod.rs
+11 -7
View File
@@ -60,12 +60,16 @@ pub struct Soulfind {
}
impl Soulfind {
/// soulfind exits if its port is taken between `free_port()` handing it
/// out and soulfind binding it, so a start that fails gets a fresh port.
pub fn start() -> Option<Self> {
let port = free_port()?;
let db = std::env::temp_dir().join(format!("soulfind-{port}.db"));
let _ = std::fs::remove_file(&db);
let child = Self::spawn(port, &db)?;
Some(Self { child, db, port })
(0..3).find_map(|_| {
let port = free_port()?;
let db = std::env::temp_dir().join(format!("soulfind-{port}.db"));
let _ = std::fs::remove_file(&db);
let child = Self::spawn(port, &db)?;
Some(Self { child, db, port })
})
}
fn spawn(port: u16, db: &Path) -> Option<Child> {
@@ -78,8 +82,8 @@ impl Soulfind {
let deadline = Instant::now() + Duration::from_secs(10);
while Instant::now() < deadline {
// Whoever answers on the port has to be this child: one that
// lost the port and died would otherwise pass as up.
// A child that has exited is never taken as up: soulfind exits
// when its port is gone, and whoever took it may answer instead.
if child.try_wait().ok().flatten().is_some() {
return None;
}