aster_forge_test/
redis.rs

1//! Shared reusable Redis container for integration tests.
2//!
3//! The container is shared by suite name and reused across runs, so data persists between test
4//! processes. Tests should use unique key prefixes or clean up after themselves.
5
6use crate::state::{ContainerLease, ContainerStateLock};
7use crate::suite::TestContainerSuite;
8use crate::wait::wait_until;
9use std::net::{SocketAddr, TcpListener, TcpStream};
10use std::time::Duration;
11use testcontainers::core::{ContainerAsync, IntoContainerPort, WaitFor};
12use testcontainers::{GenericImage, ImageExt, ReuseDirective, runners::AsyncRunner};
13
14/// Handle to the suite's shared Redis container.
15pub struct RedisTestContainer {
16    url: String,
17    address: SocketAddr,
18    _container: ContainerAsync<GenericImage>,
19    _lease: ContainerLease,
20}
21
22impl RedisTestContainer {
23    /// Starts (or reuses) the shared Redis container and waits for it to accept connections.
24    pub async fn start(suite: &TestContainerSuite) -> Self {
25        // Keep the host port fixed across stop/start. Docker assigns a new ephemeral port to a
26        // container whose mapping leaves HostPort empty, stranding already-running processes on
27        // the old Redis endpoint after a restart.
28        let lock = ContainerStateLock::acquire(suite, "redis-fixed");
29        let mut state = lock.load();
30        let _ = state.prune_stale();
31        state.register_pid(std::process::id());
32        lock.save(&state);
33        let host_port = TcpListener::bind(("127.0.0.1", 0))
34            .expect("reserve Redis test host port")
35            .local_addr()
36            .expect("resolve Redis test host port")
37            .port();
38
39        let container = GenericImage::new("redis", "7-alpine")
40            .with_wait_for(WaitFor::message_on_stdout("Ready to accept connections"))
41            .with_mapped_port(host_port, IntoContainerPort::tcp(6379))
42            .with_container_name(suite.container_name("redis-fixed"))
43            .with_reuse(ReuseDirective::Always)
44            .start()
45            .await
46            .expect("failed to start Redis test container");
47        let port = container
48            .get_host_port_ipv4(IntoContainerPort::tcp(6379))
49            .await
50            .expect("Redis test port should be exposed");
51        drop(lock);
52
53        let address = SocketAddr::from(([127, 0, 0, 1], port));
54        let ready = wait_until(
55            Duration::from_secs(90),
56            Duration::from_millis(250),
57            || async { TcpStream::connect_timeout(&address, Duration::from_millis(500)).is_ok() },
58        )
59        .await;
60        assert!(ready, "Redis test container did not become ready");
61
62        Self {
63            url: format!("redis://127.0.0.1:{port}/0"),
64            address,
65            _container: container,
66            _lease: ContainerLease::new(suite.clone(), "redis-fixed"),
67        }
68    }
69
70    /// Returns the Redis URL, for example `redis://127.0.0.1:6379/0`.
71    pub fn url(&self) -> &str {
72        &self.url
73    }
74
75    /// Stops Redis immediately to simulate a broker outage.
76    pub async fn stop(&self) {
77        self._container
78            .stop_with_timeout(Some(0))
79            .await
80            .expect("failed to stop Redis test container");
81    }
82
83    /// Restarts a previously stopped Redis container.
84    pub async fn restart(&self) {
85        self._container
86            .start()
87            .await
88            .expect("failed to restart Redis test container");
89        let ready = wait_until(
90            Duration::from_secs(90),
91            Duration::from_millis(250),
92            || async {
93                TcpStream::connect_timeout(&self.address, Duration::from_millis(500)).is_ok()
94            },
95        )
96        .await;
97        assert!(ready, "restarted Redis test container did not become ready");
98    }
99}
aster_forge_test/redis.rs

aster_forge_test/
redis.rs
