zlayer-agent 0.11.11

//! Docker runtime integration tests
//!
//! These tests verify the `DockerRuntime` implementation against a real Docker daemon.
//! Tests are gated behind the `docker` feature and will be skipped if Docker is not available.
//!
//! # Requirements
//! - Docker daemon must be running
//! - The `docker` feature must be enabled
//!
//! # Running
//! ```bash
//! cargo test -p zlayer-agent --features docker -- --nocapture
//! ```

#![cfg(feature = "docker")]

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use zlayer_agent::runtimes::DockerRuntime;
use zlayer_agent::{ContainerId, ContainerState, Runtime};
use zlayer_spec::{
    CommandSpec, ErrorsSpec, HealthCheck, HealthSpec, ImageSpec, InitSpec, NodeMode, PullPolicy,
    ResourceType, ResourcesSpec, ScaleSpec, ServiceNetworkSpec, ServiceSpec, ServiceType,
};

// =============================================================================
// Test Configuration
// =============================================================================

/// Test image - alpine is small and fast to pull
const TEST_IMAGE: &str = "alpine:latest";

/// Timeout for operations that might take a while (like pulling images)
const LONG_TIMEOUT: Duration = Duration::from_secs(120);

/// Timeout for quick operations
const SHORT_TIMEOUT: Duration = Duration::from_secs(30);

// =============================================================================
// Helper Functions
// =============================================================================

/// Attempt to connect to Docker. Returns None if Docker is not available.
async fn skip_if_no_docker() -> Option<DockerRuntime> {
    DockerRuntime::new(None).await.ok()
}

/// Generate a unique container name with random suffix to avoid conflicts
fn unique_container_name(prefix: &str) -> String {
    use rand::Rng;
    let suffix: u32 = rand::rng().random_range(10000..99999);
    #[allow(clippy::cast_possible_truncation)]
    let timestamp = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis() as u64
        % 1_000_000;
    format!("test-{prefix}-{timestamp}-{suffix}")
}

/// Create a `ContainerId` with a unique service name
fn unique_container_id(prefix: &str) -> ContainerId {
    ContainerId {
        service: unique_container_name(prefix),
        replica: 1,
    }
}

/// Create a minimal `ServiceSpec` for testing
fn create_test_spec(image: &str) -> ServiceSpec {
    ServiceSpec {
        rtype: ResourceType::Service,
        schedule: None,
        image: ImageSpec {
            name: image.parse().expect("valid image reference"),
            pull_policy: PullPolicy::IfNotPresent,
        },
        resources: ResourcesSpec::default(),
        env: HashMap::new(),
        command: CommandSpec::default(),
        network: ServiceNetworkSpec::default(),
        endpoints: vec![],
        scale: ScaleSpec::default(),
        depends: vec![],
        health: HealthSpec {
            start_grace: None,
            interval: None,
            timeout: None,
            retries: 3,
            check: HealthCheck::Tcp { port: 0 },
        },
        init: InitSpec::default(),
        errors: ErrorsSpec::default(),
        lifecycle: zlayer_spec::LifecycleSpec::default(),
        devices: vec![],
        storage: vec![],
        port_mappings: vec![],
        capabilities: vec![],
        cap_drop: vec![],
        privileged: false,
        node_mode: NodeMode::default(),
        node_selector: None,
        service_type: ServiceType::default(),
        wasm: None,
        logs: None,
        host_network: false,
        hostname: None,
        dns: Vec::new(),
        extra_hosts: Vec::new(),
        restart_policy: None,
        labels: HashMap::new(),
        user: None,
        stop_signal: None,
        stop_grace_period: None,
        sysctls: HashMap::new(),
        ulimits: HashMap::new(),
        security_opt: Vec::new(),
        pid_mode: None,
        ipc_mode: None,
        network_mode: zlayer_spec::NetworkMode::default(),
        extra_groups: Vec::new(),
        read_only_root_fs: false,
        init_container: None,
        platform: None,
        tty: false,
        stdin_open: false,
        userns_mode: None,
        cgroup_parent: None,
        expose: Vec::new(),
    }
}

