redisctl-core 0.12.1

//! Orchestrates a full `cloud auth login`: build the OIDC flow clients, and after a flow
//! yields tokens, run the SM exchange and mint a CAPI key.
//!
//! Returns [`MintedCredentials`] for the caller to persist (see
//! `config::Config::apply_cloud_login`). Persistence lives in the config layer so this stays
//! free of file/keyring I/O and easy to test. The flow itself (device polling with progress,
//! or loopback with a browser) is driven by the CLI using [`CloudAuthenticator::device`]
//! / [`CloudAuthenticator::loopback`].

use url::Url;

use super::sm_api::{LoginFlow, SmAccount, SmApiClient, SmUser};
use super::{AuthError, DeviceFlowClient, LoopbackFlowClient, TokenSet, default_http_client};

/// Result of a completed login: a Redis Cloud CAPI key pair plus context, ready to persist.
///
/// Secret fields are redacted from `Debug`.
#[derive(Clone)]
pub struct MintedCredentials {
    /// Numeric account id, matching the `id` of the matching entry in `accounts`, so the two can
    /// be compared directly by a caller reading the JSON output.
    pub account_id: Option<u64>,
    pub email: Option<String>,
    /// Account-level CAPI key (`x-api-key`).
    pub api_key: String,
    /// Minted user secret (`x-api-secret-key`).
    pub api_secret: String,
    /// CAPI base URL to record in the resulting cloud profile.
    pub api_url: String,
    /// Okta refresh token (rotating) to persist for silent re-auth, if the IdP issued one.
    pub refresh_token: Option<String>,
    /// Name of the minted `redisctl-*` CAPI key (visible/revocable in the console).
    pub capi_key_name: String,
    /// How many `redisctl-*` CAPI keys the account has after this mint (best-effort; 0 if the
    /// listing failed). The CLI warns when this grows, since each login mints a new key (D5).
    pub redisctl_key_count: usize,
    /// Name of the account the key was minted for, when the API reports one.
    pub account_name: Option<String>,
    /// Whether the key this switch replaced was revoked. `None` when there was none to revoke.
    pub superseded_revoked: Option<bool>,
    /// The name of that key, so a failed revocation can say which one is left behind.
    pub superseded_key_name: Option<String>,
    /// Whether this login is what switched account-wide programmatic access on. Reported so an
    /// account-level change is not made silently.
    pub capi_newly_enabled: bool,
    /// Every account the signed-in user belongs to. The key is scoped to exactly one of them —
    /// the session's *current* account — so the CLI can both name the one it used and list the
    /// alternatives, which are otherwise only discoverable in the console.
    pub accounts: Vec<LoginAccount>,
}

/// How the account to mint for is decided.
///
/// [`AccountChoice::Prompt`] exists because a picker cannot run before the exchange: listing the
/// accounts needs a session, and re-logging-in to act on the answer would mean a second sign-in
/// (and a second MFA challenge). The callback is invoked mid-exchange instead, on the one session.
/// A key a switch is about to replace, revoked on the same session once its successor exists.
pub struct SupersededKey {
    pub account_id: u64,
    pub key_name: String,
}

pub enum AccountChoice {
    /// Whatever account the session is already on.
    Current,
    /// A specific account id.
    Id(u64),
    /// Decide once the accounts are known. Called with every account the user belongs to and the
    /// id of the current one, when the API reports it.
    Prompt(AccountPrompt),
}

/// Callback for [`AccountChoice::Prompt`]: pick an account id, or fail with the reason.
pub type AccountPrompt =
    Box<dyn Fn(&[LoginAccount], Option<u64>) -> Result<u64, AuthError> + Send + Sync>;

impl std::fmt::Debug for AccountChoice {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Current => f.write_str("Current"),
            Self::Id(id) => write!(f, "Id({id})"),
            Self::Prompt(_) => f.write_str("Prompt(..)"),
        }
    }
}

/// One account the signed-in user belongs to, as reported during login.
#[derive(Debug, Clone)]
pub struct LoginAccount {
    pub id: u64,
    pub name: Option<String>,
}

impl LoginAccount {
    /// `Acme (#316941)`, or `#316941` when the API reports no name. Used both in the CLI listing
    /// and in the `UnknownAccount` message, so the two always read the same.
    pub fn label(&self) -> String {
        match &self.name {
            Some(n) => format!("{} (#{})", n, self.id),
            None => format!("#{}", self.id),
        }
    }
}

impl MintedCredentials {
    /// How many accounts the signed-in user belongs to.
    pub fn account_count(&self) -> usize {
        self.accounts.len()
    }

    /// The account the key is for, rendered like the listing (`Acme (#316941)`).
    ///
    /// Shared so every place that names the account spells it the same way. `account_id` is always
    /// one of `accounts` — both come from the same `/accounts` response — so the lookup only
    /// misses for a hand-built value.
    pub fn account_label(&self) -> String {
        self.account_id
            .and_then(|id| self.accounts.iter().find(|a| a.id == id))
            .map(LoginAccount::label)
            .unwrap_or_else(|| "your current account".to_string())
    }
}

impl std::fmt::Debug for MintedCredentials {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MintedCredentials")
            .field("account_id", &self.account_id)
            .field("email", &self.email)
            .field("api_key", &"<redacted>")
            .field("api_secret", &"<redacted>")
            .field("api_url", &self.api_url)
            .field(
                "refresh_token",
                &self.refresh_token.as_ref().map(|_| "<redacted>"),
            )
            .field("capi_key_name", &self.capi_key_name)
            .field("redisctl_key_count", &self.redisctl_key_count)
            .field("account_name", &self.account_name)
            .field("capi_newly_enabled", &self.capi_newly_enabled)
            .field("superseded_revoked", &self.superseded_revoked)
            .field("superseded_key_name", &self.superseded_key_name)
            .field("accounts", &self.accounts)
            .finish()
    }
}

/// Ties the OIDC endpoints (Okta) and the SM API together for one environment.
#[derive(Clone)]
pub struct CloudAuthenticator {
    issuer: Url,
    client_id: String,
    sm_api_url: Url,
    capi_url: String,
    http: reqwest::Client,
}

impl CloudAuthenticator {
    /// Build for one environment. `issuer`/`client_id` drive the Okta flows, `sm_api_url` the
    /// key-minting exchange, and `capi_url` is recorded in the resulting profile.
    pub fn new(
        issuer: Url,
        client_id: impl Into<String>,
        sm_api_url: Url,
        capi_url: impl Into<String>,
    ) -> Self {
        Self {
            issuer,
            client_id: client_id.into(),
            sm_api_url,
            capi_url: capi_url.into(),
            http: default_http_client(),
        }
    }

    /// Use a caller-provided reqwest client (tests / shared client).
    pub fn with_http_client(mut self, http: reqwest::Client) -> Self {
        self.http = http;
        self
    }

    /// Device-authorization-grant client for headless / agent logins. The flow runs on the
    /// `oauth2` crate's own HTTP stack, so it does not share this authenticator's SM client.
    pub fn device(&self) -> DeviceFlowClient {
        DeviceFlowClient::new(self.issuer.clone(), self.client_id.clone())
    }

    /// Auth-code + PKCE loopback client for interactive human logins.
    pub fn loopback(&self) -> LoopbackFlowClient {
        LoopbackFlowClient::new(self.issuer.clone(), self.client_id.clone())
    }

    /// Refresh an Okta refresh token for a fresh token set (Okta rotates it). The grant is
    /// flow-agnostic, so it goes straight through `oidc` rather than a specific flow client.
    pub async fn refresh(&self, refresh_token: &str) -> Result<TokenSet, AuthError> {
        super::oidc::refresh(&self.issuer, &self.client_id, refresh_token).await
    }

