refactor: test directory (#88)

* refactor: test directory

* chore: rename

libs/client-api/src/http.rs

@@ -513,10 +513,10 @@ impl Client {
     Ok(format!("{}/{}/{}", self.ws_addr, access_token, device_id))
   }
 
-  pub async fn put_file_storage_object(
+  pub async fn put_file_storage_object<T: Into<Bytes>>(
     &self,
     path: &str,
-    data: Bytes,
+    data: T,
     mime: &Mime,
   ) -> Result<(), AppError> {
     let url = format!("{}/api/file_storage/{}", self.base_url, path);
@@ -524,7 +524,7 @@ impl Client {
       .http_client_with_auth(Method::PUT, &url)
       .await?
       .header(header::CONTENT_TYPE, mime.to_string())
-      .body(data)
+      .body(data.into())
       .send()
       .await?;
     AppResponse::<()>::from_response(resp).await?.into_error()

libs/realtime/src/collaborate/retry.rs

@@ -23,22 +23,21 @@ use crate::collaborate::group::CollabGroupCache;
 use crate::error::RealtimeError;
 use tracing::{error, trace, warn};
 
-pub(crate) struct SubscribeGroupIfNeedAction<'a, U, S> {
+pub(crate) struct SubscribeGroupIfNeed<'a, U, S> {
   pub(crate) client_msg: &'a ClientMessage<U>,
   pub(crate) groups: &'a Arc<CollabGroupCache<S, U>>,
   pub(crate) edit_collab_by_user: &'a Arc<Mutex<HashMap<U, HashSet<Editing>>>>,
   pub(crate) client_stream_by_user: &'a Arc<RwLock<HashMap<U, CollabClientStream>>>,
 }
 
-impl<'a, U, S> SubscribeGroupIfNeedAction<'a, U, S>
+impl<'a, U, S> SubscribeGroupIfNeed<'a, U, S>
 where
   U: RealtimeUser,
   S: CollabStorage,
 {
   pub(crate) fn run(
     self,
-  ) -> RetryIf<Take<FixedInterval>, SubscribeGroupIfNeedAction<'a, U, S>, SubscribeGroupCondition<U>>
+  ) -> RetryIf<Take<FixedInterval>, SubscribeGroupIfNeed<'a, U, S>, SubscribeGroupCondition<U>> {
-  {
     let weak_client_stream = Arc::downgrade(self.client_stream_by_user);
     let retry_strategy = FixedInterval::new(Duration::from_secs(2)).take(5);
     RetryIf::spawn(
@@ -49,7 +48,7 @@ where
   }
 }
 
-impl<'a, U, S> Action for SubscribeGroupIfNeedAction<'a, U, S>
+impl<'a, U, S> Action for SubscribeGroupIfNeed<'a, U, S>
 where
   U: RealtimeUser,
   S: CollabStorage,
@@ -87,7 +86,7 @@ where
         }
       }
 
-      // If the client's stream is already subscribed to the collab group, return.
+      // If the client's stream is already subscribe to the collab, return.
       if self
         .groups
         .contains_user(object_id, &self.client_msg.user)
@@ -136,11 +135,9 @@ where
                   });
 
                 let (sink, stream) = client_stream
-                  .client_channel::<CollabMessage, _, _>(
+                  .client_channel::<CollabMessage, _>(object_id, move |object_id, msg| {
-                    object_id,
+                    msg.object_id() == object_id
-                    move |object_id, msg| msg.object_id() == object_id,
+                  })
-                    move |object_id, msg| msg.object_id == object_id,
-                  )
                   .unwrap();
 
                 collab_group

libs/realtime/src/collaborate/server.rs

@@ -17,7 +17,7 @@ use tracing::{info, trace};
 
 use crate::client::ClientWSSink;
 use crate::collaborate::group::CollabGroupCache;
-use crate::collaborate::retry::SubscribeGroupIfNeedAction;
+use crate::collaborate::retry::SubscribeGroupIfNeed;
 use crate::util::channel_ext::UnboundedSenderSink;
 use database::collab::CollabStorage;
 
@@ -149,7 +149,7 @@ where
     let edit_collab_by_user = self.editing_collab_by_user.clone();
 
     Box::pin(async move {
-      SubscribeGroupIfNeedAction {
+      SubscribeGroupIfNeed {
         client_msg: &client_msg,
         groups: &groups,
         edit_collab_by_user: &edit_collab_by_user,
@@ -158,14 +158,14 @@ where
       .run()
       .await?;
 
-      forward_message_to_collab_group(&client_msg, &client_stream_by_user).await;
+      broadcast_message(&client_msg, &client_stream_by_user).await;
       Ok(())
     })
   }
 }
 
 #[inline]
