sysknife-daemon 0.15.0

//! Integration tests for the daemon's automatic rollback execution path.
//!
//! When a High-risk rpm-ostree action (e.g. `UpdateSystem`) fails, the daemon
//! automatically runs `rpm-ostree rollback` and transitions the job to
//! `JobState::RolledBack`. These tests verify the full round-trip through the
//! IPC framing layer: preview → execute → job_started → job_progress (rollback
//! messages) → job_completed with `status: "rolled_back"`.
//!
//! `MockRunner` provides deterministic state collection, and
//! `FailThenRollbackExecutor` controls which commands succeed or fail so the
//! tests never depend on real system commands.

use std::io;
use std::sync::Arc;

use async_trait::async_trait;
use serde_json::{json, Value};
use sysknife_daemon::dispatcher::connection_handler_with_executor;
use sysknife_daemon::executor::{ActionExecutor, ExecutionOutput, ExecutorError};
use sysknife_daemon::state::{DaemonConfig, DaemonState};
use sysknife_daemon::state_collector::CommandRunner;
use sysknife_daemon::transport::{framing::FramedStream, listen::ListenTarget};
use sysknife_types::CallerRole;
use tempfile::tempdir;
use tokio::net::UnixStream;

// ---------------------------------------------------------------------------
// Test doubles
// ---------------------------------------------------------------------------

/// State-collection mock: returns deterministic hostname, empty everything else.
struct MockRunner;

impl CommandRunner for MockRunner {
    fn run(&self, program: &str, _args: &[&str]) -> Result<String, io::Error> {
        match program {
            "hostname" => Ok("rollback-test-host\n".to_string()),
            "rpm-ostree" => Ok("{}".to_string()),
            _ => Ok(String::new()),
        }
    }
}

/// Action executor that FAILS the primary action (simulating a failed
/// `rpm-ostree upgrade`) but SUCCEEDS the rollback (`rpm-ostree rollback`).
///
/// This is injected into `connection_handler_with_executor` so that
/// `attempt_rollback_if_needed` calls `executor.execute(&rb_spec)` on the
/// mock instead of spawning a real `rpm-ostree` process.
struct FailThenRollbackExecutor;

#[async_trait]
impl ActionExecutor for FailThenRollbackExecutor {
    async fn execute(
        &self,
        spec: &sysknife_daemon::actions::ActionSpec,
    ) -> Result<ExecutionOutput, ExecutorError> {
        match spec.action_name {
            // The rollback command succeeds.
            "RollbackDeployment" => Ok(ExecutionOutput {
                stdout: "Rollback successful\n".to_string(),
                stderr: String::new(),
                exit_code: 0,
            }),
            // All other actions fall through to RealActionExecutor behavior,
            // but we should not reach here for the rollback test because the
            // primary action is executed via stream_command_with_progress
            // (which spawns a real process). This arm is a safety net.
            _ => Ok(ExecutionOutput {
                stdout: String::new(),
                stderr: "mock: unexpected action".to_string(),
                exit_code: 99,
            }),
        }
    }
}

/// Action executor where BOTH the primary action AND the rollback fail.
struct FailBothExecutor;

