Writing basic tests for impulses

Signed-off-by: Michael X. Grey <grey@openrobotics.org>

Author: mxgrey

src/callback.rs

@@ -17,7 +17,7 @@
 
 use crate::{
     BlockingCallback, AsyncCallback, Channel, InnerChannel, ChannelQueue,
-    OperationRoster, StreamPack, Input, Provider, ProvideOnce, UnhandledErrors,
+    OperationRoster, StreamPack, Input, Provider, ProvideOnce,
     AddOperation, OperateCallback, ManageInput, OperationError, SetupFailure,
     OrBroken, OperateTask, Operation, OperationSetup,
 };
@@ -108,14 +108,8 @@ impl<'a> CallbackRequest<'a> {
         let sender = self.world.get_resource_or_insert_with(|| ChannelQueue::new()).sender.clone();
         let task = AsyncComputeTaskPool::get().spawn(task);
         let task_id = self.world.spawn(()).id();
-        let operation = OperateTask::new(task_id, session, self.source, self.target, task, None, sender);
-        let setup = OperationSetup { source: task_id, world: self.world };
-        if let Err(error) = operation.setup(setup) {
-            self.world.get_resource_or_insert_with(|| UnhandledErrors::default())
-                .setup
-                .push(SetupFailure { broken_node: self.source, error });
-        }
-        self.roster.queue(task_id);
+        OperateTask::new(task_id, session, self.source, self.target, task, None, sender)
+            .add(self.world, self.roster);
         Ok(())
     }
 

src/errors.rs

@@ -40,6 +40,20 @@ pub struct UnhandledErrors {
     pub miscellaneous: Vec<MiscellaneousFailure>,
 }
 