/// Create a `ServiceSpec` with a command that outputs to stdout and exits
fn create_echo_spec(message: &str) -> ServiceSpec {
    let mut spec = create_test_spec(TEST_IMAGE);
    spec.command = CommandSpec {
        entrypoint: None,
        args: Some(vec![
            "sh".to_string(),
            "-c".to_string(),
            format!("echo '{}'", message),
        ]),
        workdir: None,
    };
    spec
}

/// Create a `ServiceSpec` that sleeps for the specified number of seconds
fn create_sleep_spec(seconds: u32) -> ServiceSpec {
    let mut spec = create_test_spec(TEST_IMAGE);
    spec.command = CommandSpec {
        entrypoint: None,
        args: Some(vec!["sleep".to_string(), seconds.to_string()]),
        workdir: None,
    };
    spec
}

/// RAII guard that ensures container cleanup even on test failure
struct ContainerGuard {
    runtime: Arc<DockerRuntime>,
    id: ContainerId,
}

impl ContainerGuard {
    fn new(runtime: Arc<DockerRuntime>, id: ContainerId) -> Self {
        Self { runtime, id }
    }
}

impl Drop for ContainerGuard {
    fn drop(&mut self) {
        let runtime = self.runtime.clone();
        let id = self.id.clone();

        // Use tokio::task::block_in_place to run async cleanup in drop
        // This ensures cleanup happens even if the test panics
        if let Ok(handle) = tokio::runtime::Handle::try_current() {
            handle.spawn(async move {
                // Try to stop the container (ignore errors - it might already be stopped)
                let _ = runtime.stop_container(&id, Duration::from_secs(5)).await;
                // Force remove the container
                let _ = runtime.remove_container(&id).await;
            });
        }
    }
}

/// Wait for a container to reach the expected state
async fn wait_for_state(
    runtime: &DockerRuntime,
    id: &ContainerId,
    expected: ContainerState,
    timeout: Duration,
) -> Result<ContainerState, String> {
    let start = std::time::Instant::now();
    let poll_interval = Duration::from_millis(100);

    while start.elapsed() < timeout {
        match runtime.container_state(id).await {
            Ok(state) => {
                // For Exited state, match the variant regardless of exit code
                match (&state, &expected) {
                    (ContainerState::Exited { .. }, ContainerState::Exited { .. }) => {
                        return Ok(state);
                    }
                    _ if state == expected => return Ok(state),
                    _ => {}
                }
            }
            Err(e) => {
                // Container might not exist yet, keep waiting
                if start.elapsed() > Duration::from_secs(5) {
                    return Err(format!("Error getting container state: {e}"));
                }
            }
        }
        tokio::time::sleep(poll_interval).await;
    }

    Err(format!(
        "Timeout waiting for container {id:?} to reach state {expected:?}"
    ))
}

// =============================================================================
// Tests
// =============================================================================

/// Test that we can connect to the Docker daemon
#[tokio::test]
async fn test_docker_connection() {
    let Some(_runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };

    // If we get here, connection was successful
    println!("Successfully connected to Docker daemon");
}

/// Test pulling an image with default policy (`IfNotPresent`)
#[tokio::test]
async fn test_pull_image() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };

    println!("Pulling image: {TEST_IMAGE}");

    let result = tokio::time::timeout(LONG_TIMEOUT, runtime.pull_image(TEST_IMAGE)).await;

    match result {
        Ok(Ok(())) => println!("Image pulled successfully"),
        Ok(Err(e)) => panic!("Failed to pull image: {e}"),
        Err(e) => panic!("Timeout pulling image: {e}"),
    }
}

