redisctl 0.12.0

//! Mocked end-to-end tests for `cloud workflow quick-database`.
//!
//! Each test points a cloud profile at a wiremock CAPI and drives the real binary. Every
//! scenario asserts the secret-leak guard: the mock password and the `rediss://…@` URL must
//! appear in the credentials file but NEVER on stdout/stderr.

use assert_cmd::Command;
use serde_json::{Value, json};
use tempfile::TempDir;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};

const MOCK_PASSWORD: &str = "s3cr3t-mock-pw";
const MOCK_ENDPOINT: &str = "mock-host.example.com:12000";
const DB_NAME: &str = "quick-db-test";
const SUB_ID: i64 = 501;
const DB_ID: i64 = 9001;

fn write_cloud_profile(temp_dir: &TempDir, api_url: &str) {
    let config = format!(
        r#"
[profiles.test]
deployment_type = "cloud"
api_key = "test-api-key"
api_secret = "test-api-secret"
api_url = "{api_url}"

default_cloud = "test"
"#
    );
    std::fs::write(temp_dir.path().join("config.toml"), config).unwrap();
}

fn run_quick_database(
    temp_dir: &TempDir,
    env_path: &std::path::Path,
) -> assert_cmd::assert::Assert {
    let mut cmd = Command::cargo_bin("redisctl").unwrap();
    cmd.env_remove("REDIS_CLOUD_API_KEY");
    cmd.env_remove("REDIS_CLOUD_SECRET_KEY");
    cmd.env_remove("REDIS_CLOUD_API_URL");
    cmd.env_remove("REDISCTL_PROFILE");
    cmd.arg("--config-file")
        .arg(temp_dir.path().join("config.toml"))
        .arg("cloud")
        .arg("workflow")
        .arg("quick-database")
        .arg("--name")
        .arg(DB_NAME)
        .arg("--output-credentials")
        .arg(env_path)
        .arg("--wait-interval")
        .arg("1")
        .arg("--wait-timeout")
        .arg("30")
        .arg("-o")
        .arg("json");
    cmd.assert()
}

fn fixed_database_body() -> Value {
    database_body(DB_ID, DB_NAME)
}

fn database_body(id: i64, name: &str) -> Value {
    json!({
        "databaseId": id,
        "name": name,
        "region": "us-east-1",
        "publicEndpoint": MOCK_ENDPOINT,
        "security": { "enableTls": true, "password": MOCK_PASSWORD }
    })
}

/// Our subscription as the list reports it. `plan_id` is what reuse checks; `None` models an
/// API that does not report one.
fn subscription_body(plan_id: Option<i64>) -> Value {
    subscription_with(plan_id, None)
}

/// Our subscription as the list reports it. `price` settles free-ness on its own; `plan_id` is
/// the fallback when the API omits a price. `None` for both models an API that reports neither.
fn subscription_with(plan_id: Option<i64>, price: Option<i64>) -> Value {
    let mut sub = json!({
        "id": SUB_ID,
        "name": format!("redisctl-{DB_NAME}"),
        "status": "active"
    });
    if let Some(plan_id) = plan_id {
        sub["planId"] = json!(plan_id);
        sub["planName"] = json!("Standard 1GB");
    }
    if let Some(price) = price {
        sub["price"] = json!(price);
    }
    sub
}

async fn mock_subscription_list(server: &MockServer, plan_id: Option<i64>) {
    mount_subscription_list(server, subscription_body(plan_id)).await;
}

async fn mount_subscription_list(server: &MockServer, subscription: Value) {
    Mock::given(method("GET"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscriptions": [ subscription ]
        })))
        .mount(server)
        .await;
}

async fn mock_database_list(server: &MockServer, databases: Vec<Value>) {
    Mock::given(method("GET"))
        .and(path(format!("/fixed/subscriptions/{SUB_ID}/databases")))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscription": { "subscriptionId": SUB_ID, "databases": databases }
        })))
        .mount(server)
        .await;
}

