gio/cancellable-future: Add tests verifying the behavior with GioFuture

Since CancellableFuture can be used group and manage async operations
that support cancellation, add tests to ensure that this is the case

Author: 3v1n0

gio/src/cancellable_future.rs

@@ -161,6 +161,81 @@ mod tests {
         .unwrap()
     }
 
+    fn async_begin<P: FnOnce(Result<(), glib::Error>) + Send + 'static>(
+        duration: Duration,
+        must_be_cancelled_on_begin: bool,
+        must_be_cancelled_after_sleep: bool,
+        cancellable: &Cancellable,
+        callback: P,
+    ) {
+        // We do not use std::thread here since we want to simulate what C code normally does.
+        // Also not using spawn_blocking() directly, since we want to have the full control
+        // for the test case.
+        let callback = Box::new(callback);
+        let task = unsafe {
+            crate::Task::<bool>::new(None::<&glib::Binding>, Some(cancellable), move |t, _| {
+                let cancellable = t.cancellable().unwrap();
+                let ret = t.propagate();
+                println!(
+                    "Task callback, returning {:?} - cancelled {}",
+                    ret,
+                    cancellable.is_cancelled()
+                );
+                assert_eq!(cancellable.is_cancelled(), must_be_cancelled_after_sleep);
+                match ret {
+                    Err(e) => callback(Err(e)),
+                    Ok(_) => callback(Ok(())),
+                };
+            })
+        };
+
+        task.run_in_thread(move |task, _: Option<&glib::Binding>, cancellable| {
+            let cancellable = cancellable.unwrap();
+            let func = || {
+                println!(
+                    "Task thread started, cancelled {} - want {}",
+                    cancellable.is_cancelled(),
+                    must_be_cancelled_on_begin
+                );
+                assert_eq!(cancellable.is_cancelled(), must_be_cancelled_on_begin);
+                std::thread::sleep(duration);
+                assert_eq!(cancellable.is_cancelled(), must_be_cancelled_after_sleep);
+                println!(
+                    "Task thread done, cancelled {} - want {}",
+                    cancellable.is_cancelled(),
+                    must_be_cancelled_after_sleep
+                )
+            };
+
+            if let Err(e) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(func)) {
+                std::panic::resume_unwind(e);
+            }
+
+            unsafe {
+                task.return_result(match cancellable.set_error_if_cancelled() {
+                    Err(e) => Err(e),
+                    Ok(_) => Ok(true),
+                });
+            }
+        });
+    }
+
+    fn async_future(
+        duration: Duration,
+        must_be_cancelled_on_begin: bool,
+        must_be_cancelled_after_sleep: bool,
+    ) -> std::pin::Pin<Box<dyn Future<Output = Result<(), glib::Error>> + 'static>> {
+        Box::pin(crate::GioFuture::new(&(), move |_, cancellable, send| {
+            async_begin(
+                duration,
+                must_be_cancelled_on_begin,
+                must_be_cancelled_after_sleep,
+                cancellable,
+                move |res| send.resolve(res),
+            );
+        }))
+    }
+
     #[test]
     fn cancellable_future_ok() {
         let ctx = glib::MainContext::new();
@@ -234,4 +309,108 @@ mod tests {
         assert!(matches!(r1, Err(Cancelled)));
         assert!(matches!(r2, ()));
     }
+
+    #[test]
+    fn cancellable_future_immediate_cancel_with_gio_future() {
+        let ctx = glib::MainContext::new();
+        let c = Cancellable::new();
+
+        async fn async_chain() -> Result<(), glib::Error> {
+            async_future(Duration::from_millis(250), true, true).await?;
+            async_future(Duration::from_hours(99999), true, true).await
+        }
+
+        c.cancel();
+
+        let result = ctx
+            .block_on(ctx.spawn_local({
+                CancellableFuture::new(
+                    futures_util::future::join5(
+                        async_chain(),
+                        async_chain(),
+                        async_chain(),
+                        async_chain(),
+                        CancellableFuture::new(async_chain(), Cancellable::new()),
+                    ),
+                    c.clone(),
+                )
+            }))
+            .expect("futures must be executed");
+
+        assert!(matches!(result, Err(Cancelled)));
+    }
+
+    #[test]
+    fn cancellable_future_delayed_cancel_with_gio_future() {
+        let ctx = glib::MainContext::new();
+        let c = Cancellable::new();
+
+        async fn async_chain() -> Result<(), glib::Error> {
+            async_future(Duration::from_millis(250), false, true).await?;
+            async_future(Duration::from_hours(99999), true, true).await
+        }
+
+        let (result, _, _) = ctx
+            .block_on(ctx.spawn_local({
+                futures_util::future::join3(
+                    CancellableFuture::new(
+                        futures_util::future::join5(
+                            async_chain(),
+                            async_chain(),
+                            async_chain(),
+                            async_chain(),
+                            CancellableFuture::new(async_chain(), Cancellable::new()),
+                        ),
+                        c.clone(),
+                    ),
+                    cancel_after_sleep_in_thread(Duration::from_millis(100), c.clone()),
+                    // Let's wait a bit more to ensure that more events are processed
+                    // by the loop. Not required, but it simulates a more real
+                    // scenario.
+                    glib::timeout_future(Duration::from_millis(350)),
+                )
+            }))
+            .expect("futures must be executed");
+
+        assert!(matches!(result, Err(Cancelled)));
+    }
+
+    #[test]
+    fn cancellable_future_late_cancel_with_gio_future() {
+        let ctx = glib::MainContext::new();
+        let c = Cancellable::new();
+
+        async fn async_chain() -> Result<(), glib::Error> {
+            async_future(Duration::from_millis(100), false, false).await?;
+            async_future(Duration::from_millis(100), false, false).await
+        }
+
+        let results = ctx
+            .block_on(ctx.spawn_local(async move {
+                let ret = CancellableFuture::new(
+                    futures_util::future::join5(
+                        async_chain(),
+                        async_chain(),
+                        async_chain(),
+                        async_chain(),
+                        CancellableFuture::new(async_chain(), Cancellable::new()),
+                    ),
+                    c.clone(),
+                )
+                .await;
+
+                c.cancel();
+                ret
+            }))
+            .expect("futures must be executed");
+
+        assert!(results.is_ok());
+
+        let r1 = results.unwrap();
+        assert!(matches!(r1.0, Ok(())));
+        assert!(matches!(r1.1, Ok(())));
+        assert!(matches!(r1.2, Ok(())));
+        assert!(matches!(r1.3, Ok(())));
+        assert!(matches!(r1.4.unwrap(), Ok(())));
+    }
 }