    /// Invalidate a stored refresh token at the identity provider.
    pub async fn revoke_refresh_token(&self, refresh_token: &str) -> Result<(), AuthError> {
        super::oidc::revoke_refresh_token(&self.issuer, &self.client_id, refresh_token).await
    }

    /// List the accounts the signed-in user belongs to, minting nothing and switching nothing.
    pub async fn list_accounts<F>(
        &self,
        tokens: &TokenSet,
        mut mfa_prompt: F,
    ) -> Result<AccountListing, AuthError>
    where
        F: FnMut(&[String], u32) -> Result<Option<String>, AuthError>,
    {
        let mut sm = SmApiClient::with_http_client(
            self.sm_api_url.clone(),
            self.http.clone(),
            LoginFlow::Switch,
        );
        match sm.login(&tokens.access_token, None).await {
            Ok(()) => {}
            Err(AuthError::MfaRequired { factors }) => {
                self.satisfy_mfa(&mut sm, tokens, &factors, &mut mfa_prompt)
                    .await?
            }
            Err(e) => return Err(e),
        }
        let user = sm.fetch_current_user().await?;
        let current = user
            .current_account_id
            .as_deref()
            .and_then(|s| s.parse::<u64>().ok());
        Ok(AccountListing {
            email: user.email,
            accounts: login_accounts(&sm.fetch_accounts().await?),
            session_account: current,
        })
    }

    /// Revoke a minted CAPI key by name from `account_id`, using a session established from
    /// `tokens`.
    ///
    /// Returns whether a key of that name was found. Both the listing and the delete are scoped
    /// to the session's account, so the caller has to say which account holds the key: a sign-in
    /// starts on the user's server-side default, which is not necessarily the one a profile's key
    /// was minted for. Passing `None` searches wherever the session lands, which is all an older
    /// profile that recorded no account can do.
    pub async fn revoke_capi_key(
        &self,
        tokens: &TokenSet,
        account_id: Option<u64>,
        key_name: &str,
    ) -> Result<bool, AuthError> {
        let mut sm = SmApiClient::with_http_client(
            self.sm_api_url.clone(),
            self.http.clone(),
            LoginFlow::Switch,
        );
        sm.login(&tokens.access_token, None).await?;
        // A fresh sign-in starts on the user's server-side default account, and both the listing
        // and the delete are scoped to the session's account. Without this, revoking a key that
        // belongs to any other account looks like a key that is already gone.
        if let Some(account_id) = account_id {
            set_current_account_verified(&sm, account_id).await?;
        }
        let entries = sm.fetch_capi_key_entries().await?;
        let Some((id, _)) = entries.iter().find(|(_, name)| name == key_name) else {
            // Same diagnostic the login/switch path logs on a miss: without the names, "not
            // found" gives the reader nothing to compare against.
            tracing::warn!(
                "key {key_name} is not on {}; it holds: {}",
                match account_id {
                    Some(account) => format!("account {account}"),
                    None => "the account this sign-in defaults to".to_string(),
                },
                entries
                    .iter()
                    .map(|(_, name)| name.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            );
            return Ok(false);
        };
        sm.delete_capi_key(*id).await?;
        Ok(true)
    }

    /// Given tokens from a flow, run the SM exchange and mint a CAPI key named `key_name`.
    ///
    /// Propagates [`AuthError::MfaRequired`] if the account is MFA-protected; use
    /// [`CloudAuthenticator::complete_login_with_mfa`] to supply codes interactively.
    pub async fn complete_login(
        &self,
        tokens: &TokenSet,
        key_name: &str,
        flow: LoginFlow,
        account: AccountChoice,
    ) -> Result<MintedCredentials, AuthError> {
        self.complete_login_with_mfa(tokens, key_name, flow, account, None, |_, _| Ok(None))
            .await
            .map(|(creds, _)| creds)
    }

    /// As [`CloudAuthenticator::complete_login`], but `mfa_prompt` is consulted when SM challenges
    /// the login for multi-factor authentication.
    ///
    /// `mfa_prompt(factors, attempt)` is called with the factor types SM offered (possibly empty)
    /// and a 1-based attempt number. Return `Some(code)` to submit a TOTP code, or `None` to give
    /// up — which surfaces the original [`AuthError::MfaRequired`] to the caller, the right
    /// behaviour when there is no terminal to prompt on.
    pub async fn complete_login_with_mfa<F>(
        &self,
        tokens: &TokenSet,
        key_name: &str,
        flow: LoginFlow,
        account: AccountChoice,
        superseded: Option<SupersededKey>,
        mut mfa_prompt: F,
    ) -> Result<(MintedCredentials, Option<SupersededRevoker>), AuthError>
    where
        F: FnMut(&[String], u32) -> Result<Option<String>, AuthError>,
    {
        let mut sm =
            SmApiClient::with_http_client(self.sm_api_url.clone(), self.http.clone(), flow);
        // Google/GitHub logins must not send Sm-Id-Token (SSO-only); see sm_api docs.
        match sm.login(&tokens.access_token, None).await {
            Ok(()) => {}
            Err(AuthError::MfaRequired { factors }) => {
                self.satisfy_mfa(&mut sm, tokens, &factors, &mut mfa_prompt)
                    .await?
            }
            Err(e) => return Err(e),
        }
        let mut user = sm.fetch_current_user().await?;
        let want = match account {
            AccountChoice::Current => None,
            AccountChoice::Id(id) => Some(id),
            // The picker needs the list, so fetch it here; `switch_account` re-reads it to
            // validate, which also covers ids that did not come from a picker.
            AccountChoice::Prompt(choose) => {
                let accounts = login_accounts(&sm.fetch_accounts().await?);
                let current = user
                    .current_account_id
                    .as_deref()
                    .and_then(|s| s.parse::<u64>().ok());
                Some(choose(&accounts, current)?)
            }
        };
        if let Some(want) = want {
            user = self.switch_account(&sm, user, want).await?;
        }
        // Pick the account matching the logged-in user's current_account_id; `/accounts` order
        // is not guaranteed, so there is nothing safe to fall back to.
        //
        // Settled *before* `ensure_capi_enabled`, which switches programmatic access on for the
        // account: a login that is going to refuse must not leave that behind unreported, and
        // `capi_newly_enabled` is only delivered on the success path.
        let chosen = resolve_account(
            sm.fetch_accounts().await?,
            user.current_account_id.as_deref(),
        )?
        .id;
        let capi_newly_enabled = sm.ensure_capi_enabled().await?;
        // Re-read: the account access key only exists once CAPI is on, so the entry the key
        // comes from has to be the one fetched after enabling it.
        let accounts = sm.fetch_accounts().await?;
        let all_accounts = login_accounts(&accounts);
        let account = accounts
            .into_iter()
            .find(|a| a.id == chosen)
            .ok_or_else(|| {
                AuthError::Protocol(format!(
                    "account {chosen} was no longer listed after enabling programmatic access"
                ))
            })?;
        let account_name = account.name.clone();
        // Report the account the key belongs to, taken from the same entry the key came from.
        let account_id = Some(account.id);
        let api_key = account.api_access_key.ok_or_else(|| {
            AuthError::Protocol("account has no CAPI access key after enabling CAPI".into())
        })?;
        let minted = sm.mint_capi_key(key_name, user.user_account()?).await?;
        // Best-effort: count our keys so the CLI can warn about sprawl (D5). Never fail login
        // over this — a listing error just means no warning.
        let redisctl_key_count = sm
            .fetch_capi_keys()
            .await
            .map(|keys| keys.iter().filter(|n| n.starts_with("redisctl-")).count())
            .unwrap_or(0);
        Ok((
            MintedCredentials {
                account_id,
                email: user.email,
                api_key,
                api_secret: minted.secret_key,
                api_url: self.capi_url.clone(),
                refresh_token: tokens.refresh_token.clone(),
                capi_key_name: minted.name,
                redisctl_key_count,
                account_name,
                capi_newly_enabled,
                // Nothing has been revoked yet; the caller records what the revoker reports.
                superseded_revoked: None,
                superseded_key_name: None,
                accounts: all_accounts,
            },
            superseded.map(|previous| SupersededRevoker {
                sm,
                previous,
                on: account_id,
            }),
        ))
    }

    /// Point the session at `want` before anything account-scoped happens.
    ///
    /// Verifies the switch actually took rather than assuming it: every later call resolves the
    /// account from the session, so a silent no-op here would mint the key on the wrong account.
    async fn switch_account(
        &self,
        sm: &SmApiClient,
        user: SmUser,
        want: u64,
    ) -> Result<SmUser, AuthError> {
        if user.current_account_id.as_deref() == Some(want.to_string().as_str()) {
            return Ok(user);
        }
        // Fetched here rather than reusing the later call: membership has to be checked *before*
        // `setcurrent`, while the later fetch has to come *after* `ensure_capi_enabled` to see the
        // account access key it creates. Only runs when `--account` was given.
        let accounts = sm.fetch_accounts().await?;
        if !accounts.iter().any(|a| a.id == want) {
            if accounts.is_empty() {
                return Err(AuthError::Protocol(
                    "this login is not associated with any Redis Cloud account, so there is \
                     nothing to switch to"
                        .into(),
                ));
            }
            return Err(AuthError::UnknownAccount {
                requested: want,
                available: account_labels(&accounts),
            });
        }
        set_current_account_verified(sm, want).await
    }

    /// Drive the MFA retry loop against an already-challenged client.
    async fn satisfy_mfa<F>(
        &self,
        sm: &mut SmApiClient,
        tokens: &TokenSet,
        factors: &[String],
        mfa_prompt: &mut F,
    ) -> Result<(), AuthError>
    where
        F: FnMut(&[String], u32) -> Result<Option<String>, AuthError>,
    {
        for attempt in 1..=MFA_MAX_ATTEMPTS {
            let Some(code) = mfa_prompt(factors, attempt)? else {
                // Caller can't prompt (no TTY / agent): report the challenge, not a failure.
                return Err(AuthError::MfaRequired {
                    factors: factors.to_vec(),
                });
            };
            match sm.complete_mfa(&tokens.access_token, None, &code).await {
                Ok(()) => return Ok(()),
                // Wrong code: loop and let the caller re-prompt, unless attempts are spent.
                Err(AuthError::MfaInvalidCode) if attempt < MFA_MAX_ATTEMPTS => continue,
                Err(e) => return Err(e),
            }
        }
        Err(AuthError::MfaInvalidCode)
    }
}

/// How many TOTP codes a single login will accept before giving up. SM enforces its own quota
/// (`mfa-quota-exceeded`); this only bounds our prompting.
pub const MFA_MAX_ATTEMPTS: u32 = 3;

/// Choose the account matching `current_account_id` (the logged-in user context); fall back to
/// the first account only when the id is absent or not present in the list.
/// Project the API's accounts into [`LoginAccount`]s, in a stable order.
///
/// `/accounts` order is not guaranteed. Sorting here rather than at each use means the picker's
/// numbering, the printed listing and the JSON `accounts` array all agree between runs — the last
/// of which callers are told to read for ids.
fn login_accounts(accounts: &[SmAccount]) -> Vec<LoginAccount> {
    let mut out: Vec<LoginAccount> = accounts
        .iter()
        .map(|a| LoginAccount {
            id: a.id,
            name: a.name.clone(),
        })
        .collect();
    out.sort_by_key(|a| a.id);
    out
}

/// What a sign-in can reach, read without minting or switching anything.
#[derive(Debug)]
pub struct AccountListing {
    pub email: Option<String>,
    pub accounts: Vec<LoginAccount>,
    /// The account the session starts on: the user's server-side default, which is not
    /// necessarily the one a profile's key belongs to.
    pub session_account: Option<u64>,
}

/// Point the session at `want` and confirm it landed there.
///
/// A successful response does not mean the session moved, and everything afterwards resolves the
/// account from the session — so an unverified switch quietly reads and deletes on whichever
/// account the session was already on.
async fn set_current_account_verified(sm: &SmApiClient, want: u64) -> Result<SmUser, AuthError> {
    sm.set_current_account(want).await?;
    let user = sm.fetch_current_user().await?;
    // Trust the server's answer, not the request's success.
    if user.current_account_id.as_deref() != Some(want.to_string().as_str()) {
        return Err(AuthError::Protocol(format!(
            "asked Redis Cloud to switch to account {want} but the session still reports {}",
            user.current_account_id.as_deref().unwrap_or("none")
        )));
    }
    Ok(user)
}

/// Revokes the key a freshly minted one replaces, using the session that minted it.
///
/// Handed back rather than run during the mint so a caller can store the new credentials first:
/// revoking before they are safely stored can leave a profile with the old key dead and the new
/// secret lost, since Redis Cloud returns a key's secret only when it is created.
pub struct SupersededRevoker {
    sm: SmApiClient,
    previous: SupersededKey,
    /// The account the replacement was minted on, so the revoke can skip a pointless
    /// `setcurrent` when the old key lives there too.
    on: Option<u64>,
}

impl SupersededRevoker {
    /// Which key this would revoke, for a caller that wants to report it.
    pub fn key_name(&self) -> &str {
        &self.previous.key_name
    }

