test(admin): cover real backend ILM job commands

crates/cli/tests/admin_ilm_real_backend.rs

@@ -0,0 +1,228 @@
+#![cfg(not(windows))]
+
+mod admin_support;
+
+use std::net::{SocketAddr, TcpListener};
+use std::path::{Path, PathBuf};
+use std::process::{Child, Command, Output, Stdio};
+use std::thread;
+use std::time::{Duration, Instant};
+
+use admin_support::{rc_binary, rc_host_alias};
+use serde_json::Value;
+
+const ACCESS_KEY: &str = "ACCESS_KEY";
+const SECRET_KEY: &str = "SECRET_KEY";
+const ALIAS: &str = "realilm";
+const BUCKET: &str = "rc-manual-transition-real-backend";
+const RUSTFS_BINARY_ENV: &str = "RC_REAL_RUSTFS_BINARY";
+
+struct RustfsProcess {
+    child: Child,
+}
+
+impl Drop for RustfsProcess {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }
+}
+
+#[test]
+#[ignore = "requires RC_REAL_RUSTFS_BINARY pointing at a rustfs binary"]
+fn ilm_transition_status_wait_cancel_hits_real_rustfs_backend() {
+    let Some(rustfs_binary) = std::env::var_os(RUSTFS_BINARY_ENV).map(PathBuf::from) else {
+        eprintln!("skipping real backend ILM transition CLI e2e; set {RUSTFS_BINARY_ENV}");
+        return;
+    };
+    assert!(
+        rustfs_binary.is_file(),
+        "{RUSTFS_BINARY_ENV} must point at a rustfs binary: {}",
+        rustfs_binary.display()
+    );
+
+    let config_dir = tempfile::tempdir().expect("create rc config dir");
+    let data_dir = tempfile::tempdir().expect("create rustfs data dir");
+    let address = free_loopback_addr();
+    let endpoint = format!("http://{address}");
+    let mut rustfs = start_rustfs(&rustfs_binary, &address, data_dir.path());
+    wait_for_rc_ready(config_dir.path(), &endpoint);
+
+    run_rc_success(
+        config_dir.path(),
+        &endpoint,
+        &["mb", &format!("{ALIAS}/{BUCKET}")],
+        "create test bucket",
+    );
+    let run = run_rc_success(
+        config_dir.path(),
+        &endpoint,
+        &[
+            "--json",
+            "admin",
+            "ilm",
+            "transition",
+            "run",
+            ALIAS,
+            BUCKET,
+            "--async",
+            "--max-objects",
+            "1",
+        ],
+        "start async manual transition job",
+    );
+    let run_json: Value = serde_json::from_slice(&run.stdout).expect("run stdout should be JSON");
+    assert_eq!(run_json["type"], "manual_transition_run");
+    assert_eq!(run_json["data"]["state"], "accepted");
+    assert_eq!(run_json["data"]["mode"], "durable_job");
+    let job_id = run_json["data"]["job_id"]
+        .as_str()
+        .expect("async run must return job_id");
+    assert!(!job_id.is_empty(), "async run must return non-empty job_id");
+    assert_eq!(
+        run_json["data"]["status_endpoint"],
+        run_json["data"]["cancel_endpoint"]
+    );
+
+    let status = run_rc_success(
+        config_dir.path(),
+        &endpoint,
+        &[
+            "--json",
+            "admin",
+            "ilm",
+            "transition",
+            "status",
+            ALIAS,
+            job_id,
+        ],
+        "read manual transition job status",
+    );
+    let status_json: Value =
+        serde_json::from_slice(&status.stdout).expect("status stdout should be JSON");
+    assert_eq!(status_json["type"], "manual_transition_job_status");
+    assert_eq!(status_json["data"]["job_id"], job_id);
+
+    let wait = run_rc_success(
+        config_dir.path(),
+        &endpoint,
+        &[
+            "--json",
+            "admin",
+            "ilm",
+            "transition",
+            "wait",
+            ALIAS,
+            job_id,
+            "--poll-interval-seconds",
+            "1",
+            "--timeout-seconds",
+            "30",
+        ],
+        "wait for manual transition job",
+    );
+    let wait_json: Value =
+        serde_json::from_slice(&wait.stdout).expect("wait stdout should be JSON");
+    assert_eq!(wait_json["type"], "manual_transition_job_wait");
+    assert_eq!(wait_json["data"]["job_id"], job_id);
+    assert!(
+        matches!(
+            wait_json["data"]["status"].as_str(),
+            Some("completed" | "partial")
+        ),
+        "wait should return a successful terminal state: {wait_json}"
+    );
+
+    let cancel = run_rc_success(
+        config_dir.path(),
+        &endpoint,
+        &[
+            "--json",
+            "admin",
+            "ilm",
+            "transition",
+            "cancel",
+            ALIAS,
+            job_id,
+        ],
+        "cancel/query terminal manual transition job",
+    );
+    let cancel_json: Value =
+        serde_json::from_slice(&cancel.stdout).expect("cancel stdout should be JSON");
+    assert_eq!(cancel_json["type"], "manual_transition_job_cancel");
+    assert_eq!(cancel_json["data"]["job_id"], job_id);
+    assert!(
+        matches!(
+            cancel_json["data"]["status"].as_str(),
+            Some("completed" | "partial" | "failed" | "unknown")
+        ),
+        "terminal cancel must not rewrite a finished job into cancelled: {cancel_json}"
+    );
+
+    let _ = rustfs.child.kill();
+}
+
+fn start_rustfs(binary: &Path, address: &SocketAddr, data_dir: &Path) -> RustfsProcess {
+    let child = Command::new(binary)
+        .args([
+            "--address",
+            &address.to_string(),
+            "--access-key",
+            ACCESS_KEY,
+            "--secret-key",
+            SECRET_KEY,
+            data_dir
+                .to_str()
+                .expect("temp data dir path should be UTF-8"),
+        ])
+        .env("RUSTFS_CONSOLE_ENABLE", "false")
+        .env("RUSTFS_SCANNER_ENABLED", "false")
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .spawn()
+        .expect("start rustfs process");
+    RustfsProcess { child }
+}
+
+fn wait_for_rc_ready(config_dir: &Path, endpoint: &str) {
+    let deadline = Instant::now() + Duration::from_secs(60);
+    loop {
+        let output = run_rc(config_dir, endpoint, &["ls", ALIAS]);
+        if output.status.success() {
+            return;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "rustfs backend did not become ready: stdout={}, stderr={}",
+            String::from_utf8_lossy(&output.stdout),
+            String::from_utf8_lossy(&output.stderr)
+        );
+        thread::sleep(Duration::from_millis(250));
+    }
+}
+
+fn run_rc_success(config_dir: &Path, endpoint: &str, args: &[&str], context: &str) -> Output {
+    let output = run_rc(config_dir, endpoint, args);
+    assert!(
+        output.status.success(),
+        "{context} failed: status={:?}, stdout={}, stderr={}",
+        output.status.code(),
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+    output
+}
+
+fn run_rc(config_dir: &Path, endpoint: &str, args: &[&str]) -> Output {
+    Command::new(rc_binary())
+        .args(args)
+        .env("RC_CONFIG_DIR", config_dir)
+        .env(format!("RC_HOST_{ALIAS}"), rc_host_alias(endpoint))
+        .output()
+        .expect("run rc command")
+}
+
+fn free_loopback_addr() -> SocketAddr {
+    let listener = TcpListener::bind("127.0.0.1:0").expect("bind free loopback port");
+    listener.local_addr().expect("read loopback address")
+}