1use crate::{
  2    db::{NewUserParams, TestDb, UserId},
  3    executor::Executor,
  4    rpc::{Server, CLEANUP_TIMEOUT},
  5    AppState,
  6};
  7use anyhow::anyhow;
  8use call::{ActiveCall, Room};
  9use client::{
 10    self, proto::PeerId, ChannelStore, Client, Connection, Credentials, EstablishConnectionError,
 11    UserStore,
 12};
 13use collections::{HashMap, HashSet};
 14use fs::FakeFs;
 15use futures::{channel::oneshot, StreamExt as _};
 16use gpui::{
 17    elements::*, executor::Deterministic, AnyElement, Entity, ModelHandle, Task, TestAppContext,
 18    View, ViewContext, ViewHandle, WeakViewHandle,
 19};
 20use language::LanguageRegistry;
 21use parking_lot::Mutex;
 22use project::{Project, WorktreeId};
 23use settings::SettingsStore;
 24use std::{
 25    cell::{Ref, RefCell, RefMut},
 26    env,
 27    ops::{Deref, DerefMut},
 28    path::Path,
 29    sync::{
 30        atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
 31        Arc,
 32    },
 33};
 34use util::http::FakeHttpClient;
 35use workspace::Workspace;
 36
 37mod channel_tests;
 38mod integration_tests;
 39mod randomized_integration_tests;
 40
 41struct TestServer {
 42    app_state: Arc<AppState>,
 43    server: Arc<Server>,
 44    connection_killers: Arc<Mutex<HashMap<PeerId, Arc<AtomicBool>>>>,
 45    forbid_connections: Arc<AtomicBool>,
 46    _test_db: TestDb,
 47    test_live_kit_server: Arc<live_kit_client::TestServer>,
 48}
 49
 50impl TestServer {
 51    async fn start(deterministic: &Arc<Deterministic>) -> Self {
 52        static NEXT_LIVE_KIT_SERVER_ID: AtomicUsize = AtomicUsize::new(0);
 53
 54        let use_postgres = env::var("USE_POSTGRES").ok();
 55        let use_postgres = use_postgres.as_deref();
 56        let test_db = if use_postgres == Some("true") || use_postgres == Some("1") {
 57            TestDb::postgres(deterministic.build_background())
 58        } else {
 59            TestDb::sqlite(deterministic.build_background())
 60        };
 61        let live_kit_server_id = NEXT_LIVE_KIT_SERVER_ID.fetch_add(1, SeqCst);
 62        let live_kit_server = live_kit_client::TestServer::create(
 63            format!("http://livekit.{}.test", live_kit_server_id),
 64            format!("devkey-{}", live_kit_server_id),
 65            format!("secret-{}", live_kit_server_id),
 66            deterministic.build_background(),
 67        )
 68        .unwrap();
 69        let app_state = Self::build_app_state(&test_db, &live_kit_server).await;
 70        let epoch = app_state
 71            .db
 72            .create_server(&app_state.config.zed_environment)
 73            .await
 74            .unwrap();
 75        let server = Server::new(
 76            epoch,
 77            app_state.clone(),
 78            Executor::Deterministic(deterministic.build_background()),
 79        );
 80        server.start().await.unwrap();
 81        // Advance clock to ensure the server's cleanup task is finished.
 82        deterministic.advance_clock(CLEANUP_TIMEOUT);
 83        Self {
 84            app_state,
 85            server,
 86            connection_killers: Default::default(),
 87            forbid_connections: Default::default(),
 88            _test_db: test_db,
 89            test_live_kit_server: live_kit_server,
 90        }
 91    }
 92
 93    async fn reset(&self) {
 94        self.app_state.db.reset();
 95        let epoch = self
 96            .app_state
 97            .db
 98            .create_server(&self.app_state.config.zed_environment)
 99            .await
100            .unwrap();
101        self.server.reset(epoch);
102    }
103
104    async fn create_client(&mut self, cx: &mut TestAppContext, name: &str) -> TestClient {
105        cx.update(|cx| {
106            if cx.has_global::<SettingsStore>() {
107                panic!("Same cx used to create two test clients")
108            }
109            cx.set_global(SettingsStore::test(cx));
110        });
111
112        let http = FakeHttpClient::with_404_response();
113        let user_id = if let Ok(Some(user)) = self.app_state.db.get_user_by_github_login(name).await
114        {
115            user.id
116        } else {
117            self.app_state
118                .db
119                .create_user(
120                    &format!("{name}@example.com"),
121                    false,
122                    NewUserParams {
123                        github_login: name.into(),
124                        github_user_id: 0,
125                        invite_count: 0,
126                    },
127                )
128                .await
129                .expect("creating user failed")
130                .user_id
131        };
132        let client_name = name.to_string();
133        let mut client = cx.read(|cx| Client::new(http.clone(), cx));
134        let server = self.server.clone();
135        let db = self.app_state.db.clone();
136        let connection_killers = self.connection_killers.clone();
137        let forbid_connections = self.forbid_connections.clone();
138
139        Arc::get_mut(&mut client)
140            .unwrap()
141            .set_id(user_id.0 as usize)
142            .override_authenticate(move |cx| {
143                cx.spawn(|_| async move {
144                    let access_token = "the-token".to_string();
145                    Ok(Credentials {
146                        user_id: user_id.0 as u64,
147                        access_token,
148                    })
149                })
150            })
151            .override_establish_connection(move |credentials, cx| {
152                assert_eq!(credentials.user_id, user_id.0 as u64);
153                assert_eq!(credentials.access_token, "the-token");
154
155                let server = server.clone();
156                let db = db.clone();
157                let connection_killers = connection_killers.clone();
158                let forbid_connections = forbid_connections.clone();
159                let client_name = client_name.clone();
160                cx.spawn(move |cx| async move {
161                    if forbid_connections.load(SeqCst) {
162                        Err(EstablishConnectionError::other(anyhow!(
163                            "server is forbidding connections"
164                        )))
165                    } else {
166                        let (client_conn, server_conn, killed) =
167                            Connection::in_memory(cx.background());
168                        let (connection_id_tx, connection_id_rx) = oneshot::channel();
169                        let user = db
170                            .get_user_by_id(user_id)
171                            .await
172                            .expect("retrieving user failed")
173                            .unwrap();
174                        cx.background()
175                            .spawn(server.handle_connection(
176                                server_conn,
177                                client_name,
178                                user,
179                                Some(connection_id_tx),
180                                Executor::Deterministic(cx.background()),
181                            ))
182                            .detach();
183                        let connection_id = connection_id_rx.await.unwrap();
184                        connection_killers
185                            .lock()
186                            .insert(connection_id.into(), killed);
187                        Ok(client_conn)
188                    }
189                })
190            });
191
192        let fs = FakeFs::new(cx.background());
193        let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http, cx));
194        let channel_store =
195            cx.add_model(|cx| ChannelStore::new(client.clone(), user_store.clone(), cx));
196        let app_state = Arc::new(workspace::AppState {
197            client: client.clone(),
198            user_store: user_store.clone(),
199            channel_store: channel_store.clone(),
200            languages: Arc::new(LanguageRegistry::test()),
201            fs: fs.clone(),
202            build_window_options: |_, _, _| Default::default(),
203            initialize_workspace: |_, _, _, _| Task::ready(Ok(())),
204            background_actions: || &[],
205        });
206
207        cx.update(|cx| {
208            theme::init((), cx);
209            Project::init(&client, cx);
210            client::init(&client, cx);
211            language::init(cx);
212            editor::init_settings(cx);
213            workspace::init(app_state.clone(), cx);
214            audio::init((), cx);
215            call::init(client.clone(), user_store.clone(), cx);
216        });
217
218        client
219            .authenticate_and_connect(false, &cx.to_async())
220            .await
221            .unwrap();
222
223        let client = TestClient {
224            app_state,
225            username: name.to_string(),
226            state: Default::default(),
227        };
228        client.wait_for_current_user(cx).await;
229        client
230    }
231
232    fn disconnect_client(&self, peer_id: PeerId) {
233        self.connection_killers
234            .lock()
235            .remove(&peer_id)
236            .unwrap()
237            .store(true, SeqCst);
238    }
239
240    fn forbid_connections(&self) {
241        self.forbid_connections.store(true, SeqCst);
242    }
243
244    fn allow_connections(&self) {
245        self.forbid_connections.store(false, SeqCst);
246    }
247
248    async fn make_contacts(&self, clients: &mut [(&TestClient, &mut TestAppContext)]) {
249        for ix in 1..clients.len() {
250            let (left, right) = clients.split_at_mut(ix);
251            let (client_a, cx_a) = left.last_mut().unwrap();
252            for (client_b, cx_b) in right {
253                client_a
254                    .app_state
255                    .user_store
256                    .update(*cx_a, |store, cx| {
257                        store.request_contact(client_b.user_id().unwrap(), cx)
258                    })
259                    .await
260                    .unwrap();
261                cx_a.foreground().run_until_parked();
262                client_b
263                    .app_state
264                    .user_store
265                    .update(*cx_b, |store, cx| {
266                        store.respond_to_contact_request(client_a.user_id().unwrap(), true, cx)
267                    })
268                    .await
269                    .unwrap();
270            }
271        }
272    }
273
274    async fn make_channel(
275        &self,
276        channel: &str,
277        admin: (&TestClient, &mut TestAppContext),
278        members: &mut [(&TestClient, &mut TestAppContext)],
279    ) -> u64 {
280        let (admin_client, admin_cx) = admin;
281        let channel_id = admin_client
282            .app_state
283            .channel_store
284            .update(admin_cx, |channel_store, _| {
285                channel_store.create_channel(channel, None)
286            })
287            .await
288            .unwrap();
289
290        for (member_client, member_cx) in members {
291            admin_client
292                .app_state
293                .channel_store
294                .update(admin_cx, |channel_store, _| {
295                    channel_store.invite_member(channel_id, member_client.user_id().unwrap(), false)
296                })
297                .await
298                .unwrap();
299
300            admin_cx.foreground().run_until_parked();
301
302            member_client
303                .app_state
304                .channel_store
305                .update(*member_cx, |channels, _| {
306                    channels.respond_to_channel_invite(channel_id, true)
307                })
308                .await
309                .unwrap();
310        }
311
312        channel_id
313    }
314
315    async fn create_room(&self, clients: &mut [(&TestClient, &mut TestAppContext)]) {
316        self.make_contacts(clients).await;
317
318        let (left, right) = clients.split_at_mut(1);
319        let (_client_a, cx_a) = &mut left[0];
320        let active_call_a = cx_a.read(ActiveCall::global);
321
322        for (client_b, cx_b) in right {
323            let user_id_b = client_b.current_user_id(*cx_b).to_proto();
324            active_call_a
325                .update(*cx_a, |call, cx| call.invite(user_id_b, None, cx))
326                .await
327                .unwrap();
328
329            cx_b.foreground().run_until_parked();
330            let active_call_b = cx_b.read(ActiveCall::global);
331            active_call_b
332                .update(*cx_b, |call, cx| call.accept_incoming(cx))
333                .await
334                .unwrap();
335        }
336    }
337
338    async fn build_app_state(
339        test_db: &TestDb,
340        fake_server: &live_kit_client::TestServer,
341    ) -> Arc<AppState> {
342        Arc::new(AppState {
343            db: test_db.db().clone(),
344            live_kit_client: Some(Arc::new(fake_server.create_api_client())),
345            config: Default::default(),
346        })
347    }
348}
349
350impl Deref for TestServer {
351    type Target = Server;
352
353    fn deref(&self) -> &Self::Target {
354        &self.server
355    }
356}
357
358impl Drop for TestServer {
359    fn drop(&mut self) {
360        self.server.teardown();
361        self.test_live_kit_server.teardown().unwrap();
362    }
363}
364
365struct TestClient {
366    username: String,
367    state: RefCell<TestClientState>,
368    app_state: Arc<workspace::AppState>,
369}
370
371#[derive(Default)]
372struct TestClientState {
373    local_projects: Vec<ModelHandle<Project>>,
374    remote_projects: Vec<ModelHandle<Project>>,
375    buffers: HashMap<ModelHandle<Project>, HashSet<ModelHandle<language::Buffer>>>,
376}
377
378impl Deref for TestClient {
379    type Target = Arc<Client>;
380
381    fn deref(&self) -> &Self::Target {
382        &self.app_state.client
383    }
384}
385
386struct ContactsSummary {
387    pub current: Vec<String>,
388    pub outgoing_requests: Vec<String>,
389    pub incoming_requests: Vec<String>,
390}
391
392impl TestClient {
393    pub fn fs(&self) -> &FakeFs {
394        self.app_state.fs.as_fake()
395    }
396
397    pub fn channel_store(&self) -> &ModelHandle<ChannelStore> {
398        &self.app_state.channel_store
399    }
400
401    pub fn user_store(&self) -> &ModelHandle<UserStore> {
402        &self.app_state.user_store
403    }
404
405    pub fn language_registry(&self) -> &Arc<LanguageRegistry> {
406        &self.app_state.languages
407    }
408
409    pub fn client(&self) -> &Arc<Client> {
410        &self.app_state.client
411    }
412
413    pub fn current_user_id(&self, cx: &TestAppContext) -> UserId {
414        UserId::from_proto(
415            self.app_state
416                .user_store
417                .read_with(cx, |user_store, _| user_store.current_user().unwrap().id),
418        )
419    }
420
421    async fn wait_for_current_user(&self, cx: &TestAppContext) {
422        let mut authed_user = self
423            .app_state
424            .user_store
425            .read_with(cx, |user_store, _| user_store.watch_current_user());
426        while authed_user.next().await.unwrap().is_none() {}
427    }
428
429    async fn clear_contacts(&self, cx: &mut TestAppContext) {
430        self.app_state
431            .user_store
432            .update(cx, |store, _| store.clear_contacts())
433            .await;
434    }
435
436    fn local_projects<'a>(&'a self) -> impl Deref<Target = Vec<ModelHandle<Project>>> + 'a {
437        Ref::map(self.state.borrow(), |state| &state.local_projects)
438    }
439
440    fn remote_projects<'a>(&'a self) -> impl Deref<Target = Vec<ModelHandle<Project>>> + 'a {
441        Ref::map(self.state.borrow(), |state| &state.remote_projects)
442    }
443
444    fn local_projects_mut<'a>(&'a self) -> impl DerefMut<Target = Vec<ModelHandle<Project>>> + 'a {
445        RefMut::map(self.state.borrow_mut(), |state| &mut state.local_projects)
446    }
447
448    fn remote_projects_mut<'a>(&'a self) -> impl DerefMut<Target = Vec<ModelHandle<Project>>> + 'a {
449        RefMut::map(self.state.borrow_mut(), |state| &mut state.remote_projects)
450    }
451
452    fn buffers_for_project<'a>(
453        &'a self,
454        project: &ModelHandle<Project>,
455    ) -> impl DerefMut<Target = HashSet<ModelHandle<language::Buffer>>> + 'a {
456        RefMut::map(self.state.borrow_mut(), |state| {
457            state.buffers.entry(project.clone()).or_default()
458        })
459    }
460
461    fn buffers<'a>(
462        &'a self,
463    ) -> impl DerefMut<Target = HashMap<ModelHandle<Project>, HashSet<ModelHandle<language::Buffer>>>> + 'a
464    {
465        RefMut::map(self.state.borrow_mut(), |state| &mut state.buffers)
466    }
467
468    fn summarize_contacts(&self, cx: &TestAppContext) -> ContactsSummary {
469        self.app_state
470            .user_store
471            .read_with(cx, |store, _| ContactsSummary {
472                current: store
473                    .contacts()
474                    .iter()
475                    .map(|contact| contact.user.github_login.clone())
476                    .collect(),
477                outgoing_requests: store
478                    .outgoing_contact_requests()
479                    .iter()
480                    .map(|user| user.github_login.clone())
481                    .collect(),
482                incoming_requests: store
483                    .incoming_contact_requests()
484                    .iter()
485                    .map(|user| user.github_login.clone())
486                    .collect(),
487            })
488    }
489
490    async fn build_local_project(
491        &self,
492        root_path: impl AsRef<Path>,
493        cx: &mut TestAppContext,
494    ) -> (ModelHandle<Project>, WorktreeId) {
495        let project = cx.update(|cx| {
496            Project::local(
497                self.client().clone(),
498                self.app_state.user_store.clone(),
499                self.app_state.languages.clone(),
500                self.app_state.fs.clone(),
501                cx,
502            )
503        });
504        let (worktree, _) = project
505            .update(cx, |p, cx| {
506                p.find_or_create_local_worktree(root_path, true, cx)
507            })
508            .await
509            .unwrap();
510        worktree
511            .read_with(cx, |tree, _| tree.as_local().unwrap().scan_complete())
512            .await;
513        (project, worktree.read_with(cx, |tree, _| tree.id()))
514    }
515
516    async fn build_remote_project(
517        &self,
518        host_project_id: u64,
519        guest_cx: &mut TestAppContext,
520    ) -> ModelHandle<Project> {
521        let active_call = guest_cx.read(ActiveCall::global);
522        let room = active_call.read_with(guest_cx, |call, _| call.room().unwrap().clone());
523        room.update(guest_cx, |room, cx| {
524            room.join_project(
525                host_project_id,
526                self.app_state.languages.clone(),
527                self.app_state.fs.clone(),
528                cx,
529            )
530        })
531        .await
532        .unwrap()
533    }
534
535    fn build_workspace(
536        &self,
537        project: &ModelHandle<Project>,
538        cx: &mut TestAppContext,
539    ) -> ViewHandle<Workspace> {
540        struct WorkspaceContainer {
541            workspace: Option<WeakViewHandle<Workspace>>,
542        }
543
544        impl Entity for WorkspaceContainer {
545            type Event = ();
546        }
547
548        impl View for WorkspaceContainer {
549            fn ui_name() -> &'static str {
550                "WorkspaceContainer"
551            }
552
553            fn render(&mut self, cx: &mut ViewContext<Self>) -> AnyElement<Self> {
554                if let Some(workspace) = self
555                    .workspace
556                    .as_ref()
557                    .and_then(|workspace| workspace.upgrade(cx))
558                {
559                    ChildView::new(&workspace, cx).into_any()
560                } else {
561                    Empty::new().into_any()
562                }
563            }
564        }
565
566        // We use a workspace container so that we don't need to remove the window in order to
567        // drop the workspace and we can use a ViewHandle instead.
568        let (window_id, container) = cx.add_window(|_| WorkspaceContainer { workspace: None });
569        let workspace = cx.add_view(window_id, |cx| {
570            Workspace::new(0, project.clone(), self.app_state.clone(), cx)
571        });
572        container.update(cx, |container, cx| {
573            container.workspace = Some(workspace.downgrade());
574            cx.notify();
575        });
576        workspace
577    }
578}
579
580impl Drop for TestClient {
581    fn drop(&mut self) {
582        self.app_state.client.teardown();
583    }
584}
585
586#[derive(Debug, Eq, PartialEq)]
587struct RoomParticipants {
588    remote: Vec<String>,
589    pending: Vec<String>,
590}
591
592fn room_participants(room: &ModelHandle<Room>, cx: &mut TestAppContext) -> RoomParticipants {
593    room.read_with(cx, |room, _| {
594        let mut remote = room
595            .remote_participants()
596            .iter()
597            .map(|(_, participant)| participant.user.github_login.clone())
598            .collect::<Vec<_>>();
599        let mut pending = room
600            .pending_participants()
601            .iter()
602            .map(|user| user.github_login.clone())
603            .collect::<Vec<_>>();
604        remote.sort();
605        pending.sort();
606        RoomParticipants { remote, pending }
607    })
608}