d-OAuth-Server/tests/topology.rs

//! `test(topology)` — Redis topology harness (task 42).
//!
//! In-process fixtures via `redis-test` (no Docker needed):
//! * `topology_sentinel_master_resolves` — a 1-master/1-replica/1-sentinel
//!   `RedisSentinelCluster` boots, the master is resolved through
//!   `SentinelClientBuilder` exactly like `state::sentinel` does, and a
//!   SET/GET round-trip flows through the resolved master.
//! * `topology_cluster_single_slot_roundtrip` — a 3-node `RedisCluster`
//!   boots and a `{state}`-tagged key round-trips through
//!   `ClusterClient` (the single-slot convention the token store relies on).
//!
//! Docker/testcontainers parity for CI environments without a `redis-server`
//! binary:
//! * `topology_single_node_store_roundtrip` — same assertions shape against a
//!   `redis:7.2-alpine` container owned by `ContainerGuard`, driving the real
//!   `Store::put_access`/`get_access` through our key layout.
//!
//! The `redis-test` legs skip gracefully (green, noted on stderr) when no
//! `redis-server` binary is on `PATH`; CI installs one via apt (see
//! `.github/workflows/ci.yml`). Bodies use `Result` + `?`
//! (`unwrap`/`expect`/`panic` are denied workspace-wide).

use std::time::{Duration, Instant};

use d_oauth::{
  config::DurabilityMode,
  state::RedisConn,
  store::{
    Store, StoreError,
    tokens::{AccessRecord, get_access, put_access},
  },
};
use redis::{
  ConnectionAddr,
  sentinel::{SentinelClientBuilder, SentinelServerType},
};
use redis_test::{
  cluster::{RedisCluster, RedisClusterConfiguration},
  sentinel::RedisSentinelCluster,
};
use testcontainers::{
  GenericImage, ImageExt,
  core::{IntoContainerPort, WaitFor},
  runners::AsyncRunner,
};

#[path = "../src/test_guard.rs"]
#[allow(dead_code)]
mod test_guard;

use test_guard::ContainerGuard;