async fn mock_free_plan(server: &MockServer) {
    Mock::given(method("GET"))
        .and(path("/fixed/plans"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "plans": [
                { "id": 34, "name": "Standard", "price": 5 },
                { "id": 12, "name": "Free", "price": 0, "provider": "AWS", "region": "us-east-1" },
                { "id": 99, "name": "Free", "price": 0, "provider": "GCP", "region": "europe-west1" }
            ]
        })))
        .mount(server)
        .await;
}

async fn mock_task_completed(server: &MockServer, task_id: &str, resource_id: i64) {
    Mock::given(method("GET"))
        .and(path(format!("/tasks/{task_id}")))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "taskId": task_id,
            "status": "processing-completed",
            "response": { "resourceId": resource_id }
        })))
        .mount(server)
        .await;
}

/// Assert the leak guard: secrets in the file, never in the captured output.
fn assert_no_secret_leak(stdout: &[u8], stderr: &[u8]) {
    let out = String::from_utf8_lossy(stdout);
    let err = String::from_utf8_lossy(stderr);
    for stream in [&out, &err] {
        assert!(
            !stream.contains(MOCK_PASSWORD),
            "password leaked into output: {stream}"
        );
        assert!(
            !stream.contains("rediss://default:"),
            "connection URL leaked into output: {stream}"
        );
    }
}

#[tokio::test]
async fn fresh_create_writes_env_and_prints_schema() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // No existing subscription → fresh provisioning.
    Mock::given(method("GET"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "subscriptions": [] })))
        .mount(&server)
        .await;
    mock_free_plan(&server).await;
    Mock::given(method("POST"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(202).set_body_json(json!({
            "taskId": "task-sub", "status": "received"
        })))
        .mount(&server)
        .await;
    mock_task_completed(&server, "task-sub", SUB_ID).await;
    Mock::given(method("POST"))
        .and(path(format!("/fixed/subscriptions/{SUB_ID}/databases")))
        .respond_with(ResponseTemplate::new(202).set_body_json(json!({
            "taskId": "task-db", "status": "received"
        })))
        .mount(&server)
        .await;
    mock_task_completed(&server, "task-db", DB_ID).await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(fixed_database_body()))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .success()
        .get_output()
        .clone();
    assert_no_secret_leak(&output.stdout, &output.stderr);

    // Schema lock (PRD §5.2): exact key set, exact values.
    let report: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(report["status"], "ok");
    assert_eq!(report["database"]["id"], DB_ID.to_string());
    assert_eq!(report["database"]["name"], DB_NAME);
    assert_eq!(report["database"]["region"], "us-east-1");
    assert_eq!(report["database"]["plan"], "free");
    assert_eq!(report["database"]["tls"], true);
    assert_eq!(report["credentials_variable"], "REDIS_URL");
    assert_eq!(
        report["credentials_written_to"],
        env_path.display().to_string()
    );
    // Key set is locked (serializer emits keys sorted, so compare sorted).
    let mut top_keys: Vec<&str> = report
        .as_object()
        .unwrap()
        .keys()
        .map(String::as_str)
        .collect();
    top_keys.sort_unstable();
    assert_eq!(
        top_keys,
        [
            "credentials_variable",
            "credentials_written_to",
            "database",
            "status"
        ]
    );
    let mut db_keys: Vec<&str> = report["database"]
        .as_object()
        .unwrap()
        .keys()
        .map(String::as_str)
        .collect();
    db_keys.sort_unstable();
    assert_eq!(db_keys, ["id", "name", "plan", "region", "tls"]);

    // The credentials file holds the full connection string plus broken-out fields.
    let env_body = std::fs::read_to_string(&env_path).unwrap();
    assert!(env_body.contains(&format!(
        "REDIS_URL=rediss://default:{MOCK_PASSWORD}@{MOCK_ENDPOINT}"
    )));
    assert!(env_body.contains("REDIS_HOST=mock-host.example.com"));
    assert!(env_body.contains("REDIS_PORT=12000"));
    assert!(env_body.contains(&format!("REDIS_PASSWORD={MOCK_PASSWORD}")));
    assert!(env_body.contains("REDIS_USERNAME=default"));
    assert!(env_body.contains("REDIS_TLS=true"));
}