/// Test that `IfNotPresent` policy skips pulling when image exists
#[tokio::test]
async fn test_pull_image_if_not_present() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };

    // First, ensure the image exists by pulling it
    println!("Ensuring image exists: {TEST_IMAGE}");
    let _ = tokio::time::timeout(LONG_TIMEOUT, runtime.pull_image(TEST_IMAGE)).await;

    // Now pull with IfNotPresent - this should be instant since image exists
    println!("Testing IfNotPresent policy (should skip pull)");
    let start = std::time::Instant::now();

    let result = tokio::time::timeout(
        SHORT_TIMEOUT,
        runtime.pull_image_with_policy(TEST_IMAGE, PullPolicy::IfNotPresent, None),
    )
    .await;

    let elapsed = start.elapsed();
    println!("IfNotPresent pull completed in {elapsed:?}");

    match result {
        Ok(Ok(())) => {
            // Should complete quickly since image is already present
            assert!(
                elapsed < Duration::from_secs(5),
                "IfNotPresent should be fast for existing images"
            );
            println!("IfNotPresent correctly skipped pull");
        }
        Ok(Err(e)) => panic!("Failed with IfNotPresent policy: {e}"),
        Err(e) => panic!("Timeout with IfNotPresent policy: {e}"),
    }
}

/// Test complete container lifecycle: create -> start -> get state -> stop -> remove
#[tokio::test]
async fn test_container_lifecycle() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };
    let runtime = Arc::new(runtime);

    // Ensure image is available
    let _ = tokio::time::timeout(LONG_TIMEOUT, runtime.pull_image(TEST_IMAGE)).await;

    let id = unique_container_id("lifecycle");
    let spec = create_sleep_spec(300); // Sleep for 5 minutes (we'll stop it early)

    // Setup cleanup guard
    let _guard = ContainerGuard::new(runtime.clone(), id.clone());

    // 1. Create container
    println!("Creating container: {}", id.service);
    let create_result = runtime.create_container(&id, &spec).await;
    assert!(
        create_result.is_ok(),
        "Failed to create container: {create_result:?}"
    );

    // 2. Verify container exists and is in Pending state
    let state = runtime.container_state(&id).await;
    assert!(state.is_ok(), "Failed to get container state: {state:?}");
    assert_eq!(
        state.unwrap(),
        ContainerState::Pending,
        "Container should be Pending after create"
    );

    // 3. Start container
    println!("Starting container: {}", id.service);
    let start_result = runtime.start_container(&id).await;
    assert!(
        start_result.is_ok(),
        "Failed to start container: {start_result:?}"
    );

    // 4. Wait for Running state
    let wait_result = wait_for_state(&runtime, &id, ContainerState::Running, SHORT_TIMEOUT).await;
    assert!(
        wait_result.is_ok(),
        "Container did not reach Running state: {}",
        wait_result.unwrap_err()
    );
    println!("Container is running");

    // 5. Stop container
    println!("Stopping container: {}", id.service);
    let stop_result = runtime.stop_container(&id, Duration::from_secs(10)).await;
    assert!(
        stop_result.is_ok(),
        "Failed to stop container: {stop_result:?}"
    );

    // 6. Verify container is stopped (Exited state)
    let state = runtime.container_state(&id).await;
    assert!(state.is_ok(), "Failed to get container state after stop");
    match state.unwrap() {
        ContainerState::Exited { code } => {
            println!("Container exited with code: {code}");
        }
        other => panic!("Expected Exited state, got: {other:?}"),
    }

    // 7. Remove container
    println!("Removing container: {}", id.service);
    let remove_result = runtime.remove_container(&id).await;
    assert!(
        remove_result.is_ok(),
        "Failed to remove container: {remove_result:?}"
    );

    // 8. Verify container is gone
    let state = runtime.container_state(&id).await;
    assert!(state.is_err(), "Container should not exist after removal");
    println!("Container successfully removed");
}