-async fn forward_message_to_collab_group<U>(
+async fn broadcast_message<U>(
   client_msg: &ClientMessage<U>,
   client_streams: &Arc<RwLock<HashMap<U, CollabClientStream>>>,
 ) where
@@ -247,11 +247,10 @@ impl CollabClientStream {
 
   /// Returns a [UnboundedSenderSink] and a [ReceiverStream] for the object_id.
   #[allow(clippy::type_complexity)]
-  pub fn client_channel<T, F1, F2>(
+  pub fn client_channel<T, F1>(
     &mut self,
     object_id: &str,
     sink_filter: F1,
-    stream_filter: F2,
   ) -> Option<(
     UnboundedSenderSink<T>,
     ReceiverStream<Result<T, StreamError>>,
@@ -260,7 +259,6 @@ impl CollabClientStream {
     T:
       TryFrom<RealtimeMessage, Error = StreamError> + Into<RealtimeMessage> + Send + Sync + 'static,
     F1: Fn(&str, &T) -> bool + Send + Sync + 'static,
-    F2: Fn(&str, &RealtimeMessage) -> bool + Send + Sync + 'static,
   {
     let client_ws_sink = self.ws_sink.clone();
     let mut stream_rx = BroadcastStream::new(self.stream_tx.subscribe());
@@ -277,13 +275,13 @@ impl CollabClientStream {
     });
     let client_forward_sink = UnboundedSenderSink::<T>::new(tx);
 
-    // forward the message to the stream that can be subscribed by the broadcast group, which will
+    // forward the message to the stream that was subscribed by the broadcast group, which will
     // send the messages to all connected clients using the client_forward_sink
     let cloned_object_id = object_id.to_string();
     let (tx, rx) = tokio::sync::mpsc::channel(100);
     tokio::spawn(async move {
       while let Some(Ok(Ok(msg))) = stream_rx.next().await {
-        if stream_filter(&cloned_object_id, &msg) {
+        if cloned_object_id == msg.object_id {
           let _ = tx.send(T::try_from(msg)).await;
         }
       }

tests/client/constants.rs

@@ -1,3 +0,0 @@
-pub const LOCALHOST_URL: &str = "http://localhost:8000";
-pub const LOCALHOST_WS: &str = "ws://localhost:8000/ws";
-pub const LOCALHOST_GOTRUE: &str = "http://localhost:9998";

tests/collab/storage_test.rs

@@ -1,5 +1,5 @@
 use crate::{
-  client::utils::generate_unique_registered_user_client, collab::workspace_id_from_client,
+  collab::workspace_id_from_client, user::utils::generate_unique_registered_user_client,
 };
 
 use collab_define::CollabType;

tests/file_storage/file_test.rs

@@ -1,6 +1,6 @@
 use shared_entity::error_code::ErrorCode;
 
-use crate::client::utils::generate_unique_registered_user_client;
+use crate::user::utils::generate_unique_registered_user_client;
 
 #[tokio::test]
 async fn get_but_not_exists() {
@@ -18,9 +18,7 @@ async fn put_and_get() {
   let mime = mime::TEXT_PLAIN_UTF_8;
   let data = "hello world";
   let path = "mydata";
-  c1.put_file_storage_object(path, data.into(), &mime)
+  c1.put_file_storage_object(path, data, &mime).await.unwrap();
-    .await
-    .unwrap();
 
   let got_data = c1.get_file_storage_object(path).await.unwrap();
   assert_eq!(got_data, data.as_bytes());
@@ -33,10 +31,10 @@ async fn put_and_put_and_get() {
   let data1 = "my content 1";
   let data2 = "my content 2";
   let path = "mydata";
-  c1.put_file_storage_object(path, data1.into(), &mime)
+  c1.put_file_storage_object(path, data1, &mime)
     .await
     .unwrap();
-  c1.put_file_storage_object(path, data2.into(), &mime)
+  c1.put_file_storage_object(path, data2, &mime)
     .await
     .unwrap();
 
@@ -60,9 +58,7 @@ async fn put_delete_get() {
   let mime = mime::TEXT_PLAIN_UTF_8;
   let data = "my contents";
   let path = "mydata";
-  c1.put_file_storage_object(path, data.into(), &mime)
+  c1.put_file_storage_object(path, data, &mime).await.unwrap();
-    .await
-    .unwrap();
   c1.delete_file_storage_object(path).await.unwrap();
 
   let err = c1.get_file_storage_object(path).await.unwrap_err();

tests/file_storage/mod.rs

@@ -0,0 +1 @@
+mod file_test;

tests/gotrue/admin.rs

@@ -6,11 +6,9 @@ use gotrue::{
 };
 
 use crate::{
-  client::{
-    constants::LOCALHOST_GOTRUE,
-    utils::{generate_unique_email, ADMIN_USER},
-  },
   client_api_client,
+  user::utils::{generate_unique_email, ADMIN_USER},
+  LOCALHOST_GOTRUE,
 };
 
 #[tokio::test]

tests/gotrue/settings.rs

@@ -4,7 +4,7 @@ use gotrue::{
   params::AdminUserParams,
 };
 
-use crate::client::utils::{generate_unique_email, ADMIN_USER};
+use crate::user::utils::{generate_unique_email, ADMIN_USER};
 
 #[tokio::test]
 async fn gotrue_settings() {

tests/main.rs

@@ -1,11 +1,15 @@
-use crate::client::constants::{LOCALHOST_URL, LOCALHOST_WS};
-use client::constants::LOCALHOST_GOTRUE;
 use client_api::Client;
 
-mod client;
 mod collab;
 mod gotrue;
 mod realtime;
+mod user;
+
+mod file_storage;
+
+pub const LOCALHOST_URL: &str = "http://localhost:8000";
+pub const LOCALHOST_WS: &str = "ws://localhost:8000/ws";
+pub const LOCALHOST_GOTRUE: &str = "http://localhost:9998";
 
 pub fn client_api_client() -> Client {
   Client::new(LOCALHOST_URL, LOCALHOST_WS, LOCALHOST_GOTRUE)

tests/realtime/connect_test.rs

@@ -1,6 +1,6 @@
 use client_api::ws::{ConnectState, WSClient, WSClientConfig};
 
-use crate::client::utils::generate_unique_registered_user_client;
+use crate::user::utils::generate_unique_registered_user_client;
 
 #[tokio::test]
 async fn realtime_connect_test() {

tests/realtime/offline_edit_collab_test.rs

@@ -1,5 +1,5 @@
-use crate::client::utils::generate_unique_registered_user;
 use crate::realtime::test_client::{assert_client_collab, assert_remote_collab, TestClient};
+use crate::user::utils::generate_unique_registered_user;
 use std::time::Duration;
 
 use collab_define::CollabType;

tests/realtime/test_client.rs

@@ -17,8 +17,8 @@ use tracing_subscriber::fmt::Subscriber;
 use tracing_subscriber::util::SubscriberInitExt;
 use tracing_subscriber::EnvFilter;
 
-use crate::client::utils::{generate_unique_registered_user, User};
 use crate::client_api_client;
+use crate::user::utils::{generate_unique_registered_user, User};
 
 pub(crate) struct TestClient {
   pub ws_client: WSClient,

tests/user/mod.rs

@@ -1,5 +1,3 @@
-pub mod constants;
-mod file_storage;
 mod refresh;
 mod sign_in;
 mod sign_out;

tests/user/refresh.rs

@@ -1,6 +1,6 @@
 use std::time::SystemTime;
 
-use crate::client::utils::generate_unique_registered_user_client;
+use crate::user::utils::generate_unique_registered_user_client;
 
 #[tokio::test]
 async fn refresh_success() {

tests/user/sign_in.rs

@@ -1,7 +1,7 @@
 use shared_entity::error_code::ErrorCode;
 
-use crate::client::utils::{generate_unique_email, generate_unique_registered_user, ADMIN_USER};
 use crate::client_api_client;
+use crate::user::utils::{generate_unique_email, generate_unique_registered_user, ADMIN_USER};
 
 #[tokio::test]
 async fn sign_in_unknown_user() {

tests/user/sign_out.rs

@@ -1,4 +1,4 @@
-use crate::{client::utils::generate_unique_registered_user_client, client_api_client};
+use crate::{client_api_client, user::utils::generate_unique_registered_user_client};
 
 #[tokio::test]
 async fn sign_out_but_not_sign_in() {

tests/user/sign_up.rs

@@ -2,8 +2,8 @@ use gotrue_entity::OAuthProvider;
 use shared_entity::error_code::ErrorCode;
 
 use crate::{
-  client::utils::{generate_unique_email, generate_unique_registered_user_client},
   client_api_client,
+  user::utils::{generate_unique_email, generate_unique_registered_user_client},
 };
 
 #[tokio::test]

tests/user/update.rs

@@ -1,8 +1,8 @@
 use shared_entity::dto::UserUpdateParams;
 use shared_entity::error_code::ErrorCode;
 
-use crate::client::utils::{generate_unique_email, generate_unique_registered_user_client};
 use crate::client_api_client;
+use crate::user::utils::{generate_unique_email, generate_unique_registered_user_client};
 
 #[tokio::test]
 async fn update_but_not_logged_in() {

tests/user/workspace.rs

@@ -1,6 +1,6 @@
 use shared_entity::error_code::ErrorCode;
 
-use crate::client::utils::generate_unique_registered_user_client;
+use crate::user::utils::generate_unique_registered_user_client;
 
 #[tokio::test]
 async fn add_workspace_members_not_enough_permission() {