#[tokio::test]
async fn second_run_reuses_without_writes() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // Our subscription already exists, with a database. No POST mocks are mounted, so any
    // create attempt would 404 and fail the run — proving reuse performs no writes.
    Mock::given(method("GET"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscriptions": [ { "id": SUB_ID, "name": format!("redisctl-{DB_NAME}"), "status": "active" } ]
        })))
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .and(path(format!("/fixed/subscriptions/{SUB_ID}/databases")))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscription": { "subscriptionId": SUB_ID, "databases": [ fixed_database_body() ] }
        })))
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(fixed_database_body()))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .success()
        .get_output()
        .clone();
    assert_no_secret_leak(&output.stdout, &output.stderr);
    let report: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(report["status"], "reused");
    assert_eq!(report["database"]["id"], DB_ID.to_string());

    let env_body = std::fs::read_to_string(&env_path).unwrap();
    assert!(env_body.contains(MOCK_PASSWORD));
}

/// The public endpoint can lag behind task completion on a fresh create. The workflow must
/// poll the database GET until it appears, not fail the first read.
#[tokio::test]
async fn waits_for_public_endpoint_to_appear() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // Reuse path: our subscription + database already exist.
    Mock::given(method("GET"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscriptions": [ { "id": SUB_ID, "name": format!("redisctl-{DB_NAME}"), "status": "active" } ]
        })))
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .and(path(format!("/fixed/subscriptions/{SUB_ID}/databases")))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscription": { "subscriptionId": SUB_ID, "databases": [ { "databaseId": DB_ID } ] }
        })))
        .mount(&server)
        .await;
    // First GET has no endpoint yet (mounted first, capped at one match); the retry succeeds.
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "databaseId": DB_ID,
            "name": DB_NAME,
            "security": { "enableTls": true, "password": MOCK_PASSWORD }
        })))
        .up_to_n_times(1)
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(fixed_database_body()))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .success()
        .get_output()
        .clone();
    assert_no_secret_leak(&output.stdout, &output.stderr);
    let env_body = std::fs::read_to_string(&env_path).unwrap();
    assert!(env_body.contains(&format!(
        "REDIS_URL=rediss://default:{MOCK_PASSWORD}@{MOCK_ENDPOINT}"
    )));
}

/// If the endpoint never appears within the timeout, that's transient (exit 3), not `unknown`.
#[tokio::test]
async fn missing_endpoint_past_timeout_is_transient() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    Mock::given(method("GET"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscriptions": [ { "id": SUB_ID, "name": format!("redisctl-{DB_NAME}"), "status": "active" } ]
        })))
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .and(path(format!("/fixed/subscriptions/{SUB_ID}/databases")))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscription": { "subscriptionId": SUB_ID, "databases": [ { "databaseId": DB_ID } ] }
        })))
        .mount(&server)
        .await;
    // The endpoint never populates.
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "databaseId": DB_ID,
            "name": DB_NAME,
            "security": { "enableTls": true, "password": MOCK_PASSWORD }
        })))
        .mount(&server)
        .await;

    let mut cmd = Command::cargo_bin("redisctl").unwrap();
    cmd.env_remove("REDIS_CLOUD_API_KEY");
    cmd.env_remove("REDIS_CLOUD_SECRET_KEY");
    cmd.env_remove("REDIS_CLOUD_API_URL");
    cmd.env_remove("REDISCTL_PROFILE");
    let output = cmd
        .arg("--config-file")
        .arg(temp.path().join("config.toml"))
        .args([
            "cloud",
            "workflow",
            "quick-database",
            "--name",
            DB_NAME,
            "--wait-timeout",
            "1",
            "--wait-interval",
            "1",
            "-o",
            "json",
        ])
        .arg("--output-credentials")
        .arg(&env_path)
        .assert()
        .code(3)
        .get_output()
        .clone();
    let env: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(env["error"]["code"], "transient_api_error");
    assert_eq!(env["error"]["retryable"], true);
    assert!(!env_path.exists());
}

