test_server.rs

  1use crate::{
  2    auth::split_dev_server_token,
  3    db::{tests::TestDb, NewUserParams, UserId},
  4    executor::Executor,
  5    rpc::{Principal, Server, ZedVersion, CLEANUP_TIMEOUT, RECONNECT_TIMEOUT},
  6    AppState, Config, RateLimiter,
  7};
  8use anyhow::anyhow;
  9use call::ActiveCall;
 10use channel::{ChannelBuffer, ChannelStore};
 11use client::{
 12    self, proto::PeerId, ChannelId, Client, Connection, Credentials, EstablishConnectionError,
 13    UserStore,
 14};
 15use clock::FakeSystemClock;
 16use collab_ui::channel_view::ChannelView;
 17use collections::{HashMap, HashSet};
 18use fs::FakeFs;
 19use futures::{channel::oneshot, StreamExt as _};
 20use gpui::{BackgroundExecutor, Context, Model, Task, TestAppContext, View, VisualTestContext};
 21use language::LanguageRegistry;
 22use node_runtime::FakeNodeRuntime;
 23use notifications::NotificationStore;
 24use parking_lot::Mutex;
 25use project::{Project, WorktreeId};
 26use rpc::{
 27    proto::{self, ChannelRole},
 28    RECEIVE_TIMEOUT,
 29};
 30use semantic_version::SemanticVersion;
 31use serde_json::json;
 32use settings::SettingsStore;
 33use std::{
 34    cell::{Ref, RefCell, RefMut},
 35    env,
 36    ops::{Deref, DerefMut},
 37    path::Path,
 38    sync::{
 39        atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
 40        Arc,
 41    },
 42};
 43use util::http::FakeHttpClient;
 44use workspace::{Workspace, WorkspaceId, WorkspaceStore};
 45
 46pub struct TestServer {
 47    pub app_state: Arc<AppState>,
 48    pub test_live_kit_server: Arc<live_kit_client::TestServer>,
 49    server: Arc<Server>,
 50    next_github_user_id: i32,
 51    connection_killers: Arc<Mutex<HashMap<PeerId, Arc<AtomicBool>>>>,
 52    forbid_connections: Arc<AtomicBool>,
 53    _test_db: TestDb,
 54}
 55
 56pub struct TestClient {
 57    pub username: String,
 58    pub app_state: Arc<workspace::AppState>,
 59    channel_store: Model<ChannelStore>,
 60    notification_store: Model<NotificationStore>,
 61    state: RefCell<TestClientState>,
 62}
 63
 64#[derive(Default)]
 65struct TestClientState {
 66    local_projects: Vec<Model<Project>>,
 67    remote_projects: Vec<Model<Project>>,
 68    buffers: HashMap<Model<Project>, HashSet<Model<language::Buffer>>>,
 69    channel_buffers: HashSet<Model<ChannelBuffer>>,
 70}
 71
 72pub struct ContactsSummary {
 73    pub current: Vec<String>,
 74    pub outgoing_requests: Vec<String>,
 75    pub incoming_requests: Vec<String>,
 76}
 77
 78impl TestServer {
 79    pub async fn start(deterministic: BackgroundExecutor) -> Self {
 80        static NEXT_LIVE_KIT_SERVER_ID: AtomicUsize = AtomicUsize::new(0);
 81
 82        let use_postgres = env::var("USE_POSTGRES").ok();
 83        let use_postgres = use_postgres.as_deref();
 84        let test_db = if use_postgres == Some("true") || use_postgres == Some("1") {
 85            TestDb::postgres(deterministic.clone())
 86        } else {
 87            TestDb::sqlite(deterministic.clone())
 88        };
 89        let live_kit_server_id = NEXT_LIVE_KIT_SERVER_ID.fetch_add(1, SeqCst);
 90        let live_kit_server = live_kit_client::TestServer::create(
 91            format!("http://livekit.{}.test", live_kit_server_id),
 92            format!("devkey-{}", live_kit_server_id),
 93            format!("secret-{}", live_kit_server_id),
 94            deterministic.clone(),
 95        )
 96        .unwrap();
 97        let executor = Executor::Deterministic(deterministic.clone());
 98        let app_state = Self::build_app_state(&test_db, &live_kit_server, executor.clone()).await;
 99        let epoch = app_state
100            .db
101            .create_server(&app_state.config.zed_environment)
102            .await
103            .unwrap();
104        let server = Server::new(epoch, app_state.clone());
105        server.start().await.unwrap();
106        // Advance clock to ensure the server's cleanup task is finished.
107        deterministic.advance_clock(CLEANUP_TIMEOUT);
108        Self {
109            app_state,
110            server,
111            connection_killers: Default::default(),
112            forbid_connections: Default::default(),
113            next_github_user_id: 0,
114            _test_db: test_db,
115            test_live_kit_server: live_kit_server,
116        }
117    }
118
119    pub async fn start2(
120        cx_a: &mut TestAppContext,
121        cx_b: &mut TestAppContext,
122    ) -> (TestServer, TestClient, TestClient, ChannelId) {
123        let mut server = Self::start(cx_a.executor()).await;
124        let client_a = server.create_client(cx_a, "user_a").await;
125        let client_b = server.create_client(cx_b, "user_b").await;
126        let channel_id = server
127            .make_channel(
128                "test-channel",
129                None,
130                (&client_a, cx_a),
131                &mut [(&client_b, cx_b)],
132            )
133            .await;
134        cx_a.run_until_parked();
135
136        (server, client_a, client_b, channel_id)
137    }
138
139    pub async fn start1(cx: &mut TestAppContext) -> (TestServer, TestClient) {
140        let mut server = Self::start(cx.executor().clone()).await;
141        let client = server.create_client(cx, "user_a").await;
142        (server, client)
143    }
144
145    pub async fn reset(&self) {
146        self.app_state.db.reset();
147        let epoch = self
148            .app_state
149            .db
150            .create_server(&self.app_state.config.zed_environment)
151            .await
152            .unwrap();
153        self.server.reset(epoch);
154    }
155
156    pub async fn create_client(&mut self, cx: &mut TestAppContext, name: &str) -> TestClient {
157        cx.update(|cx| {
158            if cx.has_global::<SettingsStore>() {
159                panic!("Same cx used to create two test clients")
160            }
161            let settings = SettingsStore::test(cx);
162            cx.set_global(settings);
163            release_channel::init("0.0.0", cx);
164            client::init_settings(cx);
165        });
166
167        let clock = Arc::new(FakeSystemClock::default());
168        let http = FakeHttpClient::with_404_response();
169        let user_id = if let Ok(Some(user)) = self.app_state.db.get_user_by_github_login(name).await
170        {
171            user.id
172        } else {
173            let github_user_id = self.next_github_user_id;
174            self.next_github_user_id += 1;
175            self.app_state
176                .db
177                .create_user(
178                    &format!("{name}@example.com"),
179                    false,
180                    NewUserParams {
181                        github_login: name.into(),
182                        github_user_id,
183                    },
184                )
185                .await
186                .expect("creating user failed")
187                .user_id
188        };
189        let client_name = name.to_string();
190        let mut client = cx.update(|cx| Client::new(clock, http.clone(), cx));
191        let server = self.server.clone();
192        let db = self.app_state.db.clone();
193        let connection_killers = self.connection_killers.clone();
194        let forbid_connections = self.forbid_connections.clone();
195
196        Arc::get_mut(&mut client)
197            .unwrap()
198            .set_id(user_id.to_proto())
199            .override_authenticate(move |cx| {
200                cx.spawn(|_| async move {
201                    let access_token = "the-token".to_string();
202                    Ok(Credentials::User {
203                        user_id: user_id.to_proto(),
204                        access_token,
205                    })
206                })
207            })
208            .override_establish_connection(move |credentials, cx| {
209                assert_eq!(
210                    credentials,
211                    &Credentials::User {
212                        user_id: user_id.0 as u64,
213                        access_token: "the-token".into()
214                    }
215                );
216
217                let server = server.clone();
218                let db = db.clone();
219                let connection_killers = connection_killers.clone();
220                let forbid_connections = forbid_connections.clone();
221                let client_name = client_name.clone();
222                cx.spawn(move |cx| async move {
223                    if forbid_connections.load(SeqCst) {
224                        Err(EstablishConnectionError::other(anyhow!(
225                            "server is forbidding connections"
226                        )))
227                    } else {
228                        let (client_conn, server_conn, killed) =
229                            Connection::in_memory(cx.background_executor().clone());
230                        let (connection_id_tx, connection_id_rx) = oneshot::channel();
231                        let user = db
232                            .get_user_by_id(user_id)
233                            .await
234                            .expect("retrieving user failed")
235                            .unwrap();
236                        cx.background_executor()
237                            .spawn(server.handle_connection(
238                                server_conn,
239                                client_name,
240                                Principal::User(user),
241                                ZedVersion(SemanticVersion::new(1, 0, 0)),
242                                Some(connection_id_tx),
243                                Executor::Deterministic(cx.background_executor().clone()),
244                            ))
245                            .detach();
246                        let connection_id = connection_id_rx.await.map_err(|e| {
247                            EstablishConnectionError::Other(anyhow!(
248                                "{} (is server shutting down?)",
249                                e
250                            ))
251                        })?;
252                        connection_killers
253                            .lock()
254                            .insert(connection_id.into(), killed);
255                        Ok(client_conn)
256                    }
257                })
258            });
259
260        let fs = FakeFs::new(cx.executor());
261        let user_store = cx.new_model(|cx| UserStore::new(client.clone(), cx));
262        let workspace_store = cx.new_model(|cx| WorkspaceStore::new(client.clone(), cx));
263        let language_registry = Arc::new(LanguageRegistry::test(cx.executor()));
264        let app_state = Arc::new(workspace::AppState {
265            client: client.clone(),
266            user_store: user_store.clone(),
267            workspace_store,
268            languages: language_registry,
269            fs: fs.clone(),
270            build_window_options: |_, _| Default::default(),
271            node_runtime: FakeNodeRuntime::new(),
272        });
273
274        cx.update(|cx| {
275            theme::init(theme::LoadThemes::JustBase, cx);
276            Project::init(&client, cx);
277            client::init(&client, cx);
278            language::init(cx);
279            editor::init(cx);
280            workspace::init(app_state.clone(), cx);
281            call::init(client.clone(), user_store.clone(), cx);
282            channel::init(&client, user_store.clone(), cx);
283            notifications::init(client.clone(), user_store, cx);
284            collab_ui::init(&app_state, cx);
285            file_finder::init(cx);
286            menu::init();
287            remote_projects::init(client.clone(), cx);
288            settings::KeymapFile::load_asset("keymaps/default-macos.json", cx).unwrap();
289        });
290
291        client
292            .authenticate_and_connect(false, &cx.to_async())
293            .await
294            .unwrap();
295
296        let client = TestClient {
297            app_state,
298            username: name.to_string(),
299            channel_store: cx.read(ChannelStore::global).clone(),
300            notification_store: cx.read(NotificationStore::global).clone(),
301            state: Default::default(),
302        };
303        client.wait_for_current_user(cx).await;
304        client
305    }
306
307    pub async fn create_dev_server(
308        &self,
309        access_token: String,
310        cx: &mut TestAppContext,
311    ) -> TestClient {
312        cx.update(|cx| {
313            if cx.has_global::<SettingsStore>() {
314                panic!("Same cx used to create two test clients")
315            }
316            let settings = SettingsStore::test(cx);
317            cx.set_global(settings);
318            release_channel::init("0.0.0", cx);
319            client::init_settings(cx);
320        });
321        let (dev_server_id, _) = split_dev_server_token(&access_token).unwrap();
322
323        let clock = Arc::new(FakeSystemClock::default());
324        let http = FakeHttpClient::with_404_response();
325        let mut client = cx.update(|cx| Client::new(clock, http.clone(), cx));
326        let server = self.server.clone();
327        let db = self.app_state.db.clone();
328        let connection_killers = self.connection_killers.clone();
329        let forbid_connections = self.forbid_connections.clone();
330        Arc::get_mut(&mut client)
331            .unwrap()
332            .set_id(1)
333            .set_dev_server_token(client::DevServerToken(access_token.clone()))
334            .override_establish_connection(move |credentials, cx| {
335                assert_eq!(
336                    credentials,
337                    &Credentials::DevServer {
338                        token: client::DevServerToken(access_token.to_string())
339                    }
340                );
341
342                let server = server.clone();
343                let db = db.clone();
344                let connection_killers = connection_killers.clone();
345                let forbid_connections = forbid_connections.clone();
346                cx.spawn(move |cx| async move {
347                    if forbid_connections.load(SeqCst) {
348                        Err(EstablishConnectionError::other(anyhow!(
349                            "server is forbidding connections"
350                        )))
351                    } else {
352                        let (client_conn, server_conn, killed) =
353                            Connection::in_memory(cx.background_executor().clone());
354                        let (connection_id_tx, connection_id_rx) = oneshot::channel();
355                        let dev_server = db
356                            .get_dev_server(dev_server_id)
357                            .await
358                            .expect("retrieving dev_server failed");
359                        cx.background_executor()
360                            .spawn(server.handle_connection(
361                                server_conn,
362                                "dev-server".to_string(),
363                                Principal::DevServer(dev_server),
364                                ZedVersion(SemanticVersion::new(1, 0, 0)),
365                                Some(connection_id_tx),
366                                Executor::Deterministic(cx.background_executor().clone()),
367                            ))
368                            .detach();
369                        let connection_id = connection_id_rx.await.map_err(|e| {
370                            EstablishConnectionError::Other(anyhow!(
371                                "{} (is server shutting down?)",
372                                e
373                            ))
374                        })?;
375                        connection_killers
376                            .lock()
377                            .insert(connection_id.into(), killed);
378                        Ok(client_conn)
379                    }
380                })
381            });
382
383        let fs = FakeFs::new(cx.executor());
384        let user_store = cx.new_model(|cx| UserStore::new(client.clone(), cx));
385        let workspace_store = cx.new_model(|cx| WorkspaceStore::new(client.clone(), cx));
386        let language_registry = Arc::new(LanguageRegistry::test(cx.executor()));
387        let app_state = Arc::new(workspace::AppState {
388            client: client.clone(),
389            user_store: user_store.clone(),
390            workspace_store,
391            languages: language_registry,
392            fs: fs.clone(),
393            build_window_options: |_, _| Default::default(),
394            node_runtime: FakeNodeRuntime::new(),
395        });
396
397        cx.update(|cx| {
398            theme::init(theme::LoadThemes::JustBase, cx);
399            Project::init(&client, cx);
400            client::init(&client, cx);
401            language::init(cx);
402            editor::init(cx);
403            workspace::init(app_state.clone(), cx);
404            call::init(client.clone(), user_store.clone(), cx);
405            channel::init(&client, user_store.clone(), cx);
406            notifications::init(client.clone(), user_store, cx);
407            collab_ui::init(&app_state, cx);
408            file_finder::init(cx);
409            menu::init();
410            headless::init(
411                client.clone(),
412                headless::AppState {
413                    languages: app_state.languages.clone(),
414                    user_store: app_state.user_store.clone(),
415                    fs: fs.clone(),
416                    node_runtime: app_state.node_runtime.clone(),
417                },
418                cx,
419            );
420        });
421
422        TestClient {
423            app_state,
424            username: "dev-server".to_string(),
425            channel_store: cx.read(ChannelStore::global).clone(),
426            notification_store: cx.read(NotificationStore::global).clone(),
427            state: Default::default(),
428        }
429    }
430
431    pub fn disconnect_client(&self, peer_id: PeerId) {
432        self.connection_killers
433            .lock()
434            .remove(&peer_id)
435            .unwrap()
436            .store(true, SeqCst);
437    }
438
439    pub fn simulate_long_connection_interruption(
440        &self,
441        peer_id: PeerId,
442        deterministic: BackgroundExecutor,
443    ) {
444        self.forbid_connections();
445        self.disconnect_client(peer_id);
446        deterministic.advance_clock(RECEIVE_TIMEOUT + RECONNECT_TIMEOUT);
447        self.allow_connections();
448        deterministic.advance_clock(RECEIVE_TIMEOUT + RECONNECT_TIMEOUT);
449        deterministic.run_until_parked();
450    }
451
452    pub fn forbid_connections(&self) {
453        self.forbid_connections.store(true, SeqCst);
454    }
455
456    pub fn allow_connections(&self) {
457        self.forbid_connections.store(false, SeqCst);
458    }
459
460    pub async fn make_contacts(&self, clients: &mut [(&TestClient, &mut TestAppContext)]) {
461        for ix in 1..clients.len() {
462            let (left, right) = clients.split_at_mut(ix);
463            let (client_a, cx_a) = left.last_mut().unwrap();
464            for (client_b, cx_b) in right {
465                client_a
466                    .app_state
467                    .user_store
468                    .update(*cx_a, |store, cx| {
469                        store.request_contact(client_b.user_id().unwrap(), cx)
470                    })
471                    .await
472                    .unwrap();
473                cx_a.executor().run_until_parked();
474                client_b
475                    .app_state
476                    .user_store
477                    .update(*cx_b, |store, cx| {
478                        store.respond_to_contact_request(client_a.user_id().unwrap(), true, cx)
479                    })
480                    .await
481                    .unwrap();
482            }
483        }
484    }
485
486    pub async fn make_channel(
487        &self,
488        channel: &str,
489        parent: Option<ChannelId>,
490        admin: (&TestClient, &mut TestAppContext),
491        members: &mut [(&TestClient, &mut TestAppContext)],
492    ) -> ChannelId {
493        let (_, admin_cx) = admin;
494        let channel_id = admin_cx
495            .read(ChannelStore::global)
496            .update(admin_cx, |channel_store, cx| {
497                channel_store.create_channel(channel, parent, cx)
498            })
499            .await
500            .unwrap();
501
502        for (member_client, member_cx) in members {
503            admin_cx
504                .read(ChannelStore::global)
505                .update(admin_cx, |channel_store, cx| {
506                    channel_store.invite_member(
507                        channel_id,
508                        member_client.user_id().unwrap(),
509                        ChannelRole::Member,
510                        cx,
511                    )
512                })
513                .await
514                .unwrap();
515
516            admin_cx.executor().run_until_parked();
517
518            member_cx
519                .read(ChannelStore::global)
520                .update(*member_cx, |channels, cx| {
521                    channels.respond_to_channel_invite(channel_id, true, cx)
522                })
523                .await
524                .unwrap();
525        }
526
527        channel_id
528    }
529
530    pub async fn make_public_channel(
531        &self,
532        channel: &str,
533        client: &TestClient,
534        cx: &mut TestAppContext,
535    ) -> ChannelId {
536        let channel_id = self
537            .make_channel(channel, None, (client, cx), &mut [])
538            .await;
539
540        client
541            .channel_store()
542            .update(cx, |channel_store, cx| {
543                channel_store.set_channel_visibility(
544                    channel_id,
545                    proto::ChannelVisibility::Public,
546                    cx,
547                )
548            })
549            .await
550            .unwrap();
551
552        channel_id
553    }
554
555    pub async fn make_channel_tree(
556        &self,
557        channels: &[(&str, Option<&str>)],
558        creator: (&TestClient, &mut TestAppContext),
559    ) -> Vec<ChannelId> {
560        let mut observed_channels = HashMap::default();
561        let mut result = Vec::new();
562        for (channel, parent) in channels {
563            let id;
564            if let Some(parent) = parent {
565                if let Some(parent_id) = observed_channels.get(parent) {
566                    id = self
567                        .make_channel(channel, Some(*parent_id), (creator.0, creator.1), &mut [])
568                        .await;
569                } else {
570                    panic!(
571                        "Edge {}->{} referenced before {} was created",
572                        parent, channel, parent
573                    )
574                }
575            } else {
576                id = self
577                    .make_channel(channel, None, (creator.0, creator.1), &mut [])
578                    .await;
579            }
580
581            observed_channels.insert(channel, id);
582            result.push(id);
583        }
584
585        result
586    }
587
588    pub async fn create_room(&self, clients: &mut [(&TestClient, &mut TestAppContext)]) {
589        self.make_contacts(clients).await;
590
591        let (left, right) = clients.split_at_mut(1);
592        let (_client_a, cx_a) = &mut left[0];
593        let active_call_a = cx_a.read(ActiveCall::global);
594
595        for (client_b, cx_b) in right {
596            let user_id_b = client_b.current_user_id(cx_b).to_proto();
597            active_call_a
598                .update(*cx_a, |call, cx| call.invite(user_id_b, None, cx))
599                .await
600                .unwrap();
601
602            cx_b.executor().run_until_parked();
603            let active_call_b = cx_b.read(ActiveCall::global);
604            active_call_b
605                .update(*cx_b, |call, cx| call.accept_incoming(cx))
606                .await
607                .unwrap();
608        }
609    }
610
611    pub async fn build_app_state(
612        test_db: &TestDb,
613        live_kit_test_server: &live_kit_client::TestServer,
614        executor: Executor,
615    ) -> Arc<AppState> {
616        Arc::new(AppState {
617            db: test_db.db().clone(),
618            live_kit_client: Some(Arc::new(live_kit_test_server.create_api_client())),
619            blob_store_client: None,
620            rate_limiter: Arc::new(RateLimiter::new(test_db.db().clone())),
621            executor,
622            clickhouse_client: None,
623            config: Config {
624                http_port: 0,
625                database_url: "".into(),
626                database_max_connections: 0,
627                api_token: "".into(),
628                invite_link_prefix: "".into(),
629                live_kit_server: None,
630                live_kit_key: None,
631                live_kit_secret: None,
632                rust_log: None,
633                log_json: None,
634                zed_environment: "test".into(),
635                blob_store_url: None,
636                blob_store_region: None,
637                blob_store_access_key: None,
638                blob_store_secret_key: None,
639                blob_store_bucket: None,
640                openai_api_key: None,
641                google_ai_api_key: None,
642                anthropic_api_key: None,
643                clickhouse_url: None,
644                clickhouse_user: None,
645                clickhouse_password: None,
646                clickhouse_database: None,
647                zed_client_checksum_seed: None,
648                slack_panics_webhook: None,
649                auto_join_channel_id: None,
650                migrations_path: None,
651                seed_path: None,
652            },
653        })
654    }
655}
656
657impl Deref for TestServer {
658    type Target = Server;
659
660    fn deref(&self) -> &Self::Target {
661        &self.server
662    }
663}
664
665impl Drop for TestServer {
666    fn drop(&mut self) {
667        self.server.teardown();
668        self.test_live_kit_server.teardown().unwrap();
669    }
670}
671
672impl Deref for TestClient {
673    type Target = Arc<Client>;
674
675    fn deref(&self) -> &Self::Target {
676        &self.app_state.client
677    }
678}
679
680impl TestClient {
681    pub fn fs(&self) -> &FakeFs {
682        self.app_state.fs.as_fake()
683    }
684
685    pub fn channel_store(&self) -> &Model<ChannelStore> {
686        &self.channel_store
687    }
688
689    pub fn notification_store(&self) -> &Model<NotificationStore> {
690        &self.notification_store
691    }
692
693    pub fn user_store(&self) -> &Model<UserStore> {
694        &self.app_state.user_store
695    }
696
697    pub fn language_registry(&self) -> &Arc<LanguageRegistry> {
698        &self.app_state.languages
699    }
700
701    pub fn client(&self) -> &Arc<Client> {
702        &self.app_state.client
703    }
704
705    pub fn current_user_id(&self, cx: &TestAppContext) -> UserId {
706        UserId::from_proto(
707            self.app_state
708                .user_store
709                .read_with(cx, |user_store, _| user_store.current_user().unwrap().id),
710        )
711    }
712
713    pub async fn wait_for_current_user(&self, cx: &TestAppContext) {
714        let mut authed_user = self
715            .app_state
716            .user_store
717            .read_with(cx, |user_store, _| user_store.watch_current_user());
718        while authed_user.next().await.unwrap().is_none() {}
719    }
720
721    pub async fn clear_contacts(&self, cx: &mut TestAppContext) {
722        self.app_state
723            .user_store
724            .update(cx, |store, _| store.clear_contacts())
725            .await;
726    }
727
728    pub fn local_projects(&self) -> impl Deref<Target = Vec<Model<Project>>> + '_ {
729        Ref::map(self.state.borrow(), |state| &state.local_projects)
730    }
731
732    pub fn remote_projects(&self) -> impl Deref<Target = Vec<Model<Project>>> + '_ {
733        Ref::map(self.state.borrow(), |state| &state.remote_projects)
734    }
735
736    pub fn local_projects_mut(&self) -> impl DerefMut<Target = Vec<Model<Project>>> + '_ {
737        RefMut::map(self.state.borrow_mut(), |state| &mut state.local_projects)
738    }
739
740    pub fn remote_projects_mut(&self) -> impl DerefMut<Target = Vec<Model<Project>>> + '_ {
741        RefMut::map(self.state.borrow_mut(), |state| &mut state.remote_projects)
742    }
743
744    pub fn buffers_for_project<'a>(
745        &'a self,
746        project: &Model<Project>,
747    ) -> impl DerefMut<Target = HashSet<Model<language::Buffer>>> + 'a {
748        RefMut::map(self.state.borrow_mut(), |state| {
749            state.buffers.entry(project.clone()).or_default()
750        })
751    }
752
753    pub fn buffers(
754        &self,
755    ) -> impl DerefMut<Target = HashMap<Model<Project>, HashSet<Model<language::Buffer>>>> + '_
756    {
757        RefMut::map(self.state.borrow_mut(), |state| &mut state.buffers)
758    }
759
760    pub fn channel_buffers(&self) -> impl DerefMut<Target = HashSet<Model<ChannelBuffer>>> + '_ {
761        RefMut::map(self.state.borrow_mut(), |state| &mut state.channel_buffers)
762    }
763
764    pub fn summarize_contacts(&self, cx: &TestAppContext) -> ContactsSummary {
765        self.app_state
766            .user_store
767            .read_with(cx, |store, _| ContactsSummary {
768                current: store
769                    .contacts()
770                    .iter()
771                    .map(|contact| contact.user.github_login.clone())
772                    .collect(),
773                outgoing_requests: store
774                    .outgoing_contact_requests()
775                    .iter()
776                    .map(|user| user.github_login.clone())
777                    .collect(),
778                incoming_requests: store
779                    .incoming_contact_requests()
780                    .iter()
781                    .map(|user| user.github_login.clone())
782                    .collect(),
783            })
784    }
785
786    pub async fn build_local_project(
787        &self,
788        root_path: impl AsRef<Path>,
789        cx: &mut TestAppContext,
790    ) -> (Model<Project>, WorktreeId) {
791        let project = self.build_empty_local_project(cx);
792        let (worktree, _) = project
793            .update(cx, |p, cx| {
794                p.find_or_create_local_worktree(root_path, true, cx)
795            })
796            .await
797            .unwrap();
798        worktree
799            .read_with(cx, |tree, _| tree.as_local().unwrap().scan_complete())
800            .await;
801        (project, worktree.read_with(cx, |tree, _| tree.id()))
802    }
803
804    pub async fn build_test_project(&self, cx: &mut TestAppContext) -> Model<Project> {
805        self.fs()
806            .insert_tree(
807                "/a",
808                json!({
809                    "1.txt": "one\none\none",
810                    "2.js": "function two() { return 2; }",
811                    "3.rs": "mod test",
812                }),
813            )
814            .await;
815        self.build_local_project("/a", cx).await.0
816    }
817
818    pub async fn host_workspace(
819        &self,
820        workspace: &View<Workspace>,
821        channel_id: ChannelId,
822        cx: &mut VisualTestContext,
823    ) {
824        cx.update(|cx| {
825            let active_call = ActiveCall::global(cx);
826            active_call.update(cx, |call, cx| call.join_channel(channel_id, cx))
827        })
828        .await
829        .unwrap();
830        cx.update(|cx| {
831            let active_call = ActiveCall::global(cx);
832            let project = workspace.read(cx).project().clone();
833            active_call.update(cx, |call, cx| call.share_project(project, cx))
834        })
835        .await
836        .unwrap();
837        cx.executor().run_until_parked();
838    }
839
840    pub async fn join_workspace<'a>(
841        &'a self,
842        channel_id: ChannelId,
843        cx: &'a mut TestAppContext,
844    ) -> (View<Workspace>, &'a mut VisualTestContext) {
845        cx.update(|cx| workspace::join_channel(channel_id, self.app_state.clone(), None, cx))
846            .await
847            .unwrap();
848        cx.run_until_parked();
849
850        self.active_workspace(cx)
851    }
852
853    pub fn build_empty_local_project(&self, cx: &mut TestAppContext) -> Model<Project> {
854        cx.update(|cx| {
855            Project::local(
856                self.client().clone(),
857                self.app_state.node_runtime.clone(),
858                self.app_state.user_store.clone(),
859                self.app_state.languages.clone(),
860                self.app_state.fs.clone(),
861                cx,
862            )
863        })
864    }
865
866    pub async fn build_remote_project(
867        &self,
868        host_project_id: u64,
869        guest_cx: &mut TestAppContext,
870    ) -> Model<Project> {
871        let active_call = guest_cx.read(ActiveCall::global);
872        let room = active_call.read_with(guest_cx, |call, _| call.room().unwrap().clone());
873        room.update(guest_cx, |room, cx| {
874            room.join_project(
875                host_project_id,
876                self.app_state.languages.clone(),
877                self.app_state.fs.clone(),
878                cx,
879            )
880        })
881        .await
882        .unwrap()
883    }
884
885    pub fn build_workspace<'a>(
886        &'a self,
887        project: &Model<Project>,
888        cx: &'a mut TestAppContext,
889    ) -> (View<Workspace>, &'a mut VisualTestContext) {
890        cx.add_window_view(|cx| {
891            cx.activate_window();
892            Workspace::new(
893                WorkspaceId::default(),
894                project.clone(),
895                self.app_state.clone(),
896                cx,
897            )
898        })
899    }
900
901    pub async fn build_test_workspace<'a>(
902        &'a self,
903        cx: &'a mut TestAppContext,
904    ) -> (View<Workspace>, &'a mut VisualTestContext) {
905        let project = self.build_test_project(cx).await;
906        cx.add_window_view(|cx| {
907            cx.activate_window();
908            Workspace::new(
909                WorkspaceId::default(),
910                project.clone(),
911                self.app_state.clone(),
912                cx,
913            )
914        })
915    }
916
917    pub fn active_workspace<'a>(
918        &'a self,
919        cx: &'a mut TestAppContext,
920    ) -> (View<Workspace>, &'a mut VisualTestContext) {
921        let window = cx.update(|cx| cx.active_window().unwrap().downcast::<Workspace>().unwrap());
922
923        let view = window.root_view(cx).unwrap();
924        let cx = VisualTestContext::from_window(*window.deref(), cx).as_mut();
925        // it might be nice to try and cleanup these at the end of each test.
926        (view, cx)
927    }
928}
929
930pub fn open_channel_notes(
931    channel_id: ChannelId,
932    cx: &mut VisualTestContext,
933) -> Task<anyhow::Result<View<ChannelView>>> {
934    let window = cx.update(|cx| cx.active_window().unwrap().downcast::<Workspace>().unwrap());
935    let view = window.root_view(cx).unwrap();
936
937    cx.update(|cx| ChannelView::open(channel_id, None, view.clone(), cx))
938}
939
940impl Drop for TestClient {
941    fn drop(&mut self) {
942        self.app_state.client.teardown();
943    }
944}