/// Test container logs retrieval
#[tokio::test]
async fn test_container_logs() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };
    let runtime = Arc::new(runtime);

    // Ensure image is available
    let _ = tokio::time::timeout(LONG_TIMEOUT, runtime.pull_image(TEST_IMAGE)).await;

    let id = unique_container_id("logs");
    let test_message = "Hello from ZLayer Docker test!";
    let spec = create_echo_spec(test_message);

    // Setup cleanup guard
    let _guard = ContainerGuard::new(runtime.clone(), id.clone());

    // Create and start container
    println!("Creating container that outputs: {test_message}");
    runtime
        .create_container(&id, &spec)
        .await
        .expect("Failed to create container");
    runtime
        .start_container(&id)
        .await
        .expect("Failed to start container");

    // Wait for container to exit (it will exit quickly after echoing)
    let _ = wait_for_state(
        &runtime,
        &id,
        ContainerState::Exited { code: 0 },
        SHORT_TIMEOUT,
    )
    .await;

    // Get logs
    println!("Retrieving container logs");
    let logs = runtime
        .container_logs(&id, 100)
        .await
        .expect("Failed to get logs");

    let logs_text: String = logs
        .iter()
        .map(ToString::to_string)
        .collect::<Vec<_>>()
        .join("\n");
    println!("Logs: {logs_text}");
    assert!(
        logs.iter().any(|e| e.message.contains(test_message)),
        "Logs should contain the test message"
    );

    // Also test get_logs (returns Vec<LogEntry>)
    let log_lines = runtime
        .get_logs(&id)
        .await
        .expect("Failed to get log lines");
    println!("Log lines: {log_lines:?}");
    assert!(
        log_lines
            .iter()
            .any(|entry| entry.message.contains(test_message)),
        "Log lines should contain the test message"
    );
}

/// Test executing a command inside a running container
#[tokio::test]
async fn test_container_exec() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };
    let runtime = Arc::new(runtime);

    // Ensure image is available
    let _ = tokio::time::timeout(LONG_TIMEOUT, runtime.pull_image(TEST_IMAGE)).await;

    let id = unique_container_id("exec");
    let spec = create_sleep_spec(300); // Long-running container

    // Setup cleanup guard
    let _guard = ContainerGuard::new(runtime.clone(), id.clone());

    // Create and start container
    println!("Creating long-running container for exec test");
    runtime
        .create_container(&id, &spec)
        .await
        .expect("Failed to create container");
    runtime
        .start_container(&id)
        .await
        .expect("Failed to start container");

    // Wait for container to be running
    wait_for_state(&runtime, &id, ContainerState::Running, SHORT_TIMEOUT)
        .await
        .expect("Container did not start");

    // Execute a command
    println!("Executing command inside container");
    let cmd = vec!["echo".to_string(), "exec test output".to_string()];
    let (exit_code, stdout, stderr) = runtime.exec(&id, &cmd).await.expect("Failed to exec");

    println!("Exec result - exit_code: {exit_code}, stdout: {stdout}, stderr: {stderr}");

    assert_eq!(exit_code, 0, "Exec should succeed with exit code 0");
    assert!(
        stdout.contains("exec test output"),
        "Stdout should contain the echoed message"
    );

    // Test a command that writes to stderr
    println!("Testing command that writes to stderr");
    let cmd = vec![
        "sh".to_string(),
        "-c".to_string(),
        "echo 'stderr output' >&2".to_string(),
    ];
    let (exit_code, stdout, stderr) = runtime.exec(&id, &cmd).await.expect("Failed to exec");

    println!("Stderr exec - exit_code: {exit_code}, stdout: '{stdout}', stderr: '{stderr}'");
    assert_eq!(exit_code, 0, "Exec should succeed");
    assert!(
        stderr.contains("stderr output"),
        "Stderr should contain the error output"
    );

    // Test a command that fails
    println!("Testing command that fails");
    let cmd = vec!["sh".to_string(), "-c".to_string(), "exit 42".to_string()];
    let (exit_code, _, _) = runtime.exec(&id, &cmd).await.expect("Failed to exec");

    assert_eq!(exit_code, 42, "Exec should return the expected exit code");
}