    /// The account holding that key. A caller comparing this with the account it just minted on
    /// can tell whether the revoke will remove a key from that same account.
    pub fn account_id(&self) -> u64 {
        self.previous.account_id
    }

    /// Best-effort: `false` means the old key may still be live, never that the new one is bad.
    pub async fn revoke(self) -> bool {
        revoke_superseded(&self.sm, &self.previous, self.on).await
    }
}

/// Revoke `previous` using the current session. The delete is scoped to the session's account, so
/// the session is pointed at `previous.account_id` unless `on` says it is already there.
async fn revoke_superseded(sm: &SmApiClient, previous: &SupersededKey, on: Option<u64>) -> bool {
    let account = previous.account_id;
    let moved = on != Some(account);
    if moved && let Err(e) = set_current_account_verified(sm, account).await {
        tracing::warn!(
            "cannot reach account {account} to revoke key {}: {e}",
            previous.key_name
        );
        return false;
    }
    // Nothing uses this session afterwards — it is dropped with the revoker — so there is no
    // need to point it back at `on`.
    delete_named_key(sm, account, &previous.key_name).await
}

async fn delete_named_key(sm: &SmApiClient, account: u64, key: &str) -> bool {
    let entries = match sm.fetch_capi_key_entries().await {
        Ok(entries) => entries,
        Err(e) => {
            tracing::warn!("cannot list keys on account {account} to revoke {key}: {e}");
            return false;
        }
    };
    let Some((id, _)) = entries.iter().find(|(_, name)| name == key) else {
        tracing::warn!(
            "key {key} is not on account {account}; it holds: {}",
            entries
                .iter()
                .map(|(_, name)| name.as_str())
                .collect::<Vec<_>>()
                .join(", ")
        );
        return false;
    };
    match sm.delete_capi_key(*id).await {
        Ok(()) => true,
        Err(e) => {
            tracing::warn!("could not delete key {key} ({id}) on account {account}: {e}");
            false
        }
    }
}

/// The account the minted key will belong to: the one the session reports as current.
///
/// The chosen entry supplies the access key, while the secret is minted in the session's own
/// account context — so guessing here pairs two halves that need not belong together, and
/// `/accounts` order is not guaranteed. One account and a session that claims nothing is not a
/// guess; anything else, and the caller has to say which.
fn resolve_account(
    mut accounts: Vec<SmAccount>,
    current_account_id: Option<&str>,
) -> Result<SmAccount, AuthError> {
    let target = current_account_id.and_then(|s| s.parse::<u64>().ok());
    if let Some(at) = target.and_then(|id| accounts.iter().position(|a| a.id == id)) {
        return Ok(accounts.swap_remove(at));
    }
    if accounts.is_empty() {
        return Err(AuthError::Protocol(
            "no accounts associated with this login".into(),
        ));
    }
    // One account and nothing claimed: there is nothing else the key could belong to. A
    // `current_account_id` we could not match is a different situation — the session is naming an
    // account this list does not have, so taking the lone entry would pair its access key with a
    // secret minted somewhere else. That is a disagreement, not silence, so it refuses below.
    if accounts.len() == 1 && current_account_id.is_none() {
        return Ok(accounts.swap_remove(0));
    }
    Err(AuthError::AccountRequired(format!(
        "this sign-in does not report which Redis Cloud account is current{}, and a key minted \
         on a guess could belong to the wrong one. Re-run with `--account <id>`; you belong to: \
         {}",
        match current_account_id {
            Some(id) => format!(" (it names {id}, which is not one of yours)"),
            None => String::new(),
        },
        account_labels(&accounts)
    )))
}

/// `Acme (#316941), #451002` — the shared rendering for every message that lists accounts.
fn account_labels(accounts: &[SmAccount]) -> String {
    login_accounts(accounts)
        .iter()
        .map(LoginAccount::label)
        .collect::<Vec<_>>()
        .join(", ")
}

#[cfg(test)]
mod tests {
    use super::*;
    use wiremock::matchers::{method, path};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    fn account(id: u64) -> SmAccount {
        serde_json::from_value(serde_json::json!({
            "id": id, "api_access_key": format!("KEY-{id}")
        }))
        .unwrap()
    }