#[async_trait]
impl ActionExecutor for FailBothExecutor {
    async fn execute(
        &self,
        _spec: &sysknife_daemon::actions::ActionSpec,
    ) -> Result<ExecutionOutput, ExecutorError> {
        Ok(ExecutionOutput {
            stdout: String::new(),
            stderr: "mock: command failed".to_string(),
            exit_code: 1,
        })
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Spawn a daemon with the given executor and return the client-side
/// `FramedStream`. The daemon task runs in the background until the client
/// drops its end of the socket pair.
async fn spawn_handler_with_executor(
    state: DaemonState,
    executor: Arc<dyn ActionExecutor>,
) -> FramedStream<UnixStream> {
    let (client, server) = UnixStream::pair().unwrap();
    let runner: Arc<dyn CommandRunner + Send + Sync> = Arc::new(MockRunner);
    tokio::spawn(async move {
        connection_handler_with_executor(
            server,
            state,
            runner,
            executor,
            sysknife_daemon::auth::CallerAttribution::from_peer_uid(1000, CallerRole::Admin),
        )
        .await;
    });
    FramedStream::new(client)
}

fn test_state(dir: &tempfile::TempDir) -> DaemonState {
    let db_path = dir.path().join("rollback-test.db");
    let sock_path = dir.path().join("rollback-test.sock");
    let config = DaemonConfig::new(ListenTarget::Unix(sock_path), db_path);
    // These tests drive rpm-ostree-shaped actions, so they must declare a
    // Fedora-family host. `DaemonState::open` detects the *real* host, so they
    // used to exercise a Fedora action against whatever the test machine ran —
    // passing only because the family fence did not yet cover those actions. It
    // does now, and the fence is right: `rpm-ostree` cannot run on an apt host.
    let mut state = DaemonState::open(config).unwrap();
    state.host_distro = Some(sysknife_core::distro::DistroId::FedoraSilverblue { version: 41 });
    state
}

/// Preview and explicitly approve UpdateSystem.
async fn preview_update_system(framed: &mut FramedStream<UnixStream>) -> (String, String) {
    let preview_req = json!({
        "type": "preview",
        "request_id": "rollback-preview",
        "action_name": "UpdateSystem",
        "params": {}
    });
    framed
        .send(&serde_json::to_vec(&preview_req).unwrap())
        .await
        .unwrap();

    let raw = framed.recv().await.unwrap();
    let resp: Value = serde_json::from_slice(&raw).unwrap();
    assert_eq!(
        resp["type"], "preview_response",
        "expected preview_response, got: {resp}"
    );
    let transaction_id = resp["transaction_id"].as_str().unwrap().to_string();
    let approve_req = json!({
        "type": "approve",
        "request_id": "rollback-approve",
        "transaction_id": transaction_id,
    });
    framed
        .send(&serde_json::to_vec(&approve_req).unwrap())
        .await
        .unwrap();
    let raw = framed.recv().await.unwrap();
    let approval: Value = serde_json::from_slice(&raw).unwrap();
    assert_eq!(approval["type"], "approval_response");
    (
        transaction_id,
        approval["approval_receipt"].as_str().unwrap().to_string(),
    )
}

async fn execute_update_system(
    framed: &mut FramedStream<UnixStream>,
    transaction_id: &str,
    approval_receipt: &str,
) {
    let exec_req = json!({
        "type": "execute",
        "request_id": "rollback-execute",
        "transaction_id": transaction_id,
        "action_name": "UpdateSystem",
        "params": {},
        "approval_receipt": approval_receipt
    });
    framed
        .send(&serde_json::to_vec(&exec_req).unwrap())
        .await
        .unwrap();
}

/// Drain frames from the client, collecting all messages until job_completed.
/// Returns (all_messages, job_completed_message).
async fn drain_until_completed(framed: &mut FramedStream<UnixStream>) -> (Vec<Value>, Value) {
    let mut messages = Vec::new();
    loop {
        let raw = framed.recv().await.unwrap();
        let msg: Value = serde_json::from_slice(&raw).unwrap();
        let msg_type = msg["type"].as_str().unwrap().to_string();
        messages.push(msg.clone());
        if msg_type == "job_completed" {
            return (messages, msg);
        }
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

/// The primary action (`rpm-ostree upgrade`) fails with a non-zero exit code.
/// Because `UpdateSystem` has `rollback_available: true`, the daemon attempts
/// `rpm-ostree rollback` via the executor. The mock executor succeeds for
/// `RollbackDeployment`, so the final status must be `"rolled_back"`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn failed_update_system_triggers_automatic_rollback() {
    let dir = tempdir().unwrap();
    let state = test_state(&dir);
    let executor: Arc<dyn ActionExecutor> = Arc::new(FailThenRollbackExecutor);
    let mut framed = spawn_handler_with_executor(state, executor).await;

    // Step 1: Preview to get the request_hash.
    let (transaction_id, receipt) = preview_update_system(&mut framed).await;

    // Step 2: Execute — the primary action will fail because `rpm-ostree upgrade`
    // is not available in CI, triggering the rollback path.
    execute_update_system(&mut framed, &transaction_id, &receipt).await;

    // Step 3: Drain frames until job_completed.
    let (messages, completed) = drain_until_completed(&mut framed).await;

    // Verify job_started was the first frame.
    assert_eq!(
        messages[0]["type"], "job_started",
        "first frame must be job_started, got: {:?}",
        messages[0]
    );

    // Verify the final status is "rolled_back".
    let status = completed["result"]["status"].as_str().unwrap();
    assert_eq!(
        status, "rolled_back",
        "expected rolled_back status, got: {status}; full message: {completed}"
    );

    // Verify the rollback_ref is set.
    let rollback_ref = completed["result"]["rollback_ref"].as_str().unwrap();
    assert_eq!(
        rollback_ref, "rpm-ostree rollback",
        "rollback_ref should indicate the rollback command used"
    );

    // Verify that progress frames include rollback announcement.
    let progress_lines: Vec<&str> = messages
        .iter()
        .filter(|m| m["type"] == "job_progress")
        .filter_map(|m| m["line"].as_str())
        .collect();
    assert!(
        progress_lines
            .iter()
            .any(|l| l.contains("attempting automatic rollback")),
        "progress should announce rollback attempt; got lines: {progress_lines:?}"
    );
    assert!(
        progress_lines
            .iter()
            .any(|l| l.contains("Rollback succeeded")),
        "progress should announce rollback success; got lines: {progress_lines:?}"
    );
}

/// Verify that the transaction store records the RolledBack status after
/// a successful automatic rollback.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn rollback_updates_transaction_store_to_rolled_back() {
    let dir = tempdir().unwrap();
    let state = test_state(&dir);
    let store = std::sync::Arc::clone(&state.audit);
    let executor: Arc<dyn ActionExecutor> = Arc::new(FailThenRollbackExecutor);
    let mut framed = spawn_handler_with_executor(state, executor).await;

    let (transaction_id, receipt) = preview_update_system(&mut framed).await;
    execute_update_system(&mut framed, &transaction_id, &receipt).await;
    let (messages, _completed) = drain_until_completed(&mut framed).await;

    // Extract the transaction_id from job_started.
    let transaction_id = messages[0]["transaction_id"]
        .as_str()
        .expect("job_started must contain transaction_id")
        .to_string();

    // Verify the store records the final status as RolledBack.
    let tx = store.get(&transaction_id).await.unwrap().unwrap();
    assert_eq!(
        tx.status,
        sysknife_types::JobState::RolledBack,
        "transaction store must record RolledBack; got: {:?}",
        tx.status
    );
}

/// When both the primary action AND the rollback fail, the final status
/// should be `"failed"` (not `"rolled_back"`), and the rollback_ref should
/// be null.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn failed_rollback_leaves_status_as_failed() {
    let dir = tempdir().unwrap();
    let state = test_state(&dir);
    let executor: Arc<dyn ActionExecutor> = Arc::new(FailBothExecutor);
    let mut framed = spawn_handler_with_executor(state, executor).await;

    let (transaction_id, receipt) = preview_update_system(&mut framed).await;
    execute_update_system(&mut framed, &transaction_id, &receipt).await;
    let (messages, completed) = drain_until_completed(&mut framed).await;

    let status = completed["result"]["status"].as_str().unwrap();
    assert_eq!(
        status, "failed",
        "when rollback also fails, status must remain 'failed'; got: {status}"
    );

    assert!(
        completed["result"]["rollback_ref"].is_null(),
        "rollback_ref must be null when rollback fails; got: {:?}",
        completed["result"]["rollback_ref"]
    );

    // Verify progress mentions the rollback failure.
    let progress_lines: Vec<&str> = messages
        .iter()
        .filter(|m| m["type"] == "job_progress")
        .filter_map(|m| m["line"].as_str())
        .collect();
    assert!(
        progress_lines
            .iter()
            .any(|l| l.contains("Rollback also failed")),
        "progress should announce rollback failure; got lines: {progress_lines:?}"
    );
}

/// Actions without `rollback_available: true` must NOT trigger rollback,
/// even if they fail. `GetMemoryInfo` is a Low-risk read-only action.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn non_rollbackable_action_does_not_trigger_rollback() {
    let dir = tempdir().unwrap();
    let state = test_state(&dir);
    // Use FailBothExecutor — if rollback were attempted, it would appear in
    // the progress frames, which we assert must not happen.
    let executor: Arc<dyn ActionExecutor> = Arc::new(FailBothExecutor);
    let mut framed = spawn_handler_with_executor(state, executor).await;

