tests: rework the actually useful tests to live in the integration crate

crates/irtt-integration-tests/tests/managed_behavior.rs

@@ -0,0 +1,169 @@
+mod support;
+
+use std::{collections::HashMap, time::Duration};
+
+use irtt_client::{managed::*, ClientConfig, ClientEvent};
+use irtt_server::ServerConfig;
+use support::InTreeServer;
+use tokio::time::timeout;
+
+// A real service and public events protect cadence/fairness without depending
+// on a scheduler, a send cursor, or a private event-publication hook.
+#[tokio::test(flavor = "current_thread")]
+async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() {
+    timeout(Duration::from_secs(5), async {
+        let server = InTreeServer::start(ServerConfig::default());
+        for pacing in [ManagedPacing::Burst, ManagedPacing::Staggered] {
+            let config = ManagedClientConfig {
+                client: ClientConfig {
+                    duration: Some(Duration::from_millis(200)),
+                    interval: Duration::from_millis(20),
+                    probe_timeout: Duration::from_millis(100),
+                    open_timeouts: vec![Duration::from_millis(200)],
+                    ..ClientConfig::default()
+                },
+                pacing,
+                ..ManagedClientConfig::default()
+            };
+            let targets =
+                ["first", "second"].map(|id| ManagedTargetConfig::new(id, server.addr.to_string()));
+            let (task, handle) = ManagedClient::task(config, targets.into()).unwrap();
+            let mut events = handle.subscribe().unwrap();
+            let run = tokio::spawn(task);
+            let mut sends: HashMap<TargetInstance, Vec<_>> = HashMap::new();
+            while let Ok(event) = events.recv().await {
+                if let ManagedEvent::Client {
+                    target,
+                    event:
+                        ClientEvent::EchoSent {
+                            scheduled_at,
+                            sent_at,
+                            ..
+                        },
+                } = event
+                {
+                    let scheduled = scheduled_at.expect("managed probes expose their cadence");
+                    assert!(sent_at.mono >= scheduled);
+                    let previous = sends.entry(target).or_default();
+                    if let Some(last) = previous.last() {
+                        let elapsed: Duration = scheduled.duration_since(*last);
+                        assert!(!elapsed.is_zero());
+                        assert_eq!(elapsed.as_nanos() % Duration::from_millis(20).as_nanos(), 0);
+                    }
+                    previous.push(scheduled);
+                }
+            }
+            let outcome = run.await.unwrap();
+            assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete);
+            assert_eq!(outcome.successful_target_outcomes, 2);
+            assert_eq!(sends.len(), 2);
+            for target in outcome.recent_target_outcomes.iter() {
+                assert!(sends[&target.target].len() >= 2);
+                assert!(target.replies_received >= 2);
+            }
+        }
+    })
+    .await
+    .expect("managed pacing did not complete");
+}
+
+// Updates, receipts, generations and cleanup are observable through the control
+// handle and event subscription; no task internals or artificial races are needed.
+#[tokio::test(flavor = "current_thread")]
+async fn target_updates_preserve_identical_sessions_and_retire_removed_generations() {
+    timeout(Duration::from_secs(5), async {
+        let server = InTreeServer::start(ServerConfig::default());
+        let config = ManagedClientConfig {
+            client: ClientConfig {
+                duration: None,
+                interval: Duration::from_millis(20),
+                probe_timeout: Duration::from_millis(100),
+                open_timeouts: vec![Duration::from_millis(200)],
+                ..ClientConfig::default()
+            },
+            completion: ManagedCompletionPolicy::ExplicitStop,
+            ..ManagedClientConfig::default()
+        };
+        let configured = ManagedTargetConfig::new("one", server.addr.to_string());
+        let (task, handle) = ManagedClient::task(config, vec![]).unwrap();
+        let mut events = handle.subscribe().unwrap();
+        let run = tokio::spawn(task);
+        let added = handle
+            .update_targets(vec![configured.clone()])
+            .unwrap()
+            .await
+            .unwrap();
+        let first = added.status.targets[0].target.clone();
+        loop {
+            if matches!(events.recv().await.unwrap(), ManagedEvent::Client {
+                target, event: ClientEvent::EchoReply { .. }
+            } if target == first)
+            {
+                break;
+            }
+        }
+        let identical = handle
+            .update_targets(vec![configured.clone()])
+            .unwrap()
+            .await
+            .unwrap();
+        assert_eq!(identical.status.targets.len(), 1);
+        assert_eq!(identical.status.targets[0].target, first);
+        loop {
+            match events.recv().await.unwrap() {
+                ManagedEvent::Client {
+                    target,
+                    event: ClientEvent::SessionStarted { .. },
+                } if target == first => panic!("identical update reopened the session"),
+                ManagedEvent::Client {
+                    target,
+                    event: ClientEvent::EchoReply { .. },
+                } if target == first => break,
+                _ => {}
+            }
+        }
+        handle.update_targets(vec![]).unwrap().await.unwrap();
+        loop {
+            if let ManagedEvent::TargetFinished { outcome } = events.recv().await.unwrap() {
+                assert_eq!(outcome.target, first);
+                assert_eq!(outcome.end_reason, ManagedTargetEndReason::Removed);
+                assert_eq!(outcome.cleanup_failure, None);
+                break;
+            }
+        }
+        let readded = handle
+            .update_targets(vec![configured])
+            .unwrap()
+            .await
+            .unwrap();
+        let second = readded
+            .status
+            .targets
+            .iter()
+            .find(|target| target.desired)
+            .unwrap()
+            .target
+            .clone();
+        assert!(second.generation > first.generation);
+        loop {
+            if matches!(events.recv().await.unwrap(), ManagedEvent::Client {
+                target, event: ClientEvent::EchoReply { .. }
+            } if target == second)
+            {
+                break;
+            }
+        }
+        let stop = handle.stop();
+        let outcome = run.await.unwrap();
+        stop.await;
+        assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested);
+        assert_eq!(outcome.total_target_outcomes, 2);
+        assert!(outcome.recent_target_outcomes.iter().any(|target| {
+            target.target == second
+                && target.end_reason == ManagedTargetEndReason::Stopped
+                && target.cleanup_failure.is_none()
+        }));
+    })
+    .await
+    .expect("managed target updates did not complete");
+}