/// PRD §7.2 leak guard: even at max verbosity (`-vvv`), the password — which flows through
/// the CAPI response bodies on the reuse path — must not reach stdout or stderr.
#[tokio::test]
async fn no_secret_leak_at_max_verbosity() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    Mock::given(method("GET"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscriptions": [ { "id": SUB_ID, "name": format!("redisctl-{DB_NAME}"), "status": "active" } ]
        })))
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .and(path(format!("/fixed/subscriptions/{SUB_ID}/databases")))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscription": { "subscriptionId": SUB_ID, "databases": [ fixed_database_body() ] }
        })))
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(fixed_database_body()))
        .mount(&server)
        .await;

    let mut cmd = Command::cargo_bin("redisctl").unwrap();
    cmd.env_remove("REDIS_CLOUD_API_KEY");
    cmd.env_remove("REDIS_CLOUD_SECRET_KEY");
    cmd.env_remove("REDIS_CLOUD_API_URL");
    cmd.env_remove("REDISCTL_PROFILE");
    cmd.env_remove("RUST_LOG"); // don't let the runner's RUST_LOG change verbosity
    let output = cmd
        .arg("--config-file")
        .arg(temp.path().join("config.toml"))
        .arg("-vvv")
        .args([
            "cloud",
            "workflow",
            "quick-database",
            "--name",
            DB_NAME,
            "-o",
            "json",
        ])
        .arg("--output-credentials")
        .arg(&env_path)
        .assert()
        .success()
        .get_output()
        .clone();

    assert_no_secret_leak(&output.stdout, &output.stderr);
    // Sanity: the secret really did travel through the API responses (so the guard is meaningful).
    assert!(
        std::fs::read_to_string(&env_path)
            .unwrap()
            .contains(MOCK_PASSWORD)
    );
}

#[tokio::test]
async fn resume_after_partial_create() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // Subscription exists but has NO database (crashed between the two creates). The workflow
    // must resume by creating only the database — no subscription POST is mounted.
    Mock::given(method("GET"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscriptions": [ { "id": SUB_ID, "name": format!("redisctl-{DB_NAME}"), "status": "active" } ]
        })))
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .and(path(format!("/fixed/subscriptions/{SUB_ID}/databases")))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "subscription": { "subscriptionId": SUB_ID, "databases": [] }
        })))
        .mount(&server)
        .await;
    Mock::given(method("POST"))
        .and(path(format!("/fixed/subscriptions/{SUB_ID}/databases")))
        .respond_with(ResponseTemplate::new(202).set_body_json(json!({
            "taskId": "task-db", "status": "received"
        })))
        .mount(&server)
        .await;
    mock_task_completed(&server, "task-db", DB_ID).await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(fixed_database_body()))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .success()
        .get_output()
        .clone();
    assert_no_secret_leak(&output.stdout, &output.stderr);
    let report: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(report["status"], "ok");
    assert_eq!(report["database"]["id"], DB_ID.to_string());
}

#[tokio::test]
async fn task_failure_surfaces_error() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    Mock::given(method("GET"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "subscriptions": [] })))
        .mount(&server)
        .await;
    mock_free_plan(&server).await;
    Mock::given(method("POST"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(202).set_body_json(json!({
            "taskId": "task-sub", "status": "received"
        })))
        .mount(&server)
        .await;
    // The create task fails mid-flight.
    Mock::given(method("GET"))
        .and(path("/tasks/task-sub"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "taskId": "task-sub",
            "status": "processing-error",
            "response": { "error": "provisioning blew up" }
        })))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .code(1)
        .get_output()
        .clone();
    assert_no_secret_leak(&output.stdout, &output.stderr);
    // In JSON mode the error envelope is on stdout (agents parse stdout).
    let env: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(env["status"], "error");
    assert_eq!(env["error"]["code"], "unknown");
    assert!(
        env["error"]["message"]
            .as_str()
            .unwrap()
            .contains("provisioning blew up")
    );
    assert!(!env_path.exists(), "no credentials file on failure");
}