/// Test getting container resource statistics
#[tokio::test]
async fn test_container_stats() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };
    let runtime = Arc::new(runtime);

    // Ensure image is available
    let _ = tokio::time::timeout(LONG_TIMEOUT, runtime.pull_image(TEST_IMAGE)).await;

    let id = unique_container_id("stats");
    let spec = create_sleep_spec(300); // Long-running container

    // Setup cleanup guard
    let _guard = ContainerGuard::new(runtime.clone(), id.clone());

    // Create and start container
    println!("Creating container for stats test");
    runtime
        .create_container(&id, &spec)
        .await
        .expect("Failed to create container");
    runtime
        .start_container(&id)
        .await
        .expect("Failed to start container");

    // Wait for container to be running
    wait_for_state(&runtime, &id, ContainerState::Running, SHORT_TIMEOUT)
        .await
        .expect("Container did not start");

    // Give it a moment to accumulate some stats
    tokio::time::sleep(Duration::from_secs(1)).await;

    // Get stats
    println!("Getting container stats");
    let stats = runtime
        .get_container_stats(&id)
        .await
        .expect("Failed to get stats");

    println!(
        "Container stats - CPU usage: {} usec, Memory: {} bytes, Memory limit: {} bytes",
        stats.cpu_usage_usec, stats.memory_bytes, stats.memory_limit
    );

    // Verify stats are reasonable (memory should be > 0 for a running container)
    assert!(
        stats.memory_bytes > 0,
        "Memory usage should be greater than 0"
    );
    assert!(
        stats.memory_limit > 0,
        "Memory limit should be greater than 0"
    );
}

/// Test waiting for a container to exit with exit code 0
#[tokio::test]
async fn test_wait_container_success() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };
    let runtime = Arc::new(runtime);

    // Ensure image is available
    let _ = tokio::time::timeout(LONG_TIMEOUT, runtime.pull_image(TEST_IMAGE)).await;

    let id = unique_container_id("wait-success");
    let _guard = ContainerGuard::new(runtime.clone(), id.clone());

    let mut spec = create_test_spec(TEST_IMAGE);
    spec.command = CommandSpec {
        entrypoint: None,
        args: Some(vec![
            "sh".to_string(),
            "-c".to_string(),
            "sleep 1 && exit 0".to_string(),
        ]),
        workdir: None,
    };

    println!("Testing wait_container with exit code 0");
    runtime
        .create_container(&id, &spec)
        .await
        .expect("Failed to create container");
    runtime
        .start_container(&id)
        .await
        .expect("Failed to start container");

    let exit_code = tokio::time::timeout(SHORT_TIMEOUT, runtime.wait_container(&id))
        .await
        .expect("Timeout waiting for container")
        .expect("Failed to wait for container");

    println!("Container exited with code: {exit_code}");
    assert_eq!(exit_code, 0, "Expected exit code 0");
}

/// Test waiting for a container to exit with non-zero exit code
#[tokio::test]
async fn test_wait_container_failure() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };
    let runtime = Arc::new(runtime);

    // Ensure image is available
    let _ = tokio::time::timeout(LONG_TIMEOUT, runtime.pull_image(TEST_IMAGE)).await;

    let id = unique_container_id("wait-nonzero");
    let _guard = ContainerGuard::new(runtime.clone(), id.clone());

    // Use exactly the same spec as the success test, but with exit 42
    let mut spec = create_test_spec(TEST_IMAGE);
    spec.command = CommandSpec {
        entrypoint: None,
        args: Some(vec![
            "sh".to_string(),
            "-c".to_string(),
            "sleep 2 && exit 42".to_string(),
        ]),
        workdir: None,
    };

    println!("Creating and starting container for exit code 42 test");
    println!("Container ID: {id:?}");

    runtime
        .create_container(&id, &spec)
        .await
        .expect("Failed to create container");

    runtime
        .start_container(&id)
        .await
        .expect("Failed to start container");

    // Check container is running
    let state = runtime
        .container_state(&id)
        .await
        .expect("Failed to get state after start");
    println!("State after start: {state:?}");

    // Wait for container using container_state polling as a workaround
    // if the wait_container API is unreliable
    println!("Waiting for container to exit...");
    let exit_code = loop {
        let state = runtime
            .container_state(&id)
            .await
            .expect("Failed to get state");
        match state {
            ContainerState::Exited { code } => {
                break code;
            }
            ContainerState::Running => {
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
            other => {
                panic!("Unexpected state: {other:?}");
            }
        }
    };

    println!("Container exited with code: {exit_code}");
    assert_eq!(exit_code, 42, "Expected exit code 42");
}