    /// The label is shared by the CLI's account listing and the `UnknownAccount` message, so
    /// both read identically — including when the API reports no name.
    #[test]
    fn login_account_labels_named_and_unnamed_accounts() {
        assert_eq!(
            LoginAccount {
                id: 316941,
                name: Some("Acme".to_string()),
            }
            .label(),
            "Acme (#316941)"
        );
        assert_eq!(
            LoginAccount {
                id: 316941,
                name: None,
            }
            .label(),
            "#316941"
        );
    }

    #[test]
    fn resolve_account_prefers_current_account_id() {
        let accts = vec![account(111), account(222), account(333)];
        // Matches the user's current account, not the first in the list.
        let chosen = resolve_account(accts, Some("222")).unwrap();
        assert_eq!(chosen.id, 222);
    }

    /// The chosen entry supplies the access key while the secret is minted in the session's own
    /// account, so a guess can pair halves from different accounts. With more than one candidate
    /// and nothing to go on, refuse and name them.
    #[test]
    fn resolve_account_refuses_to_guess_between_several() {
        for current in [None, Some("999")] {
            let err = resolve_account(vec![account(111), account(222)], current).unwrap_err();
            let AuthError::AccountRequired(message) = err else {
                panic!("{current:?} gave {err:?}");
            };
            assert!(message.contains("#111"), "{message}");
            assert!(message.contains("#222"), "{message}");
            assert!(message.contains("--account"), "{message}");
        }
        // The unknown id is worth naming: it is the thing that did not match.
        let err = resolve_account(vec![account(111), account(222)], Some("999")).unwrap_err();
        assert!(err.to_string().contains("999"), "{err}");
    }

    /// One account and a session that claims nothing is not a guess — there is nothing else the
    /// key could belong to.
    #[test]
    fn resolve_account_takes_the_only_account() {
        assert_eq!(resolve_account(vec![account(111)], None).unwrap().id, 111);
        // And when the session does name it, by the matching path.
        assert_eq!(
            resolve_account(vec![account(111)], Some("111")).unwrap().id,
            111
        );
    }

    /// A `current_account_id` the list does not have is a disagreement, not silence: the session
    /// mints the secret in *that* account while the lone entry supplies the access key, so the
    /// two halves need not belong together. Refused even though there is only one candidate.
    #[test]
    fn resolve_account_refuses_a_lone_account_the_session_disowns() {
        for current in [Some("999"), Some("not-a-number")] {
            let err = resolve_account(vec![account(111)], current).unwrap_err();
            let AuthError::AccountRequired(message) = err else {
                panic!("{current:?} gave {err:?}");
            };
            assert!(message.contains("#111"), "{message}");
            assert!(message.contains("--account"), "{message}");
        }
    }

    #[test]
    fn resolve_account_reports_an_empty_list_as_protocol() {
        assert!(matches!(
            resolve_account(vec![], Some("1")),
            Err(AuthError::Protocol(_))
        ));
    }

    #[test]
    fn debug_redacts_secrets() {
        let creds = MintedCredentials {
            account_id: Some(42),
            email: Some("u@example.com".to_string()),
            api_key: "AKEY-visible-should-not-appear".to_string(),
            api_secret: "SECRET-should-not-appear".to_string(),
            api_url: "https://api.example.com/v1".to_string(),
            refresh_token: Some("RT-should-not-appear".to_string()),
            capi_key_name: "redisctl-cli-1".to_string(),
            redisctl_key_count: 3,
            account_name: Some("Acme".to_string()),
            capi_newly_enabled: false,
            superseded_revoked: None,
            superseded_key_name: None,
            accounts: vec![
                LoginAccount {
                    id: 316941,
                    name: Some("Acme".to_string()),
                },
                LoginAccount {
                    id: 481022,
                    name: Some("Contoso".to_string()),
                },
            ],
        };
        let dbg = format!("{creds:?}");
        assert!(dbg.contains("<redacted>"));
        assert!(!dbg.contains("AKEY-visible-should-not-appear"));
        assert!(!dbg.contains("SECRET-should-not-appear"));
        assert!(!dbg.contains("RT-should-not-appear"));
        // Non-secret fields remain visible for diagnostics.
        assert!(dbg.contains("u@example.com"));
        assert!(dbg.contains("redisctl-cli-1"));
    }

    #[tokio::test]
    async fn complete_login_runs_the_full_exchange() {
        let server = MockServer::start().await;
        let mount = |m: &str, p: &'static str, body: serde_json::Value, cookie: bool| {
            let mut tmpl = ResponseTemplate::new(200).set_body_json(body);
            if cookie {
                tmpl = tmpl.append_header("Set-Cookie", "JSESSIONID=SID; Path=/");
            }
            Mock::given(method(m)).and(path(p)).respond_with(tmpl)
        };
        mount("POST", "/login", serde_json::json!({}), true)
            .mount(&server)
            .await;
        mount(
            "GET",
            "/csrf",
            serde_json::json!({"csrfToken": {"csrf_token": "C"}}),
            false,
        )
        .mount(&server)
        .await;
        mount(
            "GET",
            "/users/me",
            serde_json::json!({"id": "114429", "current_account_id": "112117", "email": "u@e.com"}),
            false,
        )
        .mount(&server)
        .await;
        mount(
            "POST",
            "/accounts/cloud-api/cloudApiAccessKey",
            serde_json::json!({"cloudApiAccessKey": {"accessKey": "ACCT"}}),
            false,
        )
        .mount(&server)
        .await;
        mount(
            "GET",
            "/accounts",
            serde_json::json!({"accounts": [{"id": 112117, "api_access_key": "ACCT-KEY"}]}),
            false,
        )
        .mount(&server)
        .await;
        mount(
            "POST",
            "/accounts/cloud-api/cloudApiKeys",
            serde_json::json!({"name": "redisctl-test", "secret_key": "SECRET"}),
            false,
        )
        .mount(&server)
        .await;