#[tokio::test]
async fn free_plan_gate_rejection_is_actionable() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    Mock::given(method("GET"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "subscriptions": [] })))
        .mount(&server)
        .await;
    mock_free_plan(&server).await;
    // The public CAPI rejects free-plan creation when the trusted-UA / payment gate is unmet.
    Mock::given(method("POST"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(400).set_body_json(json!({
            "description": "FREE_PLAN_IS_ALLOWED_ONLY_FOR_ACCOUNTS_WITH_VALID_PAYMENT_INFO"
        })))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .code(4)
        .get_output()
        .clone();
    let env: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(env["status"], "error");
    assert_eq!(env["error"]["code"], "free_db_exists");
    assert_eq!(env["error"]["retryable"], false);
    assert!(!env_path.exists());
}

/// The one-free-subscription limit is reported by the API as an *async task* failure (not a
/// synchronous 4xx), so it must still classify as `free_db_exists` (exit 4), not `unknown`.
#[tokio::test]
async fn free_sub_limit_via_task_error_is_free_db_exists() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    Mock::given(method("GET"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "subscriptions": [] })))
        .mount(&server)
        .await;
    mock_free_plan(&server).await;
    Mock::given(method("POST"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(202).set_body_json(json!({
            "taskId": "task-sub", "status": "received"
        })))
        .mount(&server)
        .await;
    // The create task fails asynchronously with the one-free-sub message.
    Mock::given(method("GET"))
        .and(path("/tasks/task-sub"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "taskId": "task-sub",
            "status": "processing-error",
            "response": { "error": "The account already has a free plan Essentials subscription." }
        })))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .code(4)
        .get_output()
        .clone();
    let env: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(env["error"]["code"], "free_db_exists");
    assert!(!env_path.exists());
}

#[tokio::test]
async fn invalid_name_is_rejected_before_any_call() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // No mocks mounted: a valid run would hit the API, but validation must fail first.
    let mut cmd = Command::cargo_bin("redisctl").unwrap();
    cmd.env_remove("REDIS_CLOUD_API_KEY");
    cmd.env_remove("REDIS_CLOUD_SECRET_KEY");
    cmd.env_remove("REDIS_CLOUD_API_URL");
    let output = cmd
        .arg("--config-file")
        .arg(temp.path().join("config.toml"))
        .args([
            "cloud",
            "workflow",
            "quick-database",
            "--name",
            "Invalid_Name",
            "-o",
            "json",
        ])
        .arg("--output-credentials")
        .arg(&env_path)
        .assert()
        .code(2)
        .get_output()
        .clone();
    let env: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(env["status"], "error");
    assert_eq!(env["error"]["code"], "invalid_name");
    assert!(
        env["error"]["message"]
            .as_str()
            .unwrap()
            .contains("invalid database name")
    );
    assert!(!env_path.exists());
}

#[tokio::test]
async fn persistent_5xx_is_retryable_exit_3() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // The very first CAPI call (list subscriptions) fails with a persistent 503.
    Mock::given(method("GET"))
        .and(path("/fixed/subscriptions"))
        .respond_with(ResponseTemplate::new(503).set_body_json(json!({
            "description": "service temporarily unavailable"
        })))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .code(3)
        .get_output()
        .clone();
    let env: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(env["status"], "error");
    assert_eq!(env["error"]["code"], "transient_api_error");
    assert_eq!(env["error"]["retryable"], true);
    assert!(!env_path.exists());
}