/// Test waiting for a container that runs briefly
#[tokio::test]
async fn test_wait_container_timing() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };
    let runtime = Arc::new(runtime);

    // Ensure image is available
    let _ = tokio::time::timeout(LONG_TIMEOUT, runtime.pull_image(TEST_IMAGE)).await;

    let id = unique_container_id("wait-brief");
    let _guard = ContainerGuard::new(runtime.clone(), id.clone());

    let mut spec = create_test_spec(TEST_IMAGE);
    spec.command = CommandSpec {
        entrypoint: None,
        args: Some(vec!["sleep".to_string(), "1".to_string()]),
        workdir: None,
    };

    println!("Testing wait_container with brief sleep");
    runtime
        .create_container(&id, &spec)
        .await
        .expect("Failed to create container");
    runtime
        .start_container(&id)
        .await
        .expect("Failed to start container");

    let start = std::time::Instant::now();
    let exit_code = tokio::time::timeout(SHORT_TIMEOUT, runtime.wait_container(&id))
        .await
        .expect("Timeout waiting for container")
        .expect("Failed to wait for container");
    let elapsed = start.elapsed();

    println!("Container exited with code {exit_code} after {elapsed:?}");
    assert_eq!(exit_code, 0, "Expected exit code 0");
    assert!(
        elapsed >= Duration::from_millis(900),
        "Should have waited at least ~1 second"
    );
}

/// Test that multiple containers can be managed concurrently
#[tokio::test]
async fn test_concurrent_containers() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };
    let runtime = Arc::new(runtime);

    // Ensure image is available
    let _ = tokio::time::timeout(LONG_TIMEOUT, runtime.pull_image(TEST_IMAGE)).await;

    let container_count = 3;
    let mut guards = Vec::new();
    let mut ids = Vec::new();

    // Create multiple containers concurrently
    println!("Creating {container_count} containers concurrently");
    let mut create_handles = Vec::new();

    for i in 0..container_count {
        let runtime_clone = runtime.clone();
        let id = unique_container_id(&format!("concurrent-{i}"));
        let spec = create_sleep_spec(300);
        let id_clone = id.clone();

        ids.push(id.clone());
        guards.push(ContainerGuard::new(runtime.clone(), id));

        create_handles.push(tokio::spawn(async move {
            runtime_clone.create_container(&id_clone, &spec).await
        }));
    }

    // Wait for all creates to complete
    for handle in create_handles {
        let result = handle.await.expect("Task panicked");
        assert!(result.is_ok(), "Failed to create container: {result:?}");
    }

    // Start all containers concurrently
    println!("Starting {container_count} containers concurrently");
    let mut start_handles = Vec::new();

    for id in &ids {
        let runtime_clone = runtime.clone();
        let id_clone = id.clone();

        start_handles.push(tokio::spawn(async move {
            runtime_clone.start_container(&id_clone).await
        }));
    }

    for handle in start_handles {
        let result = handle.await.expect("Task panicked");
        assert!(result.is_ok(), "Failed to start container: {result:?}");
    }

    // Verify all are running
    for id in &ids {
        let state = runtime
            .container_state(id)
            .await
            .expect("Failed to get state");
        assert_eq!(
            state,
            ContainerState::Running,
            "Container {} should be running",
            id.service
        );
    }

    println!("All {container_count} containers are running");

    // Stop all containers concurrently
    println!("Stopping {container_count} containers concurrently");
    let mut stop_handles = Vec::new();

    for id in &ids {
        let runtime_clone = runtime.clone();
        let id_clone = id.clone();

        stop_handles.push(tokio::spawn(async move {
            runtime_clone
                .stop_container(&id_clone, Duration::from_secs(10))
                .await
        }));
    }

    for handle in stop_handles {
        let result = handle.await.expect("Task panicked");
        assert!(result.is_ok(), "Failed to stop container: {result:?}");
    }

    println!("All containers stopped successfully");
}