        let auth = CloudAuthenticator::new(
            Url::parse("https://issuer.example/oauth2/default").unwrap(),
            "cid",
            Url::parse(&server.uri()).unwrap(),
            "https://capi.example/v1",
        );
        let tokens = TokenSet {
            access_token: "AT".into(),
            refresh_token: Some("RT".into()),
            expires_in: 3600,
        };

        let creds = auth
            .complete_login(
                &tokens,
                "redisctl-test",
                LoginFlow::Loopback,
                AccountChoice::Current,
            )
            .await
            .unwrap();
        assert_eq!(creds.api_key, "ACCT-KEY");
        assert_eq!(creds.api_secret, "SECRET");
        assert_eq!(creds.api_url, "https://capi.example/v1");
        assert_eq!(creds.account_id, Some(112117));
        assert_eq!(creds.email.as_deref(), Some("u@e.com"));
        assert_eq!(creds.refresh_token.as_deref(), Some("RT"));
        assert_eq!(creds.capi_key_name, "redisctl-test");
        // secrets must not leak via Debug
        let dbg = format!("{creds:?}");
        assert!(!dbg.contains("SECRET") && !dbg.contains("ACCT-KEY") && !dbg.contains("RT"));
    }

    fn tokens() -> TokenSet {
        TokenSet {
            access_token: "AT".to_string(),
            refresh_token: None,
            expires_in: 3600,
        }
    }

    fn authenticator(server: &MockServer) -> CloudAuthenticator {
        CloudAuthenticator::new(
            Url::parse("https://issuer.example/oauth2/default").unwrap(),
            "client",
            Url::parse(&server.uri()).unwrap(),
            "https://capi.example/v1",
        )
    }