/// `database-credentials`: write an EXISTING database's connection string to a file with no
/// provisioning (single GET). Report status is "existing"; the leak guard still holds.
#[tokio::test]
async fn database_credentials_writes_existing_db_without_provisioning() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // Only a GET on the database — no list/plan/create mocks, so any provisioning attempt 404s.
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(fixed_database_body()))
        .mount(&server)
        .await;

    let mut cmd = Command::cargo_bin("redisctl").unwrap();
    cmd.env_remove("REDIS_CLOUD_API_KEY");
    cmd.env_remove("REDIS_CLOUD_SECRET_KEY");
    cmd.env_remove("REDIS_CLOUD_API_URL");
    cmd.env_remove("REDISCTL_PROFILE");
    let output = cmd
        .arg("--config-file")
        .arg(temp.path().join("config.toml"))
        .args([
            "cloud",
            "workflow",
            "database-credentials",
            "--subscription-id",
            &SUB_ID.to_string(),
            "--database-id",
            &DB_ID.to_string(),
            "-o",
            "json",
        ])
        .arg("--output-credentials")
        .arg(&env_path)
        .assert()
        .success()
        .get_output()
        .clone();

    assert_no_secret_leak(&output.stdout, &output.stderr);
    let report: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(report["status"], "existing");
    assert_eq!(report["database"]["id"], DB_ID.to_string());
    assert_eq!(report["database"]["name"], DB_NAME);
    assert_eq!(report["database"]["plan"], "essentials");

    let env_body = std::fs::read_to_string(&env_path).unwrap();
    assert!(env_body.contains(&format!(
        "REDIS_URL=rediss://default:{MOCK_PASSWORD}@{MOCK_ENDPOINT}"
    )));
    assert!(env_body.contains("REDIS_HOST=mock-host.example.com"));
}

/// A subscription carrying our prefix on a paid plan is refused, not adopted: the report says
/// `plan: "free"` unconditionally, so reusing it would bill the caller while telling them it is
/// free. No POST mocks are mounted, so any provisioning attempt would fail the run.
#[tokio::test]
async fn reuse_refuses_a_subscription_on_a_paid_plan() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    mount_subscription_list(&server, subscription_with(Some(34), Some(5))).await;
    mock_free_plan(&server).await;

    let output = run_quick_database(&temp, &env_path)
        .code(2)
        .get_output()
        .clone();
    assert_no_secret_leak(&output.stdout, &output.stderr);
    let env: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(env["error"]["code"], "name_conflict");
    assert_eq!(env["error"]["retryable"], false);
    let message = env["error"]["message"].as_str().unwrap();
    assert!(message.contains("Standard 1GB"), "{message}");
    assert!(message.contains("database-credentials"), "{message}");
    assert!(!env_path.exists(), "no credentials file on refusal");
}

/// The free plan is reused as before. Proves the new check reads the plan rather than refusing
/// anything it does not recognise.
#[tokio::test]
async fn reuse_accepts_the_free_plan() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    mock_subscription_list(&server, Some(12)).await;
    mock_free_plan(&server).await;
    mock_database_list(&server, vec![fixed_database_body()]).await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(fixed_database_body()))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .success()
        .get_output()
        .clone();
    assert_no_secret_leak(&output.stdout, &output.stderr);
    let report: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(report["status"], "reused");
    assert_eq!(report["database"]["plan"], "free");
}

/// Reuse resolves the database by name. The requested one is listed second, so taking the first
/// entry would deliver another database's credentials.
#[tokio::test]
async fn reuse_picks_the_database_matching_the_name() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    mock_subscription_list(&server, None).await;
    mock_database_list(
        &server,
        vec![
            database_body(8000, "someone-elses-db"),
            fixed_database_body(),
        ],
    )
    .await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(fixed_database_body()))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .success()
        .get_output()
        .clone();
    assert_no_secret_leak(&output.stdout, &output.stderr);
    let report: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(report["status"], "reused");
    assert_eq!(report["database"]["id"], DB_ID.to_string());
    assert_eq!(report["database"]["name"], DB_NAME);
}