/// Test that removing a non-existent container succeeds (idempotent)
#[tokio::test]
async fn test_remove_nonexistent_container() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };

    let id = unique_container_id("nonexistent");

    println!(
        "Attempting to remove non-existent container: {}",
        id.service
    );
    let result = runtime.remove_container(&id).await;

    // Remove should succeed for non-existent containers (force: true handles this)
    // Or it might return NotFound - both are acceptable
    match result {
        Ok(()) => println!("Remove succeeded (idempotent behavior)"),
        Err(e) => println!("Remove returned error (also acceptable): {e}"),
    }
}

/// Test that getting state of non-existent container returns `NotFound` error
#[tokio::test]
async fn test_state_nonexistent_container() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };

    let id = unique_container_id("ghost");

    println!("Getting state of non-existent container: {}", id.service);
    let result = runtime.container_state(&id).await;

    assert!(result.is_err(), "Should fail for non-existent container");
    println!("Got expected error: {:?}", result.unwrap_err());
}

/// Test pull with Never policy (should not pull)
#[tokio::test]
async fn test_pull_never_policy() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };

    // Use an image that definitely doesn't exist locally
    let nonexistent_image = "this-image-definitely-does-not-exist:never";

    println!("Testing Never pull policy with non-existent image");
    let result = runtime
        .pull_image_with_policy(nonexistent_image, PullPolicy::Never, None)
        .await;

    // Never policy should return Ok immediately without pulling
    assert!(
        result.is_ok(),
        "Never policy should succeed without pulling: {result:?}"
    );
    println!("Never policy correctly skipped pull attempt");
}

/// Test pull with Always policy
#[tokio::test]
async fn test_pull_always_policy() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };

    println!("Testing Always pull policy");
    let result = tokio::time::timeout(
        LONG_TIMEOUT,
        runtime.pull_image_with_policy(TEST_IMAGE, PullPolicy::Always, None),
    )
    .await;

    match result {
        Ok(Ok(())) => println!("Always policy pulled successfully"),
        Ok(Err(e)) => panic!("Failed with Always policy: {e}"),
        Err(e) => panic!("Timeout with Always policy: {e}"),
    }
}

/// Test container with environment variables
#[tokio::test]
async fn test_container_with_env() {
    let Some(runtime) = skip_if_no_docker().await else {
        eprintln!("Skipping test: Docker not available");
        return;
    };
    let runtime = Arc::new(runtime);

    // Ensure image is available
    let _ = tokio::time::timeout(LONG_TIMEOUT, runtime.pull_image(TEST_IMAGE)).await;

    let id = unique_container_id("env");
    let mut spec = create_test_spec(TEST_IMAGE);

    // Add environment variables
    spec.env
        .insert("TEST_VAR".to_string(), "test_value".to_string());
    spec.env
        .insert("ANOTHER_VAR".to_string(), "another_value".to_string());

    // Command that prints env vars
    spec.command = CommandSpec {
        entrypoint: None,
        args: Some(vec!["sh".to_string(), "-c".to_string(), "env".to_string()]),
        workdir: None,
    };

    let _guard = ContainerGuard::new(runtime.clone(), id.clone());

    println!("Creating container with environment variables");
    runtime
        .create_container(&id, &spec)
        .await
        .expect("Failed to create container");
    runtime
        .start_container(&id)
        .await
        .expect("Failed to start container");

    // Wait for container to exit
    let _ = wait_for_state(
        &runtime,
        &id,
        ContainerState::Exited { code: 0 },
        SHORT_TIMEOUT,
    )
    .await;

    // Get logs and verify env vars
    let logs = runtime
        .container_logs(&id, 100)
        .await
        .expect("Failed to get logs");

    let logs_text: String = logs
        .iter()
        .map(ToString::to_string)
        .collect::<Vec<_>>()
        .join("\n");
    println!("Container output:\n{logs_text}");
    assert!(
        logs.iter()
            .any(|e| e.message.contains("TEST_VAR=test_value")),
        "Logs should contain TEST_VAR"
    );
    assert!(
        logs.iter()
            .any(|e| e.message.contains("ANOTHER_VAR=another_value")),
        "Logs should contain ANOTHER_VAR"
    );
}