+impl UnhandledErrors {
+    pub fn is_empty(&self) -> bool {
+        self.setup.is_empty()
+        && self.cancellations.is_empty()
+        && self.operations.is_empty()
+        && self.disposals.is_empty()
+        && self.stop_tasks.is_empty()
+        && self.broken.is_empty()
+        && self.unused_targets.is_empty()
+        && self.connections.is_empty()
+        && self.miscellaneous.is_empty()
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct SetupFailure {
     pub broken_node: Entity,

src/impulse.rs

@@ -89,7 +89,7 @@ where
     pub fn take(self) -> Recipient<Response, Streams> {
         let (response_sender, response_promise) = Promise::<Response>::new();
         self.commands.add(AddImpulse::new(
-            self.target,
+            dbg!(self.target),
             TakenResponse::<Response>::new(response_sender),
         ));
         let mut map = StreamTargetMap::default();
@@ -107,7 +107,7 @@ where
     pub fn take_response(self) -> Promise<Response> {
         let (response_sender, response_promise) = Promise::<Response>::new();
         self.commands.add(AddImpulse::new(
-            self.target,
+            dbg!(self.target),
             TakenResponse::<Response>::new(response_sender),
         ));
         response_promise
@@ -323,3 +323,43 @@ impl<T> Default for Collection<T> {
         Self { items: Default::default() }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::{*, testing::*};
+    use std::time::Duration;
+
+    #[test]
+    fn test_provide() {
+        let mut context = TestingContext::minimal_plugins();
+
+        let mut promise = context.build(|commands| {
+            commands.provide("hello".to_owned()).take_response()
+        });
+        assert!(promise.peek().available().is_some_and(|v| v == "hello"));
+    }
+
+    #[test]
+    fn test_async_map() {
+        let mut context = TestingContext::minimal_plugins();
+
+        let mut promise = context.build(|commands| {
+            commands
+            .request(
+                WaitRequest {
+                    duration: Duration::from_secs_f64(0.001),
+                    value: "hello".to_owned(),
+                },
+                wait.into_async_map(),
+            )
+            .take_response()
+        });
+
+        context.run_with_conditions(
+            &mut promise,
+            FlushConditions::new()
+            .with_timeout(Duration::from_secs_f64(5.0)),
+        );
+        assert!(promise.peek().available().is_some_and(|v| v == "hello"));
+    }
+}

src/impulse/map.rs

@@ -26,7 +26,7 @@ use crate::{
     Impulsive, OperationSetup, OperationRequest, SingleTargetStorage, StreamPack,
     InputBundle, OperationResult, OrBroken, Input, ManageInput,
     ChannelQueue, BlockingMap, AsyncMap, InnerChannel, OperateTask, ActiveTasksStorage,
-    CallBlockingMapOnce, CallAsyncMapOnce, Operation,
+    CallBlockingMapOnce, CallAsyncMapOnce,
 };
 
 /// The key difference between this and [`crate::OperateBlockingMap`] is that
@@ -176,8 +176,7 @@ where
 
         let task_source = world.spawn(()).id();
         OperateTask::new(task_source, session, source, target, task, None, sender)
-            .setup(OperationSetup { source: task_source, world })?;
-        roster.queue(task_source);
+            .add(world, roster);
         Ok(())
     }
 }

src/operation/operate_map.rs

@@ -15,6 +15,13 @@
  *
 */
 
+use bevy::{
+    prelude::{Component, Entity, Bundle},
+    tasks::AsyncComputeTaskPool,
+};
+
+use std::future::Future;
+
 use crate::{
     BlockingMap, AsyncMap, Operation, ChannelQueue, InnerChannel,
     SingleTargetStorage, StreamPack, Input, ManageInput, OperationCleanup,
@@ -23,13 +30,6 @@ use crate::{
     OperationReachability, ReachabilityResult, InputBundle,
 };
 
-use bevy::{
-    prelude::{Component, Entity, Bundle},
-    tasks::AsyncComputeTaskPool,
-};
-
-use std::future::Future;
-
 #[derive(Bundle)]
 pub(crate) struct OperateBlockingMap<F, Request, Response, Streams>
 where
@@ -189,9 +189,9 @@ where
             .f = Some(f);
 
         let task_source = world.spawn(()).id();
-        OperateTask::new(task_source, session, source, target, task, None, sender)
-            .setup(OperationSetup { source: task_source, world })?;
-        roster.queue(task_source);
+        OperateTask::new(
+            dbg!(task_source), session, source, dbg!(target), task, None, sender
+        ).add(world, roster);
         Ok(())
     }
 

src/operation/operate_task.rs

@@ -18,6 +18,7 @@
 use bevy::{
     prelude::{Component, Entity, World, Resource, BuildWorldChildren},
     tasks::{Task as BevyTask, AsyncComputeTaskPool},
+    ecs::system::Command,
 };
 
 use std::{
@@ -36,7 +37,7 @@ use smallvec::SmallVec;
 
 use crate::{
     OperationRoster, Blocker, ManageInput, ChannelQueue, UnhandledErrors,
-    OperationSetup, OperationRequest, OperationResult, Operation,
+    OperationSetup, OperationRequest, OperationResult, Operation, AddOperation,
     OrBroken, OperationCleanup, ChannelItem, OperationError, Broken,
     OperationReachability, ReachabilityResult, emit_disposal, Disposal,
 };
@@ -92,6 +93,12 @@ impl<Response: 'static + Send + Sync> OperateTask<Response> {
     ) -> Self {
         Self { source, session, node, target, task: Some(task), blocker, sender, disposal: None }
     }
+
+    pub(crate) fn add(self, world: &mut World, roster: &mut OperationRoster) {
+        let source = self.source;
+        AddOperation::new(source, self).apply(world);
+        roster.queue(source);
+    }
 }
 
 impl<Response: 'static + Send + Sync> Drop for OperateTask<Response> {

src/request.rs

@@ -20,8 +20,8 @@ use bevy::prelude::{Commands, BuildChildren};
 use std::future::Future;
 
 use crate::{
-    UnusedTarget, InputCommand, StreamPack, ProvideOnce, IntoAsyncMap, Impulse,
-    Detached,
+    UnusedTarget, StreamPack, ProvideOnce, IntoBlockingMapOnce, IntoAsyncMap,
+    Impulse, Detached, InputCommand,
 };
 
 /// Extensions for creating impulse chains by making a request to a provider or
@@ -97,9 +97,9 @@ impl<'w, 's> RequestExt<'w, 's> for Commands<'w, 's> {
             .set_parent(target)
             .id();
 
-        provider.connect(source, target, self);
+        provider.connect(source, dbg!(target), self);
 
-        self.add(InputCommand { session: source, target: source, data: request });
+        self.add(InputCommand { session: dbg!(source), target: dbg!(source), data: request });
 
         Impulse {
             source,
@@ -113,22 +113,7 @@ impl<'w, 's> RequestExt<'w, 's> for Commands<'w, 's> {
         &'a mut self,
         value: T,
     ) -> Impulse<'w, 's, 'a, T, ()> {
-        let target = self.spawn((
-            Detached::default(),
-            UnusedTarget,
-        )).id();
-
-        self.add(InputCommand { session: target, target, data: value });
-
-        Impulse {
-            target,
-            // The source field won't actually matter for an impulse produced by
-            // this provide method, so we'll just use the session value as a
-            // placeholder
-            source: target,
-            commands: self,
-            _ignore: Default::default(),
-        }
+        self.request(value, provide_value.into_blocking_map_once())
     }
 
     fn serve<'a, T: 'static + Send + Sync + Future>(
@@ -142,6 +127,10 @@ impl<'w, 's> RequestExt<'w, 's> for Commands<'w, 's> {
     }
 }
 
+fn provide_value<T>(value: T) -> T {
+    value
+}
+
 async fn async_server<T: Future>(value: T) -> T::Output {
     value.await
 }

src/service/async_srv.rs

@@ -19,8 +19,8 @@ use crate::{
     AsyncService, InAsyncService, IntoService, ServiceTrait, ServiceBundle, ServiceRequest,
     InnerChannel, ChannelQueue, OperationRoster, Blocker,
     StreamPack, ServiceBuilder, ChooseAsyncServiceDelivery, OperationRequest,
-    OperationError, OrBroken, ManageInput, Input, OperateTask, Operation,
-    OperationSetup, SingleTargetStorage, dispose_for_despawned_service,
+    OperationError, OrBroken, ManageInput, Input, OperateTask,
+    SingleTargetStorage, dispose_for_despawned_service,
     service::builder::{SerialChosen, ParallelChosen}, Disposal, emit_disposal,
     StopTask, UnhandledErrors, StopTaskFailure, Delivery, DeliveryInstructions,
     Deliver, DeliveryOrder, DeliveryUpdate, insert_new_order, pop_next_delivery,
@@ -238,8 +238,7 @@ where
     let task = AsyncComputeTaskPool::get().spawn(job);
 
     OperateTask::new(task_id, session, source, target, task, blocker, sender)
-        .setup(OperationSetup { source: task_id, world })?;
-    roster.queue(task_id);
+        .add(world, roster);
     Ok(())
 }
 

src/testing.rs

@@ -24,7 +24,10 @@ pub use bevy::{
     ecs::system::CommandQueue,
 };
 
-use crate::{Promise, Service, InAsyncService, InBlockingService, flush_impulses};
+use crate::{
+    Promise, Service, InAsyncService, InBlockingService, UnhandledErrors,
+    flush_impulses,
+};
 
 pub struct TestingContext {
     pub app: App,
@@ -109,6 +112,18 @@ impl TestingContext {
 
         return true;
     }
+
+    pub fn no_unhandled_errors(&self) -> bool {
+        let Some(errors) = self.app.world.get_resource::<UnhandledErrors>() else {
+            return true;
+        };
+
+        errors.is_empty()
+    }
+
+    pub fn get_unhandled_errors(&self) -> Option<&UnhandledErrors> {
+        self.app.world.get_resource::<UnhandledErrors>()
+    }
 }
 
 #[derive(Default, Clone)]