/// Several databases and none named as asked: which one gets delivered would come down to list
/// order, so refuse and name what is there.
#[tokio::test]
async fn reuse_refuses_when_no_database_matches_and_several_exist() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    mock_subscription_list(&server, None).await;
    mock_database_list(
        &server,
        vec![
            database_body(8000, "first-db"),
            database_body(8001, "second-db"),
        ],
    )
    .await;

    let output = run_quick_database(&temp, &env_path)
        .code(2)
        .get_output()
        .clone();
    assert_no_secret_leak(&output.stdout, &output.stderr);
    let env: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(env["error"]["code"], "name_conflict");
    let message = env["error"]["message"].as_str().unwrap();
    assert!(message.contains("first-db"), "{message}");
    assert!(message.contains("second-db"), "{message}");
    assert!(!env_path.exists(), "no credentials file on refusal");
}

/// A sole database renamed outside redisctl is still reused: with one candidate there is nothing
/// to be ambiguous about, and refusing would break a re-run after a console edit.
#[tokio::test]
async fn reuse_accepts_a_single_renamed_database() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    mock_subscription_list(&server, None).await;
    mock_database_list(&server, vec![database_body(8000, "renamed-by-hand")]).await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/8000"
        )))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(database_body(8000, "renamed-by-hand")),
        )
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .success()
        .get_output()
        .clone();
    assert_no_secret_leak(&output.stdout, &output.stderr);
    let report: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(report["status"], "reused");
    assert_eq!(report["database"]["id"], "8000");
    assert_eq!(report["database"]["name"], "renamed-by-hand");
}

/// The free plan is not one plan: Essentials plans are per provider and region, so several ids
/// are free. A free subscription on any of them must still be reused — comparing against a
/// single "the" free plan id refuses a subscription created in another region.
#[tokio::test]
async fn reuse_accepts_a_free_plan_from_another_region() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // Plan 99 is free, and is not the first free plan the list reports. No price on the
    // subscription, so the plan id is what gets checked.
    mount_subscription_list(&server, subscription_with(Some(99), None)).await;
    mock_free_plan(&server).await;
    mock_database_list(&server, vec![fixed_database_body()]).await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(fixed_database_body()))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .success()
        .get_output()
        .clone();
    let report: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(report["status"], "reused");
}

/// A price of zero settles it, without asking for the plan list at all.
#[tokio::test]
async fn reuse_trusts_a_zero_price_without_listing_plans() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // No /fixed/plans mock: reaching for one would 404 and fail the run.
    mount_subscription_list(&server, subscription_with(Some(34), Some(0))).await;
    mock_database_list(&server, vec![fixed_database_body()]).await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(fixed_database_body()))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .success()
        .get_output()
        .clone();
    let report: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(report["status"], "reused");
}

/// A plan id the account's plan list does not offer is reused, not refused. `/fixed/plans` is
/// the list of plans this account can *create*, and the free plan drops off it once the account
/// holds its one free subscription — so absence from it is no evidence of a bill, and treating it
/// as evidence broke re-running against our own subscription.
#[tokio::test]
async fn reuse_accepts_a_plan_the_list_does_not_offer() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // Plan 777 appears nowhere in /fixed/plans, and the subscription reports no price.
    mount_subscription_list(&server, subscription_with(Some(777), None)).await;
    mock_free_plan(&server).await;
    mock_database_list(&server, vec![fixed_database_body()]).await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(fixed_database_body()))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .success()
        .get_output()
        .clone();
    let report: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(report["status"], "reused");
}

/// The plan-id fallback still refuses a plan the list positively prices above zero, so dropping
/// the "absence means billable" rule did not give up the check itself.
#[tokio::test]
async fn reuse_refuses_a_paid_plan_id_when_no_price_is_reported() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // Plan 34 is listed at price 5; the subscription itself reports no price.
    mount_subscription_list(&server, subscription_with(Some(34), None)).await;
    mock_free_plan(&server).await;

    let output = run_quick_database(&temp, &env_path)
        .code(2)
        .get_output()
        .clone();
    assert_no_secret_leak(&output.stdout, &output.stderr);
    let env: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(env["error"]["code"], "name_conflict");
    assert!(!env_path.exists(), "no credentials file on refusal");
}