    // Preview GetMemoryInfo (Low risk, no rollback).
    let preview_req = json!({
        "type": "preview",
        "request_id": "no-rollback-preview",
        "action_name": "GetMemoryInfo",
        "params": {}
    });
    framed
        .send(&serde_json::to_vec(&preview_req).unwrap())
        .await
        .unwrap();
    let raw = framed.recv().await.unwrap();
    let resp: Value = serde_json::from_slice(&raw).unwrap();
    assert_eq!(resp["type"], "preview_response");
    let transaction_id = resp["transaction_id"].as_str().unwrap().to_string();
    framed
        .send(
            &serde_json::to_vec(&json!({
                "type": "approve",
                "request_id": "no-rollback-approve",
                "transaction_id": transaction_id,
            }))
            .unwrap(),
        )
        .await
        .unwrap();
    let approval: Value = serde_json::from_slice(&framed.recv().await.unwrap()).unwrap();
    let receipt = approval["approval_receipt"].as_str().unwrap();

    // Execute.
    let exec_req = json!({
        "type": "execute",
        "request_id": "no-rollback-execute",
        "transaction_id": transaction_id,
        "action_name": "GetMemoryInfo",
        "params": {},
        "approval_receipt": receipt
    });
    framed
        .send(&serde_json::to_vec(&exec_req).unwrap())
        .await
        .unwrap();

