| 1 | use std::time::Duration; |
| 2 | |
| 3 | use macltok::{ |
| 4 | ClientHandle, Command, CommandErrorKind, CommandId, CommandKind, Engine, EngineConfig, |
| 5 | EngineError, Event, EventStream, LifecycleCommand, MessageContent, MessageEditContent, |
| 6 | RichText, RoomCommand, TimelineCommand, |
| 7 | }; |
| 8 | use matrix_sdk::{ |
| 9 | ruma::{RoomId, api::client::sync::sync_events::v5, event_id, room_id}, |
| 10 | test_utils::mocks::MatrixMockServer, |
| 11 | }; |
| 12 | use tokio::task::JoinHandle; |
| 13 | |
| 14 | #[tokio::test] |
| 15 | async fn empty_message_bodies_report_their_command_ids() { |
| 16 | let room_id = room_id!("!empty-bodies:localhost"); |
| 17 | let (server, handle, mut events, running) = start_engine(room_id).await; |
| 18 | open_room(&handle, &mut events, room_id).await; |
| 19 | |
| 20 | let commands = [ |
| 21 | TimelineCommand::Send { |
| 22 | timeline_id: room_id.to_owned().into(), |
| 23 | content: MessageContent::text(String::new()), |
| 24 | }, |
| 25 | TimelineCommand::Send { |
| 26 | timeline_id: room_id.to_owned().into(), |
| 27 | content: MessageContent::emote(RichText::plain(" "), Default::default()), |
| 28 | }, |
| 29 | TimelineCommand::Send { |
| 30 | timeline_id: room_id.to_owned().into(), |
| 31 | content: MessageContent::reply( |
| 32 | event_id!("$reply:localhost").to_owned(), |
| 33 | RichText::plain("\n"), |
| 34 | Default::default(), |
| 35 | ), |
| 36 | }, |
| 37 | TimelineCommand::Edit { |
| 38 | timeline_id: room_id.to_owned().into(), |
| 39 | event_id: event_id!("$edit:localhost").to_owned(), |
| 40 | content: MessageEditContent::text("\t"), |
| 41 | }, |
| 42 | ]; |
| 43 | |
| 44 | for command in commands { |
| 45 | let id = handle.send(Command::Timeline(command)).await.unwrap(); |
| 46 | let failure = receive_command_failure(&mut events, id).await; |
| 47 | assert_eq!(failure.id, id); |
| 48 | assert_eq!(failure.command, CommandKind::Timeline); |
| 49 | assert_eq!(failure.kind, CommandErrorKind::EmptyBody); |
| 50 | } |
| 51 | |
| 52 | stop_engine(handle, &mut events, running).await; |
| 53 | drop(server); |
| 54 | } |
| 55 | |
| 56 | #[tokio::test] |
| 57 | async fn open_resets_the_timeline_before_succeeding() { |
| 58 | let room_id = room_id!("!open-order:localhost"); |
| 59 | let (server, handle, mut events, running) = start_engine(room_id).await; |
| 60 | |
| 61 | open_room(&handle, &mut events, room_id).await; |
| 62 | |
| 63 | stop_engine(handle, &mut events, running).await; |
| 64 | drop(server); |
| 65 | } |
| 66 | |
| 67 | #[tokio::test] |
| 68 | async fn same_room_sends_complete_in_submission_order() { |
| 69 | let room_id = room_id!("!send-order:localhost"); |
| 70 | let (server, handle, mut events, running) = start_engine(room_id).await; |
| 71 | open_room(&handle, &mut events, room_id).await; |
| 72 | |
| 73 | let first_id = handle |
| 74 | .send(Command::Timeline(TimelineCommand::Send { |
| 75 | timeline_id: room_id.to_owned().into(), |
| 76 | content: MessageContent::text("first"), |
| 77 | })) |
| 78 | .await |
| 79 | .unwrap(); |
| 80 | let second_id = handle |
| 81 | .send(Command::Timeline(TimelineCommand::Send { |
| 82 | timeline_id: room_id.to_owned().into(), |
| 83 | content: MessageContent::text("second"), |
| 84 | })) |
| 85 | .await |
| 86 | .unwrap(); |
| 87 | |
| 88 | assert_eq!( |
| 89 | receive_timeline_completion(&mut events, &[first_id, second_id]).await, |
| 90 | vec![first_id, second_id] |
| 91 | ); |
| 92 | |
| 93 | stop_engine(handle, &mut events, running).await; |
| 94 | drop(server); |
| 95 | } |
| 96 | |
| 97 | async fn start_engine( |
| 98 | room_id: &RoomId, |
| 99 | ) -> ( |
| 100 | MatrixMockServer, |
| 101 | ClientHandle, |
| 102 | EventStream, |
| 103 | JoinHandle<Result<(), EngineError>>, |
| 104 | ) { |
| 105 | let server = MatrixMockServer::new().await; |
| 106 | server |
| 107 | .mock_sliding_sync() |
| 108 | .ok(v5::Response::new("1".to_owned())) |
| 109 | .mount() |
| 110 | .await; |
| 111 | let client = server.client_builder().build().await; |
| 112 | server.mock_room_state_encryption().plain().mount().await; |
| 113 | server.sync_joined_room(&client, room_id).await; |
| 114 | let config = EngineConfig::default().with_initial_history_limit(0); |
| 115 | let (engine, handle, events) = Engine::from_client(client, config).await.unwrap(); |
| 116 | let running = tokio::spawn(engine.run()); |
| 117 | (server, handle, events, running) |
| 118 | } |
| 119 | |
| 120 | async fn open_room(handle: &ClientHandle, events: &mut EventStream, room_id: &RoomId) -> CommandId { |
| 121 | let id = handle |
| 122 | .send(Command::Room(RoomCommand::Open(room_id.to_owned()))) |
| 123 | .await |
| 124 | .unwrap(); |
| 125 | let mut reset_received = false; |
| 126 | |
| 127 | tokio::time::timeout(Duration::from_secs(2), async { |
| 128 | loop { |
| 129 | match events.recv().await.expect("event stream closed") { |
| 130 | Event::TimelineReset { |
| 131 | timeline_id: reset_timeline, |
| 132 | .. |
| 133 | } if reset_timeline.room_id() == room_id => reset_received = true, |
| 134 | Event::CommandSucceeded { |
| 135 | id: succeeded_id, |
| 136 | command: CommandKind::Room, |
| 137 | } if succeeded_id == id => { |
| 138 | assert!(reset_received); |
| 139 | break; |
| 140 | } |
| 141 | Event::CommandFailed(failure) if failure.id == id => { |
| 142 | panic!("opening room failed {failure:?}") |
| 143 | } |
| 144 | _ => {} |
| 145 | } |
| 146 | } |
| 147 | }) |
| 148 | .await |
| 149 | .expect("room did not open"); |
| 150 | id |
| 151 | } |
| 152 | |
| 153 | async fn receive_command_failure( |
| 154 | events: &mut EventStream, |
| 155 | expected_id: CommandId, |
| 156 | ) -> macltok::CommandFailure { |
| 157 | tokio::time::timeout(Duration::from_secs(2), async { |
| 158 | loop { |
| 159 | match events.recv().await.expect("event stream closed") { |
| 160 | Event::CommandFailed(failure) if failure.id == expected_id => return failure, |
| 161 | Event::CommandSucceeded { id, .. } if id == expected_id => { |
| 162 | panic!("command unexpectedly succeeded") |
| 163 | } |
| 164 | _ => {} |
| 165 | } |
| 166 | } |
| 167 | }) |
| 168 | .await |
| 169 | .expect("command failure did not arrive") |
| 170 | } |
| 171 | |
| 172 | async fn receive_timeline_completion( |
| 173 | events: &mut EventStream, |
| 174 | expected_ids: &[CommandId], |
| 175 | ) -> Vec<CommandId> { |
| 176 | tokio::time::timeout(Duration::from_secs(2), async { |
| 177 | let mut completed = Vec::new(); |
| 178 | while completed.len() < expected_ids.len() { |
| 179 | match events.recv().await.expect("event stream closed") { |
| 180 | Event::CommandSucceeded { |
| 181 | id, |
| 182 | command: CommandKind::Timeline, |
| 183 | } if expected_ids.contains(&id) => completed.push(id), |
| 184 | Event::CommandFailed(failure) if expected_ids.contains(&failure.id) => { |
| 185 | panic!("timeline command failed {failure:?}") |
| 186 | } |
| 187 | _ => {} |
| 188 | } |
| 189 | } |
| 190 | completed |
| 191 | }) |
| 192 | .await |
| 193 | .expect("timeline commands did not complete") |
| 194 | } |
| 195 | |
| 196 | async fn stop_engine( |
| 197 | handle: ClientHandle, |
| 198 | events: &mut EventStream, |
| 199 | running: JoinHandle<Result<(), EngineError>>, |
| 200 | ) { |
| 201 | let id = handle |
| 202 | .send(Command::Lifecycle(LifecycleCommand::Shutdown)) |
| 203 | .await |
| 204 | .unwrap(); |
| 205 | tokio::time::timeout(Duration::from_secs(2), async { |
| 206 | let mut succeeded = false; |
| 207 | while let Some(event) = events.recv().await { |
| 208 | match event { |
| 209 | Event::CommandSucceeded { |
| 210 | id: succeeded_id, |
| 211 | command: CommandKind::Lifecycle, |
| 212 | } if succeeded_id == id => succeeded = true, |
| 213 | Event::Stopped => { |
| 214 | assert!(succeeded); |
| 215 | break; |
| 216 | } |
| 217 | _ => {} |
| 218 | } |
| 219 | } |
| 220 | }) |
| 221 | .await |
| 222 | .expect("engine did not stop"); |
| 223 | running.await.unwrap().unwrap(); |
| 224 | } |