/// Test-body error.
#[derive(Debug, thiserror::Error)]
enum TestError {
  /// A testcontainers operation failed.
  #[error("container error: {0}")]
  Containers(#[from] testcontainers::TestcontainersError),
  /// A Redis command failed.
  #[error("redis error: {0}")]
  Redis(#[from] redis::RedisError),
  /// A store operation failed.
  #[error("store error: {0}")]
  Store(#[from] StoreError),
  /// A test assertion observed an unexpected value.
  #[error("unexpected test value: {0}")]
  Unexpected(String),
}

/// Test body result (no `unwrap`/`expect` under the deny lints).
type TestResult = Result<(), TestError>;

/// Whether the `redis-test` fixtures can spawn (they exec `redis-server`).
fn redis_server_available() -> bool {
  std::process::Command::new("redis-server")
    .arg("--version")
    .output()
    .map(|output| output.status.success())
    .unwrap_or(false)
}

/// RAII owner of a Sentinel fixture: `stop()` runs even when the test body
/// returns early via `?` (child processes must not outlive the test).
struct SentinelFixture {
  inner: Option<RedisSentinelCluster>,
}

impl SentinelFixture {
  fn start() -> Self {
    Self {
      inner: Some(RedisSentinelCluster::new(1, 1, 1)),
    }
  }
}

impl Drop for SentinelFixture {
  fn drop(&mut self) {
    if let Some(mut cluster) = self.inner.take() {
      cluster.stop();
    }
  }
}

/// RAII owner of a cluster fixture (same teardown contract as above).
struct ClusterFixture {
  inner: Option<RedisCluster>,
}

impl ClusterFixture {
  fn start() -> Self {
    Self {
      inner: Some(RedisCluster::new(RedisClusterConfiguration::default())),
    }
  }

  /// Three masters with one replica each (the node-failover harness).
  fn start_with_replicas() -> Self {
    Self {
      inner: Some(RedisCluster::new(
        RedisClusterConfiguration::single_replica_config(),
      )),
    }
  }
}

impl Drop for ClusterFixture {
  fn drop(&mut self) {
    if let Some(mut cluster) = self.inner.take() {
      cluster.stop();
    }
  }
}

/// Sentinel resolves the current master; reads/writes flow through it.
#[tokio::test]
async fn topology_sentinel_master_resolves() -> TestResult {
  if !redis_server_available() {
    eprintln!("SKIP topology_sentinel_master_resolves: no redis-server on PATH");
    return Ok(());
  }
  let mut fixture = SentinelFixture::start();
  let cluster = fixture
    .inner
    .as_ref()
    .ok_or(TestError::Unexpected("fixture missing".to_owned()))?;

  // Only the sentinel addresses are known up front — the master port is
  // discovered through Sentinel, exactly like `state::sentinel::connect`.
  let mut addrs = Vec::new();
  for server in cluster.iter_sentinel_servers() {
    match server.host_and_port() {
      Some((host, port)) => addrs.push(ConnectionAddr::Tcp(host.to_owned(), port)),
      None => return Err(TestError::Unexpected("sentinel has no address".to_owned())),
    }
  }
  if addrs.is_empty() {
    return Err(TestError::Unexpected("no sentinel addresses".to_owned()));
  }
  let mut sentinel =
    SentinelClientBuilder::new(addrs, "master0", SentinelServerType::Master)?.build()?;
  let master = sentinel.async_get_client().await?;
  let mut conn = master.get_multiplexed_async_connection().await?;

  let key = "topology-sentinel-probe";
  let _: () = redis::cmd("SET")
    .arg(key)
    .arg("master-ok")
    .query_async(&mut conn)
    .await?;
  let got: String = redis::cmd("GET").arg(key).query_async(&mut conn).await?;
  if got != "master-ok" {
    return Err(TestError::Unexpected(
      "sentinel master round-trip mismatch".to_owned(),
    ));
  }
  if let Some(mut cluster) = fixture.inner.take() {
    cluster.stop();
  }
  Ok(())
}

/// A `{state}`-tagged key round-trips on a real 3-node cluster.
#[tokio::test]
async fn topology_cluster_single_slot_roundtrip() -> TestResult {
  if !redis_server_available() {
    eprintln!("SKIP topology_cluster_single_slot_roundtrip: no redis-server on PATH");
    return Ok(());
  }
  let mut fixture = ClusterFixture::start();
  let cluster = fixture
    .inner
    .as_ref()
    .ok_or(TestError::Unexpected("fixture missing".to_owned()))?;

  let mut seeds = Vec::new();
  for server in cluster.iter_servers() {
    match server.host_and_port() {
      Some((_, port)) => seeds.push(format!("redis://127.0.0.1:{port}")),
      None => {
        return Err(TestError::Unexpected(
          "cluster node has no address".to_owned(),
        ));
      }
    }
  }
  if seeds.is_empty() {
    return Err(TestError::Unexpected("no cluster seeds".to_owned()));
  }
  let client = redis::cluster::ClusterClient::new(seeds)?;
  let mut conn = client.get_async_connection().await?;

  // The token store keeps multi-key Lua ops in one slot via the `{state}`
  // hash tag; the harness pins that the tag round-trips on a real cluster.
  let key = "oauth:{state}:topology-probe";
  let _: () = redis::cmd("SET")
    .arg(key)
    .arg("cluster-ok")
    .query_async(&mut conn)
    .await?;
  let got: String = redis::cmd("GET").arg(key).query_async(&mut conn).await?;
  if got != "cluster-ok" {
    return Err(TestError::Unexpected(
      "cluster round-trip mismatch".to_owned(),
    ));
  }
  if let Some(mut cluster) = fixture.inner.take() {
    cluster.stop();
  }
  Ok(())
}

/// Parity leg that needs only Docker: the real token-store access path
/// against a single node, so CI stays green where no `redis-server` binary
/// exists but Docker does.
#[tokio::test]
async fn topology_single_node_store_roundtrip() -> TestResult {
  let container = GenericImage::new("redis", "7.2-alpine")
    .with_exposed_port(6379.tcp())
    .with_wait_for(WaitFor::message_on_stdout("Ready to accept connections"))
    .with_cmd(["redis-server", "--appendonly", "yes"])
    .start()
    .await?;
  let guarded = ContainerGuard::new(container);
  let port = guarded.get_host_port_ipv4(6379).await?;
  let url = format!("redis://127.0.0.1:{port}");

  let client = redis::Client::open(url.as_str())?;
  let manager_config = redis::aio::ConnectionManagerConfig::new()
    .set_response_timeout(Some(Duration::from_secs(10)))
    .set_connection_timeout(Some(Duration::from_secs(5)));
  let manager = redis::aio::ConnectionManager::new_with_config(client, manager_config).await?;
  let mut conn = RedisConn::Single(manager);
  let store = Store::from_parts(
    "oauth:{state}:".to_owned(),
    "oauth:{rl}:".to_owned(),
    DurabilityMode::WaitAof,
  );

  let token_hash = "topology-parity-probe";
  let record = AccessRecord {
    client_id: "topology-client".to_owned(),
    grant_id: "topology-grant".to_owned(),
    scope: "read".to_owned(),
    subject: "user-1".to_owned(),
    cnf: String::new(),
  };
  put_access(
    &mut conn,
    &store,
    token_hash,
    &record,
    Duration::from_secs(300),
  )
  .await?;
  match get_access(&mut conn, &store, token_hash).await? {
    Some(found) => {
      if found != record {
        return Err(TestError::Unexpected(
          "stored access record mismatch".to_owned(),
        ));
      }
    }
    None => {
      return Err(TestError::Unexpected(
        "stored access record missing".to_owned(),
      ));
    }
  }
  Ok(())
}

/// The current Sentinel master port, resolved by the sentinel itself.
async fn sentinel_master_port(host: &str, port: u16) -> Result<u16, TestError> {
  let client = redis::Client::open(format!("redis://{host}:{port}"))?;
  let mut conn = client.get_multiplexed_async_connection().await?;
  let reply: Vec<String> = redis::cmd("SENTINEL")
    .arg("GET-MASTER-ADDR-BY-NAME")
    .arg("master0")
    .query_async(&mut conn)
    .await?;
  reply
    .get(1)
    .and_then(|value| value.parse::<u16>().ok())
    .ok_or_else(|| TestError::Unexpected("sentinel has no master address".to_owned()))
}

/// A real `SENTINEL FAILOVER master0` promotes the replica: the resolved
/// master changes and a write round-trips through the promoted node.
#[tokio::test]
async fn topology_sentinel_failover_promotes_replica() -> TestResult {
  if !redis_server_available() {
    eprintln!("SKIP topology_sentinel_failover_promotes_replica: no redis-server on PATH");
    return Ok(());
  }
  let mut fixture = SentinelFixture::start();
  let cluster = fixture
    .inner
    .as_ref()
    .ok_or(TestError::Unexpected("fixture missing".to_owned()))?;

  let mut sentinel_addr = None;
  for server in cluster.iter_sentinel_servers() {
    match server.host_and_port() {
      Some((host, port)) => {
        sentinel_addr.get_or_insert((host.to_owned(), port));
      }
      None => return Err(TestError::Unexpected("sentinel has no address".to_owned())),
    }
  }
  let Some((sentinel_host, sentinel_port)) = sentinel_addr else {
    return Err(TestError::Unexpected("no sentinel addresses".to_owned()));
  };

  let old_port = sentinel_master_port(&sentinel_host, sentinel_port).await?;

  let admin = redis::Client::open(format!("redis://{sentinel_host}:{sentinel_port}"))?;
  let mut admin_conn = admin.get_multiplexed_async_connection().await?;
  let _: () = redis::cmd("SENTINEL")
    .arg("FAILOVER")
    .arg("master0")
    .query_async(&mut admin_conn)
    .await?;

  let deadline = Instant::now() + Duration::from_secs(30);
  let new_port = loop {
    if let Ok(port) = sentinel_master_port(&sentinel_host, sentinel_port).await
      && port != old_port
    {
      break port;
    }
    if Instant::now() >= deadline {
      return Err(TestError::Unexpected(
        "SENTINEL FAILOVER did not promote a new master".to_owned(),
      ));
    }
    tokio::time::sleep(Duration::from_millis(500)).await;
  };

  let promoted = redis::Client::open(format!("redis://127.0.0.1:{new_port}"))?;
  let mut conn = promoted.get_multiplexed_async_connection().await?;
  let key = "topology-sentinel-failover-probe";
  let _: () = redis::cmd("SET")
    .arg(key)
    .arg("promoted-ok")
    .query_async(&mut conn)
    .await?;
  let got: String = redis::cmd("GET").arg(key).query_async(&mut conn).await?;
  if got != "promoted-ok" {
    return Err(TestError::Unexpected(
      "promoted master round-trip mismatch".to_owned(),
    ));
  }
  if let Some(mut cluster) = fixture.inner.take() {
    cluster.stop();
  }
  Ok(())
}

/// Whether a `ROLE` reply names a master.
fn role_is_master(role: &redis::Value) -> bool {
  match role {
    redis::Value::Array(items) => matches!(
      items.first(),
      Some(redis::Value::BulkString(bytes)) if bytes == b"master"
    ),
    _ => false,
  }
}

/// `CLUSTER NODES` text from one cluster seed.
async fn cluster_nodes(seed: &str) -> Result<String, TestError> {
  let client = redis::Client::open(seed)?;
  let mut conn = client.get_multiplexed_async_connection().await?;
  let text: String = redis::cmd("CLUSTER")
    .arg("NODES")
    .query_async(&mut conn)
    .await?;
  Ok(text)
}

/// `CLUSTER INFO` text from one cluster seed.
async fn cluster_info(seed: &str) -> Result<String, TestError> {
  let client = redis::Client::open(seed)?;
  let mut conn = client.get_multiplexed_async_connection().await?;
  let text: String = redis::cmd("CLUSTER")
    .arg("INFO")
    .query_async(&mut conn)
    .await?;
  Ok(text)
}

/// The slot spec of the node whose address ends in `port`, if any.
fn node_slots(nodes: &str, port: u16) -> Option<(String, String)> {
  let needle = format!(":{port}@");
  for line in nodes.lines() {
    let fields: Vec<&str> = line.split_whitespace().collect();
    let addr = fields.get(1)?;
    if !addr.contains(&needle) {
      continue;
    }
    let id = fields.first().copied().unwrap_or_default().to_owned();
    let slots = fields.get(8..).unwrap_or_default().join(" ");
    return Some((id, slots));
  }
  None
}

/// A real cluster node failure: the stopped master's slots are served by a
/// different node (its replica) and the cluster returns to `cluster_state:ok`.
#[tokio::test]
async fn topology_cluster_node_failover_heals() -> TestResult {
  if !redis_server_available() {
    eprintln!("SKIP topology_cluster_node_failover_heals: no redis-server on PATH");
    return Ok(());
  }
  let mut fixture = ClusterFixture::start_with_replicas();
  let (seeds, victim_port) = {
    let cluster = fixture
      .inner
      .as_ref()
      .ok_or(TestError::Unexpected("fixture missing".to_owned()))?;
    let mut seeds = Vec::new();
    let mut victim = None;
    for server in cluster.iter_servers() {
      let Some((host, port)) = server.host_and_port() else {
        continue;
      };
      seeds.push(format!("redis://{host}:{port}"));
      if victim.is_none() {
        let client = redis::Client::open(format!("redis://{host}:{port}"))?;
        let mut conn = client.get_multiplexed_async_connection().await?;
        let role: redis::Value = redis::cmd("ROLE").query_async(&mut conn).await?;
        if role_is_master(&role) {
          victim = Some(port);
        }
      }
    }
    let victim = victim.ok_or_else(|| TestError::Unexpected("no cluster master".to_owned()))?;
    (seeds, victim)
  };

  let survivor = seeds
    .iter()
    .find(|seed| !seed.ends_with(&format!(":{victim_port}")))
    .cloned()
    .ok_or_else(|| TestError::Unexpected("no surviving seed".to_owned()))?;
  let before = cluster_nodes(&survivor).await?;
  let (victim_id, victim_slots) = node_slots(&before, victim_port)
    .ok_or_else(|| TestError::Unexpected("victim slots not found".to_owned()))?;
  if victim_slots.is_empty() || victim_slots == "-" {
    return Err(TestError::Unexpected(
      "victim master owns no slots".to_owned(),
    ));
  }

  {
    let cluster = fixture
      .inner
      .as_mut()
      .ok_or(TestError::Unexpected("fixture missing".to_owned()))?;
    for server in &mut cluster.servers {
      if server.host_and_port().map(|(_, port)| port) == Some(victim_port) {
        server.stop();
      }
    }
  }

  let deadline = Instant::now() + Duration::from_secs(60);
  let promoted = loop {
    let healed = cluster_info(&survivor)
      .await
      .is_ok_and(|info| info.contains("cluster_state:ok"));
    if healed && let Ok(nodes) = cluster_nodes(&survivor).await {
      let victim_failed = nodes.lines().any(|line| {
        line.split_whitespace().next() == Some(victim_id.as_str())
          && line
            .split_whitespace()
            .nth(2)
            .is_some_and(|flags| flags.contains("fail"))
      });
      let first_slot = victim_slots.split_whitespace().next().unwrap_or_default();
      let moved = nodes.lines().any(|line| {
        let fields: Vec<&str> = line.split_whitespace().collect();
        let id = fields.first().copied().unwrap_or_default();
        id != victim_id && fields.get(8..).unwrap_or_default().contains(&first_slot)
      });
      if victim_failed && moved {
        break true;
      }
    }
    if Instant::now() >= deadline {
      break false;
    }
    tokio::time::sleep(Duration::from_millis(500)).await;
  };
  if !promoted {
    return Err(TestError::Unexpected(
      "cluster did not promote a replica for the failed master".to_owned(),
    ));
  }

  let client = redis::cluster::ClusterClient::new(seeds.clone())?;
  let mut conn = client.get_async_connection().await?;
  let key = "oauth:{state}:topology-failover-probe";
  let _: () = redis::cmd("SET")
    .arg(key)
    .arg("cluster-healed")
    .query_async(&mut conn)
    .await?;
  let got: String = redis::cmd("GET").arg(key).query_async(&mut conn).await?;
  if got != "cluster-healed" {
    return Err(TestError::Unexpected(
      "post-failover cluster round-trip mismatch".to_owned(),
    ));
  }
  if let Some(mut cluster) = fixture.inner.take() {
    cluster.stop();
  }
  Ok(())
}