    let (_messages, completed) = drain_until_completed(&mut framed).await;

    let status = completed["result"]["status"].as_str().unwrap();
    // GetMemoryInfo runs `free -h`, which is present on every host, so this
    // normally succeeds. Either way the point of the test is that rollback is
    // NOT triggered: accept "succeeded" or "failed", just not "rolled_back".
    assert_ne!(
        status, "rolled_back",
        "non-rollbackable action must never produce rolled_back status"
    );

    assert!(
        completed["result"]["rollback_ref"].is_null(),
        "rollback_ref must be null for non-rollbackable actions"
    );
}

/// Executor whose primary action reports `ActionNotStopped` — the timeout could
/// not confirm the privileged work stopped — and which counts any rollback the
/// dispatcher attempts.
///
/// This is the wiring the two dispatcher unit tests cannot reach: they call
/// `attempt_rollback_if_needed` directly and pass the veto boolean by hand, so
/// they prove the function body but not that `handle_execute` derives the
/// boolean from the executor error at all. A mutant that hardcodes the boolean
/// to `false` survives every other test; this one kills it.
struct NotStoppedExecutor {
    rollbacks: std::sync::atomic::AtomicUsize,
}

#[async_trait]
impl ActionExecutor for NotStoppedExecutor {
    async fn execute(
        &self,
        spec: &sysknife_daemon::actions::ActionSpec,
    ) -> Result<ExecutionOutput, ExecutorError> {
        if spec.action_name == "RollbackDeployment" {
            self.rollbacks
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            return Ok(ExecutionOutput {
                stdout: "Rollback successful\n".to_string(),
                stderr: String::new(),
                exit_code: 0,
            });
        }
        // The primary action: the deadline passed and the group could not be
        // confirmed stopped. Every real privileged action reaches the executor
        // through `execute_with_progress`, whose default forwards to this.
        Err(ExecutorError::ActionNotStopped {
            program: "rpm-ostree".to_string(),
            pgid: 4242,
            timeout_secs: 7200,
        })
    }
}

/// End to end through the real dispatcher: an action that could not be confirmed
/// stopped must not be rolled back over, and the operator must be told why.
///
/// This is the test that proves the veto is wired, not just implemented. It
/// drives preview -> execute and asserts the rollback executor was never called,
/// the job stays failed, no rollback_ref is set, and a progress frame explains
/// the skip. See issue #140.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_action_that_could_not_be_stopped_is_not_rolled_back_over_end_to_end() {
    let dir = tempdir().unwrap();
    let state = test_state(&dir);
    let counting = Arc::new(NotStoppedExecutor {
        rollbacks: std::sync::atomic::AtomicUsize::new(0),
    });
    let executor: Arc<dyn ActionExecutor> = counting.clone();
    let mut framed = spawn_handler_with_executor(state, executor).await;

    let (transaction_id, receipt) = preview_update_system(&mut framed).await;
    execute_update_system(&mut framed, &transaction_id, &receipt).await;
    let (messages, completed) = drain_until_completed(&mut framed).await;

    assert_eq!(
        counting.rollbacks.load(std::sync::atomic::Ordering::SeqCst),
        0,
        "no rollback may run while the original action may still be in flight"
    );
    let status = completed["result"]["status"].as_str().unwrap();
    assert_eq!(
        status, "failed",
        "the job stays failed, not rolled_back; got: {status}"
    );
    assert!(
        completed["result"]["rollback_ref"].is_null(),
        "no rollback happened, so there is no rollback_ref"
    );

    let progress: Vec<&str> = messages
        .iter()
        .filter(|m| m["type"] == "job_progress")
        .filter_map(|m| m["line"].as_str())
        .collect();
    assert!(
        progress
            .iter()
            .any(|l| l.contains("could not be confirmed stopped")
                && l.contains("Inspect the host")),
        "the operator must be told the rollback was skipped and to inspect the host; got: {progress:?}"
    );
}