/// The database name is matched the way the subscription name is — ignoring case. A console-made
/// `redisctl-<Name>` subscription is adopted by the case-insensitive subscription lookup, so its
/// differently-cased database has to count as ours too rather than being refused as a stranger.
#[tokio::test]
async fn reuse_matches_the_database_name_ignoring_case() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    let upper = DB_NAME.to_uppercase();
    mock_subscription_list(&server, None).await;
    mock_database_list(
        &server,
        vec![
            database_body(8000, "someone-elses-db"),
            database_body(DB_ID, &upper),
        ],
    )
    .await;
    Mock::given(method("GET"))
        .and(path(format!(
            "/fixed/subscriptions/{SUB_ID}/databases/{DB_ID}"
        )))
        .respond_with(ResponseTemplate::new(200).set_body_json(database_body(DB_ID, &upper)))
        .mount(&server)
        .await;

    let output = run_quick_database(&temp, &env_path)
        .success()
        .get_output()
        .clone();
    let report: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(report["status"], "reused");
    assert_eq!(report["database"]["id"], DB_ID.to_string());
}

/// The database we asked for, listed without an id: its own failure, not "none is named that".
/// The old wording named it in the list it claimed did not contain it, and with a single entry
/// the run fell through to creating a second database of the same name.
#[tokio::test]
async fn reuse_reports_a_named_database_that_has_no_id() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    // No POST mock: a fall-through to create would fail the run instead of reporting this.
    mock_subscription_list(&server, None).await;
    mock_database_list(
        &server,
        vec![
            json!({ "name": DB_NAME, "region": "us-east-1" }),
            database_body(8001, "second-db"),
        ],
    )
    .await;

    let output = run_quick_database(&temp, &env_path)
        .code(3)
        .get_output()
        .clone();
    let env: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(env["error"]["retryable"], true);
    let message = env["error"]["message"].as_str().unwrap();
    assert!(message.contains("without an id"), "{message}");
    assert!(!env_path.exists(), "no credentials file on refusal");
}

/// `database-credentials` validates the variable name too. It shares the delivery path with
/// `quick-database`, where the name is interpolated as `KEY=value`, so a newline would write an
/// assignment of its own. No mocks are mounted: the refusal must come before any API call.
#[tokio::test]
async fn database_credentials_refuses_a_variable_that_is_not_an_env_var_name() {
    let temp = TempDir::new().unwrap();
    let server = MockServer::start().await;
    write_cloud_profile(&temp, &server.uri());
    let env_path = temp.path().join(".env");

    let mut cmd = Command::cargo_bin("redisctl").unwrap();
    cmd.env_remove("REDIS_CLOUD_API_KEY");
    cmd.env_remove("REDIS_CLOUD_SECRET_KEY");
    cmd.env_remove("REDIS_CLOUD_API_URL");
    cmd.env_remove("REDISCTL_PROFILE");
    let output = cmd
        .arg("--config-file")
        .arg(temp.path().join("config.toml"))
        .args([
            "cloud",
            "workflow",
            "database-credentials",
            "--subscription-id",
            &SUB_ID.to_string(),
            "--database-id",
            &DB_ID.to_string(),
            "--variable",
            "REDIS_URL\nINJECTED=owned",
            "-o",
            "json",
        ])
        .arg("--output-credentials")
        .arg(&env_path)
        .assert()
        .code(2)
        .get_output()
        .clone();

    let env: Value = serde_json::from_slice(&output.stdout).expect("stdout is JSON");
    assert_eq!(env["status"], "error");
    assert_eq!(env["error"]["code"], "invalid_name");
    assert_eq!(env["error"]["retryable"], false);
    assert!(!env_path.exists(), "no credentials file on refusal");
    assert!(
        server.received_requests().await.unwrap().is_empty(),
        "refused after calling the API"
    );
}