inbuxa/inbuxa-server
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 3 Wiki Activity
Files
1ee2e2a6a3dae9cb1e76d34b8554cebd4478c280
inbuxa-server/tests/src/utils/containers.rs
T
jcoffey-dev 6a53d47106 Mark the files this fork changed (AGPL section 5(a)) 
The AGPL asks a modified version to carry prominent notices saying it was
modified, and giving a date. Publishing the source is the conveyance that
asks for it, so it wants doing before the repository is public rather than
at the release.

Every upstream file the fork changed now says so in its header, beneath the
notice it came with: 164 files, found by diffing against the upstream
snapshot branch rather than by guessing, so the list is what actually
differs. Files the fork wrote itself already carry their own copyright and
need nothing. Upstream's notices are untouched, which its licence requires
and which was already true.

The README says the same thing in prose, since the obligation is on the
work as a whole and not only its Rust files.

Builds unchanged: the server and the test binary both compile.
2026-09-19 23:48:35 -07:00

789 lines
31 KiB
Rust
Raw Blame History

/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*
* Modified by Coffey Labs in 2026 for INBUXA.
*/
use std::time::{Duration, Instant};
use testcontainers::{
ContainerAsync, GenericBuildableImage, GenericImage, ImageExt, ReuseDirective,
core::{CmdWaitFor, ExecCommand, Host, IntoContainerPort, WaitFor},
runners::{AsyncBuilder, AsyncRunner},
};
use tokio::{net::TcpStream, sync::OnceCell};
const ACME_NETWORK: &str = "stalwart-test-acme";
const READY_TIMEOUT: Duration = Duration::from_secs(180);
static FOUNDATIONDB: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static POSTGRES: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static MYSQL: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static MARIADB: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static REDIS: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static NATS: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static MINIO: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static OPENSEARCH: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static MEILISEARCH: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static KEYCLOAK: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static OPENLDAP: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static CHALLTESTSRV: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static PEBBLE: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static POWERDNS: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static SCIM_TESTER: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
// inbuxa: scale-out storage (ST-5 to ST-15): a primary and a streaming replica
static PG_PRIMARY: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
static PG_REPLICA: OnceCell<ContainerAsync<GenericImage>> = OnceCell::const_new();
const PG_REPLICATION_NETWORK: &str = "inbuxa-test-pg-replication";
pub const PG_PRIMARY_PORT: u16 = 5442;
pub const PG_REPLICA_PORT: u16 = 5443;
pub const PG_REPLICA_CONTAINER: &str = "inbuxa-test-pg-replica";
pub const PG_PRIMARY_CONTAINER: &str = "inbuxa-test-pg-primary";
static MYSQL_GTID: OnceCell<(ContainerAsync<GenericImage>, ContainerAsync<GenericImage>)> =
OnceCell::const_new();
static MYSQL_POS: OnceCell<(ContainerAsync<GenericImage>, ContainerAsync<GenericImage>)> =
OnceCell::const_new();
const MYSQL_REPLICATION_NETWORK: &str = "inbuxa-test-mysql-replication";
/// The MySQL pair that replicates with GTIDs, and the one that replicates
/// by binary log position.
pub const MYSQL_GTID_PORTS: (u16, u16) = (3317, 3318);
pub const MYSQL_POS_PORTS: (u16, u16) = (3327, 3328);
const OPENLDAP_LDAPI_URL: &str = "ldapi://%2Fvar%2Frun%2Fslapd%2Fldapi/";
const OPENLDAP_ALLOW_UNAUTHENTICATED_BIND: &str = r#"set -e
ldapmodify -Y EXTERNAL -Q -H "$LDAPI" <<'EOF'
dn: cn=config
changetype: modify
replace: olcAllows
olcAllows: bind_anon_dn
EOF
test "$(ldapwhoami -x -H "$LDAPI" -D uid=john.doe,ou=users,dc=stalwart,dc=test -w '')" = anonymous
"#;
const POWERDNS_ZONE_INIT: &str = r#"set -e
for i in $(seq 1 60); do
pdnsutil list-all-zones >/dev/null 2>&1 && break
sleep 1
done
if pdnsutil list-zone stalwart.test >/dev/null 2>&1; then
exit 0
fi
pdnsutil create-zone stalwart.test ns1.stalwart.test
pdnsutil set-kind stalwart.test native
pdnsutil replace-rrset stalwart.test '' SOA 'ns1.stalwart.test. admin.stalwart.test. 2024010101 3600 900 604800 86400'
pdnsutil add-record stalwart.test 'ns1' A '127.0.0.1'
pdnsutil add-record stalwart.test '' A '127.0.0.1'
pdnsutil add-record stalwart.test '' MX '10 mail.stalwart.test.'
pdnsutil add-record stalwart.test 'mail' A '127.0.0.1'
pdnsutil import-tsig-key stalwart-update-key hmac-sha256 'c3RhbHdhcnQtdGVzdC10c2lnLXNlY3JldC1rZXkxMjM0NTY3ODkw'
pdnsutil activate-tsig-key stalwart.test stalwart-update-key primary
pdnsutil set-meta stalwart.test TSIG-ALLOW-DNSUPDATE stalwart-update-key
pdnsutil set-meta stalwart.test ALLOW-DNSUPDATE-FROM '0.0.0.0/0'
"#;
pub async fn ensure_foundationdb() {
let container = FOUNDATIONDB
.get_or_init(|| async {
GenericImage::new("foundationdb/foundationdb", "7.4.6")
.with_env_var("FDB_NETWORKING_MODE", "container")
.with_mapped_port(4500, 4500.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-foundationdb")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start FoundationDB container")
})
.await;
let start = Instant::now();
loop {
if fdbcli(container, "status minimal")
.await
.contains("The database is available")
{
return;
}
let created = fdbcli(container, "configure new single memory").await;
if created.contains("Database created") || created.contains("Already exists") {
continue;
}
if start.elapsed() > READY_TIMEOUT {
panic!("Timed out configuring FoundationDB: {created}");
}
tokio::time::sleep(Duration::from_secs(2)).await;
}
}
async fn fdbcli(container: &ContainerAsync<GenericImage>, command: &str) -> String {
let mut result = container
.exec(
ExecCommand::new(["fdbcli", "--exec", command, "--timeout", "5"])
.with_cmd_ready_condition(CmdWaitFor::exit()),
)
.await
.expect("Failed to exec fdbcli");
let stdout = result.stdout_to_vec().await.unwrap_or_default();
let stderr = result.stderr_to_vec().await.unwrap_or_default();
format!(
"{}{}",
String::from_utf8_lossy(&stdout),
String::from_utf8_lossy(&stderr)
)
}
pub async fn ensure_postgres() {
POSTGRES
.get_or_init(|| async {
GenericImage::new("postgres", "16-alpine")
.with_wait_for(WaitFor::message_on_stderr(
"database system is ready to accept connections",
))
.with_wait_for(WaitFor::message_on_stderr(
"database system is ready to accept connections",
))
.with_env_var("POSTGRES_USER", "stalwart")
.with_env_var("POSTGRES_PASSWORD", "stalwart")
.with_env_var("POSTGRES_DB", "stalwart")
.with_mapped_port(5432, 5432.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-postgres")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start PostgreSQL container")
})
.await;
wait_for_tcp(5432).await;
}
pub async fn ensure_mysql() {
MYSQL
.get_or_init(|| async {
GenericImage::new("mysql", "8.0")
.with_wait_for(WaitFor::message_on_stderr("port: 3306 MySQL"))
.with_env_var("MYSQL_ROOT_PASSWORD", "password")
.with_env_var("MYSQL_DATABASE", "stalwart")
.with_cmd(["--default-authentication-plugin=mysql_native_password"])
.with_mapped_port(3307, 3306.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-mysql")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start MySQL container")
})
.await;
wait_for_tcp(3307).await;
}
pub async fn ensure_mariadb() {
MARIADB
.get_or_init(|| async {
GenericImage::new("mariadb", "11.4")
.with_wait_for(WaitFor::message_on_stderr("port: 3306 mariadb.org"))
.with_env_var("MARIADB_ROOT_PASSWORD", "password")
.with_env_var("MARIADB_DATABASE", "stalwart")
.with_mapped_port(3308, 3306.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-mariadb")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start MariaDB container")
})
.await;
wait_for_tcp(3308).await;
}
pub async fn ensure_redis() {
REDIS
.get_or_init(|| async {
GenericImage::new("redis", "7-alpine")
.with_wait_for(WaitFor::message_on_stdout("Ready to accept connections"))
.with_cmd(["redis-server", "--save", "", "--appendonly", "no"])
.with_mapped_port(6379, 6379.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-redis")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start Redis container")
})
.await;
wait_for_tcp(6379).await;
}
pub async fn ensure_nats() {
NATS.get_or_init(|| async {
GenericImage::new("nats", "latest")
.with_wait_for(WaitFor::message_on_stderr("Server is ready"))
.with_cmd(["--addr", "0.0.0.0", "--port", "4222", "--http_port", "8222"])
.with_mapped_port(4222, 4222.tcp())
.with_mapped_port(8222, 8222.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-nats")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start NATS container")
})
.await;
wait_for_tcp(4222).await;
}
pub async fn ensure_minio() {
MINIO
.get_or_init(|| async {
GenericImage::new("minio/minio", "latest")
.with_env_var("MINIO_ROOT_USER", "minioadmin")
.with_env_var("MINIO_ROOT_PASSWORD", "minioadmin")
.with_cmd(["server", "/data", "--console-address", ":9001"])
.with_mapped_port(9000, 9000.tcp())
.with_mapped_port(9001, 9001.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-minio")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start MinIO container")
})
.await;
wait_for_http("http://localhost:9000/minio/health/live").await;
create_minio_bucket().await;
}
pub async fn ensure_opensearch() {
OPENSEARCH
.get_or_init(|| async {
GenericImage::new("opensearchproject/opensearch", "2")
.with_env_var("discovery.type", "single-node")
.with_env_var("DISABLE_SECURITY_PLUGIN", "true")
.with_env_var("OPENSEARCH_JAVA_OPTS", "-Xms1g -Xmx1g")
.with_env_var("DISABLE_INSTALL_DEMO_CONFIG", "true")
.with_mapped_port(9200, 9200.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-opensearch")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start OpenSearch container")
})
.await;
wait_for_http("http://localhost:9200").await;
}
pub async fn ensure_meilisearch() {
MEILISEARCH
.get_or_init(|| async {
GenericImage::new("getmeili/meilisearch", "latest")
.with_env_var("MEILI_ENV", "development")
.with_env_var("MEILI_NO_ANALYTICS", "true")
.with_env_var("MEILI_MASTER_KEY", "stalwart-master-key")
.with_mapped_port(7700, 7700.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-meilisearch")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start Meilisearch container")
})
.await;
wait_for_http("http://localhost:7700/health").await;
}
pub async fn ensure_keycloak() {
KEYCLOAK
.get_or_init(|| async {
GenericImage::new("quay.io/keycloak/keycloak", "latest")
.with_env_var("KC_BOOTSTRAP_ADMIN_USERNAME", "admin")
.with_env_var("KC_BOOTSTRAP_ADMIN_PASSWORD", "admin")
.with_env_var("KC_HTTP_PORT", "9080")
.with_env_var("KC_HEALTH_ENABLED", "true")
.with_cmd(["start-dev", "--import-realm"])
.with_copy_to(
"/opt/keycloak/data/import/stalwart-realm.json",
include_bytes!("../../docker/keycloak/stalwart-realm.json").to_vec(),
)
// inbuxa: a second provider, for per-domain directories (test 10)
.with_copy_to(
"/opt/keycloak/data/import/inbuxa-realm.json",
include_bytes!("../../docker/keycloak/inbuxa-realm.json").to_vec(),
)
.with_mapped_port(9080, 9080.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-keycloak")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start Keycloak container")
})
.await;
wait_for_http("http://localhost:9080/realms/stalwart/.well-known/openid-configuration").await;
wait_for_http("http://localhost:9080/realms/inbuxa/.well-known/openid-configuration").await;
}
pub async fn ensure_scim_tester() -> &'static ContainerAsync<GenericImage> {
SCIM_TESTER
.get_or_init(|| async {
let image = GenericBuildableImage::new("stalwart-test-scim-tester", "local")
.with_dockerfile_string(include_str!("../../docker/scim/Dockerfile"))
.build_image()
.await
.expect("Failed to build the SCIM tester image");
image
.with_host("host.docker.internal", Host::HostGateway)
// inbuxa: on the host's network, so a host firewall that
// drops the Docker bridge doesn't block the test server
.with_network("host")
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-scim-tester")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start the SCIM tester container")
})
.await
}
pub async fn scim_tester_exec(args: &[&str]) -> (String, String) {
let mut result = ensure_scim_tester()
.await
.exec(ExecCommand::new(args.iter().copied()).with_cmd_ready_condition(CmdWaitFor::exit()))
.await
.expect("Failed to exec the SCIM driver");
let stdout = result.stdout_to_vec().await.unwrap_or_default();
let stderr = result.stderr_to_vec().await.unwrap_or_default();
(
String::from_utf8_lossy(&stdout).into_owned(),
String::from_utf8_lossy(&stderr).into_owned(),
)
}
pub async fn ensure_acme() {
ensure_challtestsrv().await;
ensure_pebble().await;
}
async fn ensure_challtestsrv() {
CHALLTESTSRV
.get_or_init(|| async {
let image = GenericBuildableImage::new("stalwart-test-challtestsrv", "local")
.with_dockerfile_string(include_str!("../../docker/pebble/Dockerfile.challtestsrv"))
.build_image()
.await
.expect("Failed to build challtestsrv image");
image
.with_network(ACME_NETWORK)
.with_host("host.docker.internal", Host::HostGateway)
.with_mapped_port(8055, 8055.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-challtestsrv")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start challtestsrv container")
})
.await;
wait_for_tcp(8055).await;
}
async fn ensure_pebble() {
PEBBLE
.get_or_init(|| async {
GenericImage::new("ghcr.io/letsencrypt/pebble", "latest")
.with_env_var("PEBBLE_VA_NOSLEEP", "1")
.with_env_var("PEBBLE_WFE_NONCEREJECT", "0")
.with_env_var("PEBBLE_ALTERNATE_ROOTS", "2")
.with_cmd([
"-config",
"/test/config/pebble-config.json",
"-dnsserver",
"stalwart-test-challtestsrv:8053",
])
.with_copy_to(
"/test/config/pebble-config.json",
include_bytes!("../../docker/pebble/pebble-config.json").to_vec(),
)
.with_network(ACME_NETWORK)
.with_host("host.docker.internal", Host::HostGateway)
.with_mapped_port(14000, 14000.tcp())
.with_mapped_port(15000, 15000.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-pebble")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start Pebble container")
})
.await;
wait_for_tcp(14000).await;
}
pub async fn ensure_powerdns() {
let container = POWERDNS
.get_or_init(|| async {
GenericImage::new("powerdns/pdns-auth-49", "latest")
.with_wait_for(WaitFor::message_on_stderr("Creating backend connection"))
.with_env_var("PDNS_AUTH_API_KEY", "stalwart-api-key")
.with_copy_to(
"/etc/powerdns/pdns.d/stalwart.conf",
include_bytes!("../../docker/powerdns/pdns.conf").to_vec(),
)
.with_mapped_port(5300, 53.tcp())
.with_mapped_port(5300, 53.udp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-powerdns")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start PowerDNS container")
})
.await;
let mut result = container
.exec(
ExecCommand::new(["bash", "-c", POWERDNS_ZONE_INIT])
.with_cmd_ready_condition(CmdWaitFor::exit()),
)
.await
.expect("Failed to exec PowerDNS zone init");
if result.exit_code().await.ok().flatten() != Some(0) {
let stdout =
String::from_utf8_lossy(&result.stdout_to_vec().await.unwrap_or_default()).into_owned();
let stderr =
String::from_utf8_lossy(&result.stderr_to_vec().await.unwrap_or_default()).into_owned();
panic!("PowerDNS zone init failed:\n{stdout}\n{stderr}");
}
wait_for_tcp(5300).await;
}
pub async fn ensure_openldap() {
const BOOTSTRAP_DIR: &str = "/container/service/slapd/assets/config/bootstrap/ldif/custom";
let container = OPENLDAP
.get_or_init(|| async {
GenericImage::new("osixia/openldap", "1.5.0")
.with_wait_for(WaitFor::message_on_stderr("slapd starting"))
.with_env_var("LDAP_ORGANISATION", "Stalwart Test")
.with_env_var("LDAP_DOMAIN", "stalwart.test")
.with_env_var("LDAP_BASE_DN", "dc=stalwart,dc=test")
.with_env_var("LDAP_ADMIN_PASSWORD", "admin")
.with_env_var("LDAP_TLS", "false")
.with_copy_to(
format!("{BOOTSTRAP_DIR}/50-users.ldif"),
include_bytes!("../../docker/ldap/50-users.ldif").to_vec(),
)
.with_copy_to(
format!("{BOOTSTRAP_DIR}/60-groups.ldif"),
include_bytes!("../../docker/ldap/60-groups.ldif").to_vec(),
)
.with_mapped_port(389, 389.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name("stalwart-test-openldap")
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start OpenLDAP container")
})
.await;
wait_for_tcp(389).await;
let setup = format!("LDAPI={OPENLDAP_LDAPI_URL}\n{OPENLDAP_ALLOW_UNAUTHENTICATED_BIND}");
let mut result = container
.exec(
ExecCommand::new(["bash", "-c", setup.as_str()])
.with_cmd_ready_condition(CmdWaitFor::exit()),
)
.await
.expect("Failed to exec OpenLDAP unauthenticated bind setup");
if result.exit_code().await.ok().flatten() != Some(0) {
let stdout =
String::from_utf8_lossy(&result.stdout_to_vec().await.unwrap_or_default()).into_owned();
let stderr =
String::from_utf8_lossy(&result.stderr_to_vec().await.unwrap_or_default()).into_owned();
panic!("OpenLDAP unauthenticated bind setup failed:\n{stdout}\n{stderr}");
}
}
async fn create_minio_bucket() {
use s3::{Bucket, BucketConfiguration, Region, creds::Credentials};
let region = Region::Custom {
region: "eu-central-1".to_string(),
endpoint: "http://localhost:9000".to_string(),
};
let credentials = Credentials::new(Some("minioadmin"), Some("minioadmin"), None, None, None)
.expect("Failed to build MinIO credentials");
match Bucket::create_with_path_style(
"stalwart",
region,
credentials,
BucketConfiguration::default(),
)
.await
{
Ok(response) if response.success() => {}
Ok(_) => {}
Err(s3::error::S3Error::HttpFailWithBody(409, _)) => {}
Err(err) => panic!("Failed to create MinIO bucket: {err:?}"),
}
}
async fn wait_for_tcp(port: u16) {
let start = Instant::now();
loop {
if TcpStream::connect(("127.0.0.1", port)).await.is_ok() {
return;
}
if start.elapsed() > READY_TIMEOUT {
panic!("Timed out waiting for TCP port {port}");
}
tokio::time::sleep(Duration::from_millis(250)).await;
}
}
async fn wait_for_http(url: &str) {
let client = reqwest::Client::builder()
.danger_accept_invalid_certs(true)
.build()
.expect("Failed to build HTTP client");
let start = Instant::now();
loop {
if let Ok(response) = client.get(url).send().await
&& response.status().is_success()
{
return;
}
if start.elapsed() > READY_TIMEOUT {
panic!("Timed out waiting for {url}");
}
tokio::time::sleep(Duration::from_millis(500)).await;
}
}
/// inbuxa: a PostgreSQL primary on port 5442 with a hot-standby streaming
/// replica on 5443, for the read-replica tests (ST-5 to ST-15).
pub async fn ensure_postgres_replicated() {
PG_PRIMARY
.get_or_init(|| async {
GenericImage::new("postgres", "16-alpine")
.with_wait_for(WaitFor::message_on_stderr(
"database system is ready to accept connections",
))
.with_env_var("POSTGRES_USER", "stalwart")
.with_env_var("POSTGRES_PASSWORD", "stalwart")
.with_env_var("POSTGRES_DB", "stalwart")
.with_copy_to(
"/docker-entrypoint-initdb.d/replication.sh",
b"#!/bin/sh\necho 'host replication all all scram-sha-256' >> \"$PGDATA/pg_hba.conf\"\n"
.to_vec(),
)
.with_cmd([
"postgres",
"-c",
"wal_level=replica",
"-c",
"max_wal_senders=10",
"-c",
"hot_standby=on",
])
.with_network(PG_REPLICATION_NETWORK)
.with_mapped_port(PG_PRIMARY_PORT, 5432.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name(PG_PRIMARY_CONTAINER)
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start the PostgreSQL primary")
})
.await;
wait_for_tcp(PG_PRIMARY_PORT).await;
PG_REPLICA
.get_or_init(|| async {
GenericImage::new("postgres", "16-alpine")
.with_wait_for(WaitFor::message_on_stderr(
"database system is ready to accept read-only connections",
))
.with_env_var("PGPASSWORD", "stalwart")
.with_cmd([
"sh",
"-c",
concat!(
"if [ ! -s /var/lib/postgresql/replica/PG_VERSION ]; then ",
"until pg_basebackup -h inbuxa-test-pg-primary -U stalwart ",
"-D /var/lib/postgresql/replica -R -X stream; do sleep 1; done; fi; ",
"chown -R postgres:postgres /var/lib/postgresql/replica; ",
"chmod 700 /var/lib/postgresql/replica; ",
"exec su-exec postgres postgres -D /var/lib/postgresql/replica ",
"-c hot_standby=on"
),
])
.with_network(PG_REPLICATION_NETWORK)
.with_mapped_port(PG_REPLICA_PORT, 5432.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name(PG_REPLICA_CONTAINER)
.with_reuse(ReuseDirective::Always)
.start()
.await
.expect("Failed to start the PostgreSQL replica")
})
.await;
wait_for_tcp(PG_REPLICA_PORT).await;
}
/// inbuxa: runs SQL on a test container, through `psql`.
pub fn psql(container: &str, sql: &str) -> String {
let output = std::process::Command::new("docker")
.args([
"exec", container, "psql", "-U", "stalwart", "-d", "stalwart", "-tAc", sql,
])
.output()
.expect("docker exec");
String::from_utf8_lossy(&output.stdout).trim().to_string()
}
/// inbuxa: stops or starts a test container.
pub fn docker(action: &str, container: &str) {
let status = std::process::Command::new("docker")
.args([action, container])
.status()
.expect("docker");
assert!(status.success(), "docker {action} {container}");
}
fn mysql_container(
name: &str,
port: u16,
gtid: bool,
server_id: &str,
) -> testcontainers::ContainerRequest<GenericImage> {
let mut cmd = vec![
format!("--server-id={server_id}"),
"--log-bin=binlog".to_string(),
"--binlog-format=ROW".to_string(),
"--default-authentication-plugin=mysql_native_password".to_string(),
];
if gtid {
cmd.push("--gtid-mode=ON".to_string());
cmd.push("--enforce-gtid-consistency=ON".to_string());
}
let _ = name;
GenericImage::new("mysql", "8.0")
.with_wait_for(WaitFor::message_on_stderr("port: 3306 MySQL"))
.with_env_var("MYSQL_ROOT_PASSWORD", "password")
.with_env_var("MYSQL_DATABASE", "stalwart")
.with_cmd(cmd)
.with_network(MYSQL_REPLICATION_NETWORK)
.with_mapped_port(port, 3306.tcp())
.with_startup_timeout(READY_TIMEOUT)
.with_container_name(name)
.with_reuse(ReuseDirective::Always)
}
/// inbuxa: runs SQL on a MySQL test container, through the `mysql` client.
pub fn mysql_query(container: &str, sql: &str) -> String {
let output = std::process::Command::new("docker")
.args([
"exec",
container,
"mysql",
"-uroot",
"-ppassword",
"-N",
"-B",
"-e",
sql,
])
.output()
.expect("docker exec");
if !output.status.success() {
panic!(
"{sql}: {}",
String::from_utf8_lossy(&output.stderr).trim().to_string()
);
}
String::from_utf8_lossy(&output.stdout).trim().to_string()
}
/// inbuxa: a MySQL source and replica, replicating with GTIDs or by binary
/// log position (ST-10, ST-15).
pub async fn ensure_mysql_replicated(gtid: bool) {
let (cell, ports, names) = if gtid {
(
&MYSQL_GTID,
MYSQL_GTID_PORTS,
(
"inbuxa-test-mysql-gtid-primary",
"inbuxa-test-mysql-gtid-replica",
),
)
} else {
(
&MYSQL_POS,
MYSQL_POS_PORTS,
(
"inbuxa-test-mysql-pos-primary",
"inbuxa-test-mysql-pos-replica",
),
)
};
cell.get_or_init(|| async {
let primary = mysql_container(names.0, ports.0, gtid, "1")
.start()
.await
.expect("Failed to start the MySQL source");
let replica = mysql_container(names.1, ports.1, gtid, "2")
.start()
.await
.expect("Failed to start the MySQL replica");
wait_for_tcp(ports.0).await;
wait_for_tcp(ports.1).await;
// A replication account, and one without REPLICATION CLIENT (test 18)
mysql_query(
names.0,
concat!(
"CREATE USER IF NOT EXISTS 'repl'@'%' IDENTIFIED WITH mysql_native_password BY 'repl';",
"GRANT REPLICATION SLAVE ON *.* TO 'repl'@'%';",
"CREATE USER IF NOT EXISTS 'plain'@'%' IDENTIFIED WITH mysql_native_password BY 'plain';",
"GRANT ALL ON stalwart.* TO 'plain'@'%'; FLUSH PRIVILEGES;"
),
);
if mysql_query(names.1, "SELECT SERVICE_STATE FROM performance_schema.replication_applier_status")
.is_empty()
{
let source = if gtid {
format!(
"CHANGE REPLICATION SOURCE TO SOURCE_HOST='{}', SOURCE_PORT=3306, \
SOURCE_USER='repl', SOURCE_PASSWORD='repl', SOURCE_AUTO_POSITION=1",
names.0
)
} else {
let status = mysql_query(names.0, "SHOW MASTER STATUS");
let mut fields = status.split('\t');
let file = fields.next().unwrap_or_default().to_string();
let position = fields.next().unwrap_or("4").to_string();
format!(
"CHANGE REPLICATION SOURCE TO SOURCE_HOST='{}', SOURCE_PORT=3306, \
SOURCE_USER='repl', SOURCE_PASSWORD='repl', SOURCE_LOG_FILE='{file}', \
SOURCE_LOG_POS={position}",
names.0
)
};
mysql_query(names.1, &source);
mysql_query(names.1, "START REPLICA");
mysql_query(names.1, "SET GLOBAL super_read_only=ON");
}
(primary, replica)
})
.await;
wait_for_tcp(ports.0).await;
wait_for_tcp(ports.1).await;
}
Reference in New Issue View Git Blame Copy Permalink