    /// Everything the exchange needs apart from the account-shaped endpoints each test varies.
    async fn common_login_mocks(server: &MockServer) {
        Mock::given(method("POST"))
            .and(path("/login"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({}))
                    .append_header("Set-Cookie", "JSESSIONID=SID; Path=/"),
            )
            .mount(server)
            .await;
        Mock::given(method("GET"))
            .and(path("/csrf"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"csrfToken": {"csrf_token": "C"}})),
            )
            .mount(server)
            .await;
        Mock::given(method("POST"))
            .and(path("/accounts/cloud-api/cloudApiAccessKey"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"cloudApiAccessKey": {"accessKey": "ACCT"}})),
            )
            .mount(server)
            .await;
        Mock::given(method("POST"))
            .and(path("/accounts/cloud-api/cloudApiKeys"))
            .respond_with(ResponseTemplate::new(200).set_body_json(
                serde_json::json!({"name": "redisctl-test", "secret_key": "SECRET"}),
            ))
            .mount(server)
            .await;
    }

    /// `--account` has to switch the session *before* anything account-scoped runs: both
    /// `ensure_capi_enabled` and the mint resolve the account from the session, so a switch that
    /// happened afterwards would put the key on the previous account. The mocks answer
    /// `/users/me` differently before and after `setcurrent`, so the minted account is only right
    /// if the ordering held.
    #[tokio::test]
    async fn complete_login_switches_before_minting() {
        let server = MockServer::start().await;
        let switched = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        common_login_mocks(&server).await;

        // /users/me reports 111 until setcurrent runs, then 222.
        let flag = switched.clone();
        Mock::given(method("GET"))
            .and(path("/users/me"))
            .respond_with(move |_: &wiremock::Request| {
                let id = if flag.load(std::sync::atomic::Ordering::SeqCst) {
                    "222"
                } else {
                    "111"
                };
                ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "1", "current_account_id": id, "email": "u@e.com"
                }))
            })
            .mount(&server)
            .await;
        let flag = switched.clone();
        Mock::given(method("POST"))
            .and(path("/accounts/setcurrent/222"))
            .respond_with(move |_: &wiremock::Request| {
                flag.store(true, std::sync::atomic::Ordering::SeqCst);
                ResponseTemplate::new(200).set_body_json(serde_json::json!({}))
            })
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/accounts"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "accounts": [
                    {"id": 111, "name": "One", "api_access_key": "KEY-111"},
                    {"id": 222, "name": "Two", "api_access_key": "KEY-222"}
                ]
            })))
            .mount(&server)
            .await;

        let creds = authenticator(&server)
            .complete_login(
                &tokens(),
                "redisctl-test",
                LoginFlow::Loopback,
                AccountChoice::Id(222),
            )
            .await
            .unwrap();
        // The key, the reported id and the reported name all describe the requested account.
        assert_eq!(creds.account_id, Some(222));
        assert_eq!(creds.account_name.as_deref(), Some("Two"));
        assert_eq!(creds.api_key, "KEY-222");
        assert_eq!(creds.account_count(), 2);
    }

    /// The picker runs mid-exchange, on the one session. It must see every account in a stable
    /// order (the API does not promise one) along with the session's current account, and its
    /// answer must be what gets minted.
    #[tokio::test]
    async fn complete_login_mints_what_the_picker_chose() {
        let server = MockServer::start().await;
        let switched = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        common_login_mocks(&server).await;

        let flag = switched.clone();
        Mock::given(method("GET"))
            .and(path("/users/me"))
            .respond_with(move |_: &wiremock::Request| {
                let id = if flag.load(std::sync::atomic::Ordering::SeqCst) {
                    "111"
                } else {
                    "222"
                };
                ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "1", "current_account_id": id, "email": "u@e.com"
                }))
            })
            .mount(&server)
            .await;
        let flag = switched.clone();
        Mock::given(method("POST"))
            .and(path("/accounts/setcurrent/111"))
            .respond_with(move |_: &wiremock::Request| {
                flag.store(true, std::sync::atomic::Ordering::SeqCst);
                ResponseTemplate::new(200).set_body_json(serde_json::json!({}))
            })
            .mount(&server)
            .await;
        // Deliberately returned highest-id-first, so a stable order cannot come from the API.
        Mock::given(method("GET"))
            .and(path("/accounts"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "accounts": [
                    {"id": 222, "name": "Two", "api_access_key": "KEY-222"},
                    {"id": 111, "name": "One", "api_access_key": "KEY-111"}
                ]
            })))
            .mount(&server)
            .await;

        let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let seen_current = std::sync::Arc::new(std::sync::Mutex::new(None));
        let (s2, c2) = (seen.clone(), seen_current.clone());
        let creds = authenticator(&server)
            .complete_login(
                &tokens(),
                "k",
                LoginFlow::Switch,
                AccountChoice::Prompt(Box::new(move |accounts, current| {
                    *s2.lock().unwrap() = accounts.iter().map(|a| a.id).collect::<Vec<_>>();
                    *c2.lock().unwrap() = current;
                    Ok(111)
                })),
            )
            .await
            .unwrap();

        // Sorted by id despite the API's order, so a positional choice is stable between runs.
        assert_eq!(*seen.lock().unwrap(), vec![111, 222]);
        // The session's account is passed through for context.
        assert_eq!(*seen_current.lock().unwrap(), Some(222));
        // And the picker's answer is what got minted.
        assert_eq!(creds.account_id, Some(111));
        assert_eq!(creds.api_key, "KEY-111");
        assert_eq!(creds.account_label(), "One (#111)");
    }

    /// Refusing at the picker abandons the switch instead of minting something unasked for.
    #[tokio::test]
    async fn complete_login_propagates_a_declined_picker() {
        let server = MockServer::start().await;
        common_login_mocks(&server).await;
        Mock::given(method("GET"))
            .and(path("/users/me"))
            .respond_with(ResponseTemplate::new(200).set_body_json(
                serde_json::json!({"id": "1", "current_account_id": "111", "email": "u@e.com"}),
            ))
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/accounts"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "accounts": [{"id": 111, "name": "One", "api_access_key": "KEY-111"}]
            })))
            .mount(&server)
            .await;
        let err = authenticator(&server)
            .complete_login(
                &tokens(),
                "k",
                LoginFlow::Switch,
                AccountChoice::Prompt(Box::new(|_, _| {
                    Err(AuthError::AccountRequired("declined".into()))
                })),
            )
            .await
            .unwrap_err();
        assert!(matches!(err, AuthError::AccountRequired(_)), "got {err:?}");
    }

    /// Listing must not mint a key or move the session: only the read endpoints are mounted, so
    /// an attempt at either would 404 and fail the test.
    #[tokio::test]
    async fn list_accounts_reads_without_minting_or_switching() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/login"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({}))
                    .append_header("Set-Cookie", "JSESSIONID=SID; Path=/"),
            )
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/csrf"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(serde_json::json!({"csrfToken": {"csrf_token": "C"}})),
            )
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/users/me"))
            .respond_with(ResponseTemplate::new(200).set_body_json(
                serde_json::json!({"id": "1", "current_account_id": "222", "email": "u@e.com"}),
            ))
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/accounts"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "accounts": [
                    {"id": 222, "name": "Two"},
                    {"id": 111, "name": "One"}
                ]
            })))
            .mount(&server)
            .await;

        let listing = authenticator(&server)
            .list_accounts(&tokens(), |_, _| Ok(None))
            .await
            .unwrap();

        // Sorted, so the numbering a caller reads is stable between runs.
        assert_eq!(
            listing.accounts.iter().map(|a| a.id).collect::<Vec<_>>(),
            vec![111, 222]
        );
        assert_eq!(listing.session_account, Some(222));
        assert_eq!(listing.email.as_deref(), Some("u@e.com"));
    }

    /// An account the user does not belong to is refused before any switch is attempted, and the
    /// message names the accounts they do have.
    #[tokio::test]
    async fn complete_login_refuses_an_account_the_user_is_not_in() {
        let server = MockServer::start().await;
        common_login_mocks(&server).await;
        Mock::given(method("GET"))
            .and(path("/users/me"))
            .respond_with(ResponseTemplate::new(200).set_body_json(
                serde_json::json!({"id": "1", "current_account_id": "111", "email": "u@e.com"}),
            ))
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/accounts"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "accounts": [{"id": 111, "name": "One", "api_access_key": "KEY-111"}]
            })))
            .mount(&server)
            .await;
        // No /accounts/setcurrent/* mock is mounted: reaching one would 404 and surface as a
        // different error, which is itself the assertion that no switch was attempted.
        match authenticator(&server)
            .complete_login(&tokens(), "k", LoginFlow::Loopback, AccountChoice::Id(999))
            .await
        {
            Err(AuthError::UnknownAccount {
                requested,
                available,
            }) => {
                assert_eq!(requested, 999);
                assert_eq!(available, "One (#111)");
            }
            other => panic!("expected UnknownAccount, got {other:?}"),
        }
    }

    /// Both shapes of "the session will not say which account is current" have to stop before
    /// the mint: a key's secret is returned once, so one minted on a guess cannot be recovered
    /// or paired with the right account afterwards.
    ///
    /// And before `ensure_capi_enabled`, which switches programmatic access on for the account.
    /// A refusal must not leave that behind: `capi_newly_enabled` only reaches the caller on the
    /// success path, so enabling it here would change the account with nothing saying so.
    #[tokio::test]
    async fn complete_login_mints_nothing_when_the_account_is_ambiguous() {
        async fn attempt(current: Option<&str>) -> (AuthError, usize) {
            let server = MockServer::start().await;
            common_login_mocks(&server).await;

            let mut me = serde_json::json!({"id": "1", "email": "u@e.com"});
            if let Some(current) = current {
                me["current_account_id"] = serde_json::json!(current);
            }
            Mock::given(method("GET"))
                .and(path("/users/me"))
                .respond_with(ResponseTemplate::new(200).set_body_json(me))
                .mount(&server)
                .await;
            Mock::given(method("GET"))
                .and(path("/accounts"))
                .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "accounts": [
                        {"id": 111, "name": "One", "api_access_key": "KEY-111"},
                        {"id": 222, "name": "Two", "api_access_key": "KEY-222"}
                    ]
                })))
                .mount(&server)
                .await;

            let err = authenticator(&server)
                .complete_login(&tokens(), "k", LoginFlow::Loopback, AccountChoice::Current)
                .await
                .expect_err("an ambiguous account must not complete");
            let requests = server.received_requests().await.unwrap_or_default();
            let hits = |path: &str| requests.iter().filter(|r| r.url.path() == path).count();
            assert_eq!(
                hits("/accounts/cloud-api/cloudApiAccessKey"),
                0,
                "{current:?} enabled programmatic access before refusing"
            );
            (err, hits("/accounts/cloud-api/cloudApiKeys"))
        }

        for current in [None, Some("999")] {
            let (err, mints) = attempt(current).await;
            assert!(
                matches!(err, AuthError::AccountRequired(_)),
                "{current:?} gave {err:?}"
            );
            assert_eq!(mints, 0, "{current:?} minted a key anyway");
        }
    }

    /// A `setcurrent` that reports success but leaves the session on the old account must fail
    /// loudly: continuing would mint the key on the wrong account and report success.
    #[tokio::test]
    async fn complete_login_fails_when_the_switch_does_not_take() {
        let server = MockServer::start().await;
        common_login_mocks(&server).await;
        // Never changes, however many times it is asked.
        Mock::given(method("GET"))
            .and(path("/users/me"))
            .respond_with(ResponseTemplate::new(200).set_body_json(
                serde_json::json!({"id": "1", "current_account_id": "111", "email": "u@e.com"}),
            ))
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path("/accounts/setcurrent/222"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({})))
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/accounts"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "accounts": [
                    {"id": 111, "name": "One", "api_access_key": "KEY-111"},
                    {"id": 222, "name": "Two", "api_access_key": "KEY-222"}
                ]
            })))
            .mount(&server)
            .await;
        let err = authenticator(&server)
            .complete_login(&tokens(), "k", LoginFlow::Loopback, AccountChoice::Id(222))
            .await
            .unwrap_err();
        assert!(
            matches!(err, AuthError::Protocol(ref m) if m.contains("still reports")),
            "expected a switch-verification failure, got {err:?}"
        );
    }

    /// Asking for the account the session is already on must not issue a switch at all.
    #[tokio::test]
    async fn complete_login_skips_the_switch_when_already_current() {
        let server = MockServer::start().await;
        common_login_mocks(&server).await;
        Mock::given(method("GET"))
            .and(path("/users/me"))
            .respond_with(ResponseTemplate::new(200).set_body_json(
                serde_json::json!({"id": "1", "current_account_id": "111", "email": "u@e.com"}),
            ))
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/accounts"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "accounts": [{"id": 111, "name": "One", "api_access_key": "KEY-111"}]
            })))
            .mount(&server)
            .await;
        // Again, no setcurrent mount: if one were issued it would 404 and fail the login.
        let creds = authenticator(&server)
            .complete_login(&tokens(), "k", LoginFlow::Loopback, AccountChoice::Id(111))
            .await
            .unwrap();
        assert_eq!(creds.account_id, Some(111));
    }

    /// The mint must not revoke anything: the caller stores the new credentials first, and only
    /// then runs the revoker. No DELETE is mounted here, so revoking during the mint would 404.
    #[tokio::test]
    async fn the_mint_defers_revocation_to_the_caller() {
        let server = MockServer::start().await;
        common_login_mocks(&server).await;
        Mock::given(method("GET"))
            .and(path("/users/me"))
            .respond_with(ResponseTemplate::new(200).set_body_json(
                serde_json::json!({"id": "1", "current_account_id": "111", "email": "u@e.com"}),
            ))
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/accounts"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "accounts": [{"id": 111, "name": "One", "api_access_key": "KEY-111"}]
            })))
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/accounts/cloud-api/cloudApiKeys"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "cloudApiKeys": [{"id": 7, "name": "redisctl-cli-1"}]
            })))
            .mount(&server)
            .await;

        let (creds, revoker) = authenticator(&server)
            .complete_login_with_mfa(
                &tokens(),
                "redisctl-cli-2",
                LoginFlow::Loopback,
                AccountChoice::Current,
                Some(SupersededKey {
                    account_id: 111,
                    key_name: "redisctl-cli-1".to_string(),
                }),
                |_, _| Ok(None),
            )
            .await
            .unwrap();
        // Credentials are complete and usable, and nothing has been taken away yet.
        assert!(!creds.api_secret.is_empty());
        assert_eq!(creds.superseded_revoked, None);
        let revoker = revoker.expect("a superseded key was given, so a revoker comes back");
        assert_eq!(revoker.key_name(), "redisctl-cli-1");

        // Firing it is what issues the delete. It points the session at the key's account
        // first, which is why that is mounted only now too.
        Mock::given(method("POST"))
            .and(path("/accounts/setcurrent/111"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({})))
            .mount(&server)
            .await;
        Mock::given(method("DELETE"))
            .and(path("/accounts/cloud-api/cloudApiKeys/7"))
            .respond_with(ResponseTemplate::new(200))
            .expect(1)
            .mount(&server)
            .await;
        assert!(revoker.revoke().await);
    }

    /// Revoking across accounts has to reach the other account, which is the reason the revoker
    /// keeps the session rather than the caller signing in again. The switch back is verified
    /// too, so a session that never moved cannot delete from the wrong account.
    #[tokio::test]
    async fn revoking_across_accounts_reaches_the_other_account() {
        let server = MockServer::start().await;
        let cell = mock_session(&server, 111, vec![111, 222]).await;
        mock_keys(&server, cell, vec![(111, 9, "redisctl-cli-1")]).await;
        Mock::given(method("GET"))
            .and(path("/accounts"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "accounts": [
                    {"id": 111, "name": "One", "api_access_key": "KEY-111"},
                    {"id": 222, "name": "Two", "api_access_key": "KEY-222"}
                ]
            })))
            .mount(&server)
            .await;
        Mock::given(method("DELETE"))
            .and(path("/accounts/cloud-api/cloudApiKeys/9"))
            .respond_with(ResponseTemplate::new(200))
            .expect(1)
            .named("delete the superseded key on the account that holds it")
            .mount(&server)
            .await;

        let (creds, revoker) = authenticator(&server)
            .complete_login_with_mfa(
                &tokens(),
                "redisctl-cli-2",
                LoginFlow::Switch,
                AccountChoice::Id(222),
                Some(SupersededKey {
                    account_id: 111,
                    key_name: "redisctl-cli-1".to_string(),
                }),
                |_, _| Ok(None),
            )
            .await
            .unwrap();
        assert_eq!(creds.account_id, Some(222));
        assert!(revoker.unwrap().revoke().await);
    }

    type AccountCell = std::sync::Arc<std::sync::atomic::AtomicU64>;

    fn account_of(cell: &AccountCell) -> u64 {
        cell.load(std::sync::atomic::Ordering::SeqCst)
    }

    /// A mock session that reports whichever account it was last switched *to*, and only moves
    /// for the ids in `honoured`.
    ///
    /// An empty `honoured` models what the real API can do, and what the verification exists for:
    /// answer 200 and leave the session where it was. Returns the cell so a test can mount
    /// account-dependent mocks of its own.
    async fn mock_session(server: &MockServer, starts_on: u64, honoured: Vec<u64>) -> AccountCell {
        let current: AccountCell =
            std::sync::Arc::new(std::sync::atomic::AtomicU64::new(starts_on));
        common_login_mocks(server).await;

        let cell = current.clone();
        Mock::given(method("GET"))
            .and(path("/users/me"))
            .respond_with(move |_: &wiremock::Request| {
                ResponseTemplate::new(200).set_body_json(serde_json::json!({
                    "id": "1", "current_account_id": account_of(&cell).to_string(),
                    "email": "u@e.com"
                }))
            })
            .mount(server)
            .await;

        let cell = current.clone();
        Mock::given(method("POST"))
            .and(wiremock::matchers::path_regex(
                r"^/accounts/setcurrent/\d+$",
            ))
            .respond_with(move |req: &wiremock::Request| {
                if let Some(asked) = req
                    .url
                    .path()
                    .rsplit('/')
                    .next()
                    .and_then(|s| s.parse::<u64>().ok())
                    && honoured.contains(&asked)
                {
                    cell.store(asked, std::sync::atomic::Ordering::SeqCst);
                }
                ResponseTemplate::new(200).set_body_json(serde_json::json!({}))
            })
            .mount(server)
            .await;

        current
    }

    /// Keys per account, so the listing shows what the session's account actually holds.
    async fn mock_keys(server: &MockServer, cell: AccountCell, per_account: Vec<(u64, u64, &str)>) {
        let owned: Vec<(u64, u64, String)> = per_account
            .into_iter()
            .map(|(account, id, name)| (account, id, name.to_string()))
            .collect();
        Mock::given(method("GET"))
            .and(path("/accounts/cloud-api/cloudApiKeys"))
            .respond_with(move |_: &wiremock::Request| {
                let here = account_of(&cell);
                let keys: Vec<_> = owned
                    .iter()
                    .filter(|(account, _, _)| *account == here)
                    .map(|(_, id, name)| serde_json::json!({"id": id, "name": name}))
                    .collect();
                ResponseTemplate::new(200).set_body_json(serde_json::json!({"cloudApiKeys": keys}))
            })
            .mount(server)
            .await;
    }

    /// Logout revokes the key the profile recorded, which may not be on the account the sign-in
    /// lands on. The listing only shows it once the switch lands, so skipping the switch reads
    /// the wrong account rather than merely omitting a request.
    #[tokio::test]
    async fn revoking_by_name_points_the_session_at_the_key_account() {
        let server = MockServer::start().await;
        let cell = mock_session(&server, 111, vec![222]).await;
        mock_keys(
            &server,
            cell,
            vec![
                (111, 9, "a-key-on-the-default-account"),
                (222, 5, "redisctl-cli-on-222"),
            ],
        )
        .await;
        Mock::given(method("DELETE"))
            .and(path("/accounts/cloud-api/cloudApiKeys/5"))
            .respond_with(ResponseTemplate::new(200))
            .expect(1)
            .mount(&server)
            .await;

        assert!(
            authenticator(&server)
                .revoke_capi_key(&tokens(), Some(222), "redisctl-cli-on-222")
                .await
                .unwrap(),
            "the key on the recorded account should be found and deleted"
        );
    }

    /// `setcurrent` can answer 200 and leave the session where it was. Acting on that would read
    /// and delete on the wrong account — and where that account holds a key of the same name,
    /// delete someone else's working key. It has to fail instead.
    #[tokio::test]
    async fn revoking_by_name_refuses_when_the_switch_does_not_take() {
        let server = MockServer::start().await;
        let cell = mock_session(&server, 111, vec![]).await;
        mock_keys(&server, cell, vec![(111, 9, "redisctl-cli-shared-name")]).await;
        Mock::given(method("DELETE"))
            .and(wiremock::matchers::path_regex(
                r"^/accounts/cloud-api/cloudApiKeys/\d+$",
            ))
            .respond_with(ResponseTemplate::new(200))
            .expect(0)
            .named("nothing may be deleted from an account we did not reach")
            .mount(&server)
            .await;

        let err = authenticator(&server)
            .revoke_capi_key(&tokens(), Some(222), "redisctl-cli-shared-name")
            .await
            .expect_err("an unverified switch must not be treated as success");
        assert!(
            format!("{err}").contains("still reports"),
            "the error should say the session did not move, got: {err}"
        );
    }

    /// The switch *back* to the superseded key's account is verified too. Here only the switch to
    /// 222 lands, so the return to 111 answers 200 without moving — and a revoke that cannot
    /// reach the account must report failure rather than delete from wherever it ended up.
    #[tokio::test]
    async fn revoking_across_accounts_refuses_when_the_switch_back_does_not_take() {
        let server = MockServer::start().await;
        let cell = mock_session(&server, 111, vec![222]).await;
        mock_keys(
            &server,
            cell,
            vec![(111, 9, "redisctl-cli-1"), (222, 7, "redisctl-cli-1")],
        )
        .await;
        Mock::given(method("GET"))
            .and(path("/accounts"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "accounts": [
                    {"id": 111, "name": "One", "api_access_key": "KEY-111"},
                    {"id": 222, "name": "Two", "api_access_key": "KEY-222"}
                ]
            })))
            .mount(&server)
            .await;
        // 222 holds a key of the same name. Deleting that one would be deleting the key this
        // login just minted for, on the wrong account.
        Mock::given(method("DELETE"))
            .and(wiremock::matchers::path_regex(
                r"^/accounts/cloud-api/cloudApiKeys/\d+$",
            ))
            .respond_with(ResponseTemplate::new(200))
            .expect(0)
            .named("nothing may be deleted from an account we did not reach")
            .mount(&server)
            .await;

        let (_, revoker) = authenticator(&server)
            .complete_login_with_mfa(
                &tokens(),
                "redisctl-cli-2",
                LoginFlow::Switch,
                AccountChoice::Id(222),
                Some(SupersededKey {
                    account_id: 111,
                    key_name: "redisctl-cli-1".to_string(),
                }),
                |_, _| Ok(None),
            )
            .await
            .unwrap();
        assert!(
            !revoker.unwrap().revoke().await,
            "a switch that did not land must not be reported as a revocation"
        );
    }

    /// Revocation deletes the key the profile recorded and nothing else. An account can hold other
    /// `redisctl-*` keys — a second machine, a second profile — and the names here are chosen so a
    /// prefix or substring match would take a neighbour.
    #[tokio::test]
    async fn revocation_targets_the_recorded_key_alone() {
        let server = MockServer::start().await;
        let cell = mock_session(&server, 111, vec![]).await;
        mock_keys(
            &server,
            cell,
            vec![
                (111, 1, "redisctl-cli-0"),
                (111, 2, "redisctl-cli-11"),
                (111, 3, "redisctl-cli-1"),
                (111, 4, "someone-elses-key"),
            ],
        )
        .await;
        // Only id 3 may be deleted; any other id has no mock and would 404.
        Mock::given(method("DELETE"))
            .and(path("/accounts/cloud-api/cloudApiKeys/3"))
            .respond_with(ResponseTemplate::new(200))
            .expect(1)
            .named("delete the recorded key, by exact name")
            .mount(&server)
            .await;

        assert!(
            authenticator(&server)
                .revoke_capi_key(&tokens(), None, "redisctl-cli-1")
                .await
                .unwrap()
        );
    }

    /// A recorded key that is not on the account deletes nothing and is reported as not revoked.
    /// Guessing which key was meant is the one thing worse than saying so.
    #[tokio::test]
    async fn revocation_deletes_nothing_when_the_recorded_key_is_gone() {
        let server = MockServer::start().await;
        let cell = mock_session(&server, 111, vec![]).await;
        mock_keys(&server, cell, vec![(111, 9, "a-key-someone-else-minted")]).await;
        Mock::given(method("DELETE"))
            .and(wiremock::matchers::path_regex(
                r"^/accounts/cloud-api/cloudApiKeys/\d+$",
            ))
            .respond_with(ResponseTemplate::new(200))
            .expect(0)
            .named("nothing may be deleted when the recorded key is absent")
            .mount(&server)
            .await;

        assert!(
            !authenticator(&server)
                .revoke_capi_key(&tokens(), None, "redisctl-cli-1")
                .await
                .unwrap(),
            "a key that is not there cannot be reported as revoked"
        );
    }

    /// A profile that recorded no account — written before the id was stored — searches wherever
    /// the session lands, which is all it can do. It must not issue a switch to nowhere.
    #[tokio::test]
    async fn revoking_by_name_without_an_account_does_not_switch() {
        let server = MockServer::start().await;
        common_login_mocks(&server).await;
        Mock::given(method("GET"))
            .and(path("/accounts/cloud-api/cloudApiKeys"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "cloudApiKeys": [{"id": 3, "name": "redisctl-cli-somewhere"}]
            })))
            .mount(&server)
            .await;
        Mock::given(method("DELETE"))
            .and(path("/accounts/cloud-api/cloudApiKeys/3"))
            .respond_with(ResponseTemplate::new(200))
            .expect(1)
            .mount(&server)
            .await;

        assert!(
            authenticator(&server)
                .revoke_capi_key(&tokens(), None, "redisctl-cli-somewhere")
                .await
                .unwrap()
        );
        assert!(
            !server
                .received_requests()
                .await
                .unwrap()
                .iter()
                .any(|r| r.url.path().starts_with("/accounts/setcurrent/")),
            "no account was recorded, so there is nothing to switch to"
        );
    }

    #[tokio::test]
    async fn complete_login_errors_when_login_rejected() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/login"))
            .respond_with(ResponseTemplate::new(401).append_header("Set-Cookie", "JSESSIONID=S"))
            .mount(&server)
            .await;
        let auth = CloudAuthenticator::new(
            Url::parse("https://issuer.example/oauth2/default").unwrap(),
            "cid",
            Url::parse(&server.uri()).unwrap(),
            "https://capi.example/v1",
        );
        let tokens = TokenSet {
            access_token: "AT".into(),
            refresh_token: None,
            expires_in: 3600,
        };
        assert!(matches!(
            auth.complete_login(&tokens, "k", LoginFlow::Loopback, AccountChoice::Current)
                .await,
            Err(AuthError::Protocol(_))
        ));
    }
}