feat(cli): added cli validator tests (#286)

* feat: added cli validator tests

* suggestions

* clippy

* simplify timeout iteration

* fix(cli): address validator test PR review comments

- Replace `name()` method with `fmt::Display` impl on `ValidatorTestCase`
- Initialize tracing with `pluto_tracing::init` in `run`
- Move `start_time` immediately before `run_tests_with_timeout`
- Fix `ERR_TIMEOUT_INTERRUPTED` message to `"timeout/interrupted"`
- Replace `checked_add`/`expect` with `Duration::saturating_sub` for timeout tracking
- Remove outer `tokio::spawn` in `ping_load_test`; use a scoped block so the
  `JoinSet` is dropped (aborting workers) on timeout cancellation
- Use `workers.join_all().await` instead of manual `join_next` loop
- Change `ping_continuously` signature to `impl AsRef<str>`
- Remove unnecessary `drop(conn)` before `return`
- Qualify `super::` helper functions instead of importing them directly

Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com>

* fix(deny): add advisory ignores for transitive deps

All entries are transitive dependencies that cannot be immediately
upgraded. None are reachable from Pluto's production code paths.

Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com>

* fix(cli): address emlautarom1 review comments on validator tests

- Add CancellationToken to validator::run and run_tests_with_timeout;
  use tokio::select! so Ctrl+C (SIGINT) cancels in-progress tests
- Replace ERR_TIMEOUT_INTERRUPTED string const with CliError::TimeoutInterrupted
  and CliError::Other("test case not supported") with CliError::TestCaseNotSupported;
  remove leading underscore from both variants in the error enum
- Remove RUSTSEC advisory ignores added to deny.toml; run cargo update
  instead — all seven advisories are resolved by updated transitive deps

Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com>

* fix linter

* fix: stop ping_continuously loop on dial error to match charon behavior

In Charon's `pingValidatorContinuously`, both dial errors and timeouts
cause an immediate `return`, stopping the loop. Port this behavior to
Rust so a failing connection doesn't inflate the RTT sample set with
absent readings.

Co-Authored-By: varex83 <varex83@users.noreply.github.com>

---------

Co-authored-by: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com>
Co-authored-by: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com>
Co-authored-by: varex83 <varex83@users.noreply.github.com>

Author: 3 people

crates/cli/src/commands/test/mod.rs

@@ -69,18 +69,13 @@ pub struct TestConfigArgs {
 }
 
 /// Lists available test case names for a given test category.
-/// TODO: Fill with enums TestCases of each category
 fn list_test_cases(category: TestCategory) -> Vec<String> {
     // Returns available test case names for each category.
     match category {
-        TestCategory::Validator => {
-            // From validator::supported_validator_test_cases()
-            vec![
-                "Ping".to_string(),
-                "PingMeasure".to_string(),
-                "PingLoad".to_string(),
-            ]
-        }
+        TestCategory::Validator => validator::ValidatorTestCase::all()
+            .iter()
+            .map(|tc| tc.to_string())
+            .collect(),
         TestCategory::Beacon => beacon::test_case_names(),
         TestCategory::Mev => {
             vec![

crates/cli/src/commands/test/validator.rs

@@ -1,9 +1,61 @@
 //! Validator client connectivity tests.
 
-use super::{TestConfigArgs, helpers::TestCategoryResult};
-use crate::error::Result;
+use std::{fmt, io::Write, time::Duration};
+
 use clap::Args;
-use std::{io::Write, time::Duration};
+use rand::Rng;
+use tokio::{
+    net::TcpStream,
+    sync::mpsc,
+    time::{Instant, timeout},
+};
+use tokio_util::sync::CancellationToken;
+
+use super::{
+    AllCategoriesResult, TestCategory, TestCategoryResult, TestConfigArgs, TestResult, TestVerdict,
+};
+use crate::{
+    duration::Duration as CliDuration,
+    error::{CliError, Result},
+};
+
+// Thresholds (from Go implementation)
+const THRESHOLD_MEASURE_AVG: Duration = Duration::from_millis(50);
+const THRESHOLD_MEASURE_POOR: Duration = Duration::from_millis(240);
+const THRESHOLD_LOAD_AVG: Duration = Duration::from_millis(50);
+const THRESHOLD_LOAD_POOR: Duration = Duration::from_millis(240);
+
+/// Validator test cases.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum ValidatorTestCase {
+    /// TCP connectivity check.
+    Ping,
+    /// TCP round-trip time measurement.
+    PingMeasure,
+    /// Sustained TCP load test.
+    PingLoad,
+}
+
+impl ValidatorTestCase {
+    /// Returns all validator test cases.
+    pub fn all() -> &'static [ValidatorTestCase] {
+        &[
+            ValidatorTestCase::Ping,
+            ValidatorTestCase::PingMeasure,
+            ValidatorTestCase::PingLoad,
+        ]
+    }
+}
+
+impl fmt::Display for ValidatorTestCase {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(match self {
+            ValidatorTestCase::Ping => "Ping",
+            ValidatorTestCase::PingMeasure => "PingMeasure",
+            ValidatorTestCase::PingLoad => "PingLoad",
+        })
+    }
+}
 
 /// Arguments for the validator test command.
 #[derive(Args, Clone, Debug)]
@@ -30,10 +82,243 @@ pub struct TestValidatorArgs {
 }
 
 /// Runs the validator client tests.
-pub async fn run(_args: TestValidatorArgs, _writer: &mut dyn Write) -> Result<TestCategoryResult> {
-    // TODO: Implement validator tests
-    // - Ping
-    // - PingMeasure
-    // - PingLoad
-    unimplemented!("validator test not yet implemented")
+pub async fn run(
+    args: TestValidatorArgs,
+    writer: &mut dyn Write,
+    ct: CancellationToken,
+) -> Result<TestCategoryResult> {
+    pluto_tracing::init(
+        &pluto_tracing::TracingConfig::builder()
+            .with_default_console()
+            .build(),
+    )
+    .expect("Failed to initialize tracing");
+
+    tracing::info!("Starting validator client test");
+
+    // Get and filter test cases
+    let queued_tests: Vec<ValidatorTestCase> = if let Some(ref filter) = args.test_config.test_cases
+    {
+        ValidatorTestCase::all()
+            .iter()
+            .filter(|tc| filter.contains(&tc.to_string()))
+            .copied()
+            .collect()
+    } else {
+        ValidatorTestCase::all().to_vec()
+    };
+
+    if queued_tests.is_empty() {
+        return Err(CliError::TestCaseNotSupported);
+    }
+
+    let start_time = Instant::now();
+    let test_results = run_tests_with_timeout(&args, &queued_tests, ct).await;
+    let elapsed = start_time.elapsed();
+
+    let score = super::calculate_score(&test_results);
+
+    let mut res = TestCategoryResult::new(TestCategory::Validator);
+    res.targets.insert(args.api_address.clone(), test_results);
+    res.execution_time = Some(CliDuration::new(elapsed));
+    res.score = Some(score);
+
+    if !args.test_config.quiet {
+        super::write_result_to_writer(&res, writer)?;
+    }
+
+    if !args.test_config.output_json.is_empty() {
+        super::write_result_to_file(&res, args.test_config.output_json.as_ref()).await?;
+    }
+
+    if args.test_config.publish {
+        let all = AllCategoriesResult {
+            validator: Some(res.clone()),
+            ..Default::default()
+        };
+        super::publish_result_to_obol_api(
+            all,
+            &args.test_config.publish_addr,
+            &args.test_config.publish_private_key_file,
+        )
+        .await?;
+    }
+
+    Ok(res)
+}
+
+/// Runs tests with timeout, keeping completed tests on timeout.
+async fn run_tests_with_timeout(
+    args: &TestValidatorArgs,
+    tests: &[ValidatorTestCase],
+    ct: CancellationToken,
+) -> Vec<TestResult> {
+    let mut results = Vec::new();
+    let start = Instant::now();
+
+    for &test_case in tests {
+        let remaining = args.test_config.timeout.saturating_sub(start.elapsed());
+
+        tokio::select! {
+            result = run_single_test(args, test_case) => {
+                results.push(result);
+            }
+            _ = tokio::time::sleep(remaining) => {
+                results.push(
+                    TestResult::new(test_case.to_string())
+                        .fail(CliError::TimeoutInterrupted),
+                );
+                break;
+            }
+            _ = ct.cancelled() => {
+                results.push(
+                    TestResult::new(test_case.to_string())
+                        .fail(CliError::TimeoutInterrupted),
+                );
+                break;
+            }
+        }
+    }
+
+    results
+}
+
+/// Runs a single test case.
+async fn run_single_test(args: &TestValidatorArgs, test_case: ValidatorTestCase) -> TestResult {
+    match test_case {
+        ValidatorTestCase::Ping => ping_test(args).await,
+        ValidatorTestCase::PingMeasure => ping_measure_test(args).await,
+        ValidatorTestCase::PingLoad => ping_load_test(args).await,
+    }
+}
+
+async fn ping_test(args: &TestValidatorArgs) -> TestResult {
+    let mut result = TestResult::new(ValidatorTestCase::Ping.to_string());
+
+    match timeout(
+        Duration::from_secs(1),
+        TcpStream::connect(&args.api_address),
+    )
+    .await
+    {
+        Ok(Ok(_conn)) => {
+            result.verdict = TestVerdict::Ok;
+        }
+        Ok(Err(e)) => {
+            return result.fail(e);
+        }
+        Err(_) => {
+            return result.fail(std::io::Error::new(
+                std::io::ErrorKind::TimedOut,
+                "connection timeout",
+            ));
+        }
+    }
+
+    result
+}
+
+async fn ping_measure_test(args: &TestValidatorArgs) -> TestResult {
+    let mut result = TestResult::new(ValidatorTestCase::PingMeasure.to_string());
+    let before = Instant::now();
+
+    match timeout(
+        Duration::from_secs(1),
+        TcpStream::connect(&args.api_address),
+    )
+    .await
+    {
+        Ok(Ok(_conn)) => {
+            let rtt = before.elapsed();
+            result =
+                super::evaluate_rtt(rtt, result, THRESHOLD_MEASURE_AVG, THRESHOLD_MEASURE_POOR);
+        }
+        Ok(Err(e)) => {
+            return result.fail(e);
+        }
+        Err(_) => {
+            return result.fail(std::io::Error::new(
+                std::io::ErrorKind::TimedOut,
+                "connection timeout",
+            ));
+        }
+    }
+
+    result
+}
+
+async fn ping_load_test(args: &TestValidatorArgs) -> TestResult {
+    tracing::info!(
+        duration = ?args.load_test_duration,
+        target = %args.api_address,
+        "Running ping load tests..."
+    );
+
+    let mut result = TestResult::new(ValidatorTestCase::PingLoad.to_string());
+    let (tx, mut rx) = mpsc::channel::<Duration>(i16::MAX as usize);
+    let address = args.api_address.clone();
+    let duration = args.load_test_duration;
+
+    {
+        let start = Instant::now();
+        let mut interval = tokio::time::interval(Duration::from_secs(1));
+        let mut workers = tokio::task::JoinSet::new();
+
+        interval.tick().await;
+        while start.elapsed() < duration {
+            interval.tick().await;
+
+            let tx = tx.clone();
+            let addr = address.clone();
+            let remaining = duration.saturating_sub(start.elapsed());
+
+            workers.spawn(ping_continuously(addr, tx, remaining));
+        }
+
+        // Drop the scheduler's clone so only workers hold senders
+        drop(tx);
+
+        // Wait for all spawned ping workers to finish
+        workers.join_all().await;
+    }
+
+    tracing::info!(target = %args.api_address, "Ping load tests finished");
+
+    // All senders dropped, collect all RTTs
+    rx.close();
+    let mut rtts = Vec::new();
+    while let Some(rtt) = rx.recv().await {
+        rtts.push(rtt);
+    }
+
+    result = super::evaluate_highest_rtt(rtts, result, THRESHOLD_LOAD_AVG, THRESHOLD_LOAD_POOR);
+
+    result
+}
+
+async fn ping_continuously(
+    address: impl AsRef<str>,
+    tx: mpsc::Sender<Duration>,
+    max_duration: Duration,
+) {
+    let address = address.as_ref();
+    let start = Instant::now();
+
+    while start.elapsed() < max_duration {
+        let before = Instant::now();
+
+        match timeout(Duration::from_secs(1), TcpStream::connect(address)).await {
+            Ok(Ok(_conn)) => {
+                let rtt = before.elapsed();
+                if tx.send(rtt).await.is_err() {
+                    return;
+                }
+            }
+            Ok(Err(_)) | Err(_) => {
+                return;
+            }
+        }
+        let sleep_ms = rand::thread_rng().gen_range(0..100);
+        tokio::time::sleep(Duration::from_millis(sleep_ms)).await;
+    }
 }

crates/cli/src/main.rs

@@ -72,7 +72,7 @@ async fn run() -> std::result::Result<(), CliError> {
                             .map(|_| ())
                     }
                     TestCommands::Validator(args) => {
-                        commands::test::validator::run(args, &mut stdout)
+                        commands::test::validator::run(args, &mut stdout, ct.clone())
                             .await
                             .map(|_| ())
                     }