filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Registry testing - all tests passing

Browse Source
This commit is contained in:
Maurus Decimus
2026-03-27 16:49:26 +01:00
parent 72bc8c05ab
commit e451f037c8
26 changed files with 1174 additions and 1239 deletions
Download Patch File Download Diff File Expand all files Collapse all files

290
tests/src/cluster/broadcast.rs
View File

@@ -4,80 +4,228 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use super::ClusterTest;
use crate::imap::idle;
use groupware::cache::GroupwareCache;
use std::net::IpAddr;
use types::collection::SyncCollection;
use crate::{
imap::idle,
utils::{
imap::{ImapConnection, Type},
server::TestServerBuilder,
},
};
use imap_proto::ResponseType;
use registry::{
schema::{
enums::NetworkListenerProtocol,
prelude::{ObjectType, Property, SocketAddr},
structs::{
ClusterListenerGroup, ClusterListenerGroupProperties, ClusterRole, ClusterTaskGroup,
Coordinator, Expression, Http, NatsCoordinator, NetworkListener, RedisStore,
},
},
types::map::Map,
};
use serde_json::json;
use std::str::FromStr;
use store::registry::RegistryQuery;
use types::id::Id;
pub async fn test(cluster: &ClusterTest) {
println!("Running cluster broadcast tests...");
pub const NUM_NODES: usize = 3;
// Run IMAP idle tests across nodes
let server1 = cluster.server(1);
let server2 = cluster.server(2);
let mut node1_client = cluster.imap_client("john", 1).await;
let mut node2_client = cluster.imap_client("john", 2).await;
idle::test(&mut node1_client, &mut node2_client, true).await;
#[test]
fn cluster_tests() {
tokio::runtime::Builder::new_multi_thread()
.thread_stack_size(8 * 1024 * 1024) // 8MB stack
.enable_all()
.build()
.unwrap()
.block_on(async {
println!("Running cluster broadcast tests...");
let mut servers = Vec::with_capacity(NUM_NODES);
// Test event broadcast
let test_ip: IpAddr = "8.8.8.8".parse().unwrap();
assert!(!server1.is_ip_blocked(&test_ip));
assert!(!server2.is_ip_blocked(&test_ip));
server1.block_ip(test_ip).await.unwrap();
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
assert!(server1.is_ip_blocked(&test_ip));
assert!(server2.is_ip_blocked(&test_ip));
let coordinator_id = std::env::var("COORDINATOR").expect(concat!(
"Missing coordinator type. Try running `STORE=<store_type> ",
"COORDINATOR=<coordinator_type> cargo test`"
));
let coordinator = match coordinator_id.as_str() {
"Nats" => Coordinator::Nats(NatsCoordinator {
addresses: Map::new(vec!["127.0.0.1:4222".to_string()]),
use_tls: false,
..Default::default()
}),
"Redis" => Coordinator::Redis(RedisStore {
url: "redis://127.0.0.1".to_string(),
..Default::default()
}),
_ => panic!("Unsupported coordinator type: {}", coordinator_id),
};
// Change John's password and expect it to propagate
let account_id = cluster.account_id("john");
assert!(server1.inner.cache.access_tokens.get(&account_id).is_some());
assert!(server2.inner.cache.access_tokens.get(&account_id).is_some());
let changes = server1
.core
.storage
.data
.update_principal(
UpdatePrincipal::by_id(account_id).with_updates(vec![PrincipalUpdate {
action: PrincipalAction::AddItem,
field: PrincipalField::Secrets,
value: PrincipalValue::String("hello".into()),
}]),
)
.await
.unwrap();
server1.invalidate_principal_caches(changes).await;
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
assert!(server1.inner.cache.access_tokens.get(&account_id).is_none());
assert!(server2.inner.cache.access_tokens.get(&account_id).is_none());
// Create initial server
let test = TestServerBuilder::new("cluster_test_0")
.await
.with_object(Http {
base_url: Expression {
else_: "'https://127.0.0.1:' + local_port".to_string(),
..Default::default()
},
..Default::default()
})
.await
.with_object(coordinator)
.await
.with_listener(NetworkListenerProtocol::Http, "http_0", 11000, true)
.await
.with_imap_listener(12000)
.await
.with_listener(NetworkListenerProtocol::Lmtp, "lmtp_0", 11200, false)
.await
.build()
.await;
let admin = test.account("admin");
admin.mta_no_auth().await;
let account = admin
.create_user_account(
"jdoe@example.com",
"this is john's secret",
"John's account",
&[],
vec![],
)
.await;
admin.reload_settings().await;
// Rename John to Juan and expect DAV caches to be invalidated
let access_token = server1.get_access_token(account_id).await.unwrap();
server1
.fetch_dav_resources(&access_token, account_id, SyncCollection::Calendar)
.await
.unwrap();
server2
.fetch_dav_resources(&access_token, account_id, SyncCollection::Calendar)
.await
.unwrap();
assert!(server1.inner.cache.events.get(&account_id).is_some());
assert!(server2.inner.cache.events.get(&account_id).is_some());
let changes = server1
.core
.storage
.data
.update_principal(
UpdatePrincipal::by_id(account_id).with_updates(vec![PrincipalUpdate {
action: PrincipalAction::Set,
field: PrincipalField::Name,
value: PrincipalValue::String("juan".into()),
}]),
)
.await
.unwrap();
server1.invalidate_principal_caches(changes).await;
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
assert!(server1.inner.cache.events.get(&account_id).is_none());
assert!(server2.inner.cache.events.get(&account_id).is_none());
// Create listeners
let mut listeners = vec![
test.server
.registry()
.query::<Vec<Id>>(RegistryQuery::new(ObjectType::NetworkListener))
.await
.unwrap(),
];
for node_id in 1..NUM_NODES {
let http_listener_id = admin
.registry_create_object(NetworkListener {
name: format!("http_{}", node_id),
bind: Map::new(vec![
SocketAddr::from_str(&format!("127.0.0.1:1100{node_id}")).unwrap(),
]),
protocol: NetworkListenerProtocol::Http,
tls_implicit: true,
use_tls: true,
..Default::default()
})
.await;
let imap_listener_id = admin
.registry_create_object(NetworkListener {
name: format!("imap_{}", node_id),
bind: Map::new(vec![
SocketAddr::from_str(&format!("127.0.0.1:1200{node_id}")).unwrap(),
]),
protocol: NetworkListenerProtocol::Imap,
tls_implicit: false,
use_tls: true,
..Default::default()
})
.await;
listeners.push(vec![http_listener_id, imap_listener_id]);
}
// Create node roles
for (role_id, listener_ids) in listeners.into_iter().enumerate() {
admin
.registry_create_object(ClusterRole {
name: format!("role_{role_id}"),
listeners: ClusterListenerGroup::EnableSome(
ClusterListenerGroupProperties {
listener_ids: Map::new(listener_ids),
},
),
tasks: ClusterTaskGroup::EnableAll,
description: None,
})
.await;
}
servers.push(test);
// Build additional servers
for node_id in 1..NUM_NODES {
let test = TestServerBuilder::new_with_role(
&format!("cluster_test_{node_id}"),
format!("mail-{node_id}.example.com"),
Some(format!("role_{node_id}")),
false,
)
.await
.build_with_opts(false)
.await;
// Verify that the server was assigned the correct node id
assert_eq!(test.server.registry().node_id(), node_id as u16);
servers.push(test);
}
// Verify cross-cluster cache invalidations
let admin = servers[0].account("admin");
let server1 = &servers[1].server;
let server2 = &servers[2].server;
let account_id = account.id().document_id();
assert_eq!(
server1
.account(account_id)
.await
.unwrap()
.description
.as_deref(),
Some("John's account")
);
assert_eq!(
server2
.account(account_id)
.await
.unwrap()
.description
.as_deref(),
Some("John's account")
);
admin
.registry_update_object(
ObjectType::Account,
account.id(),
json!({
Property::Description: "John Doe"
}),
)
.await;
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
assert_eq!(
server1
.account(account_id)
.await
.unwrap()
.description
.as_deref(),
Some("John Doe")
);
assert_eq!(
server2
.account(account_id)
.await
.unwrap()
.description
.as_deref(),
Some("John Doe")
);
// Run IMAP idle tests across nodes
let mut node1_client =
imap_client("jdoe@example.com", "this is john's secret", 1).await;
let mut node2_client =
imap_client("jdoe@example.com", "this is john's secret", 2).await;
idle::test(&mut node1_client, &mut node2_client, true).await;
});
}
async fn imap_client(login: &str, secret: &str, node_id: u32) -> ImapConnection {
let mut conn = ImapConnection::connect_to(b"A1 ", format!("127.0.0.1:1200{node_id}")).await;
conn.assert_read(Type::Untagged, ResponseType::Ok).await;
conn.authenticate(login, secret).await;
conn
}

359
tests/src/cluster/mod.rs
View File

@@ -4,364 +4,5 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{
AssertConfig, TEST_USERS, add_test_certs,
directory::internal::TestInternalDirectory,
imap::{ImapConnection, Type},
jmap::server::enterprise::EnterpriseCore,
store::cleanup::store_destroy,
};
use ahash::AHashMap;
use common::{
Caches, Core, Data, Inner, Server,
config::{
server::{Listeners, ServerProtocol},
telemetry::Telemetry,
},
};
use http::HttpSessionManager;
use imap::core::ImapSessionManager;
use imap_proto::ResponseType;
use jmap_client::client::{Client, Credentials};
use managesieve::core::ManageSieveSessionManager;
use pop3::Pop3SessionManager;
use services::{SpawnServices, broadcast::subscriber::spawn_broadcast_subscriber};
use smtp::{SpawnQueueManager, core::SmtpSessionManager};
use std::{path::PathBuf, sync::Arc, time::Duration};
use tokio::sync::watch;
pub mod broadcast;
pub mod stress;
pub const NUM_NODES: usize = 3;
#[tokio::test(flavor = "multi_thread")]
pub async fn cluster_tests() {
let params = init_cluster_tests(true).await;
//stress::test(params.server.clone(), params.client).await;
broadcast::test(&params).await;
}
#[allow(dead_code)]
pub struct ClusterTest {
servers: Vec<Server>,
account_ids: AHashMap<String, u32>,
shutdown_txs: Vec<watch::Sender<bool>>,
}
async fn init_cluster_tests(delete_if_exists: bool) -> ClusterTest {
// Load and parse config
let store_id = std::env::var("STORE").expect(
"Missing store type. Try running `STORE=<store_type> PUBSUB=<pubsub_type> cargo test`",
);
let pubsub_id = std::env::var("PUBSUB").expect(
"Missing store type. Try running `STORE=<store_type> PUBSUB=<pubsub_type> cargo test`",
);
let mut pubsub_config = match pubsub_id.as_str() {
"nats" => Config::new(SERVER_NATS).unwrap(),
"redis" => Config::new(SERVER_REDIS).unwrap(),
_ => panic!("Unsupported pubsub type: {}", pubsub_id),
};
// Build configs
let mut configs = Vec::with_capacity(NUM_NODES);
for node_id in 0..NUM_NODES {
let mut config = Config::new(
add_test_certs(SERVER)
.replace("{STORE}", &store_id)
.replace("{PUBSUB}", &pubsub_id)
.replace("{NODE_ID}", &node_id.to_string())
.replace(
"{LEVEL}",
&std::env::var("LOG").unwrap_or_else(|_| "disable".to_string()),
),
)
.unwrap();
config.resolve_all_macros().await;
configs.push(config);
}
// Build stores
let stores = Stores::parse_all(configs.first_mut().unwrap(), false).await;
// Build servers
let mut servers = Vec::with_capacity(NUM_NODES);
let mut shutdown_txs = Vec::with_capacity(NUM_NODES);
for config in configs {
let mut stores = stores.clone();
stores.pubsub_stores = Stores::parse(&mut pubsub_config).await.pubsub_stores;
let (server, shutdown_tx) = build_server(config, stores).await;
servers.push(server);
shutdown_txs.push(shutdown_tx);
}
let store = servers.first().unwrap().store().clone();
if delete_if_exists {
store_destroy(&store).await;
}
// Create test users
let mut account_ids = AHashMap::new();
for (account, secret, name, email) in TEST_USERS {
let account_id = store
.create_test_user(account, secret, name, &[email])
.await;
account_ids.insert(account.to_string(), account_id);
}
ClusterTest {
servers,
shutdown_txs,
account_ids,
}
}
impl ClusterTest {
pub async fn jmap_client(&self, login: &str, node_id: u32) -> Client {
Client::new()
.credentials(Credentials::basic(login, find_account_secret(login)))
.timeout(Duration::from_secs(3600))
.accept_invalid_certs(true)
.connect(&format!("https://127.0.0.1:1800{node_id}"))
.await
.unwrap()
}
pub async fn imap_client(&self, login: &str, node_id: u32) -> ImapConnection {
let mut conn = ImapConnection::connect_to(b"A1 ", format!("127.0.0.1:1900{node_id}")).await;
conn.assert_read(Type::Untagged, ResponseType::Ok).await;
conn.authenticate(login, find_account_secret(login)).await;
conn
}
pub fn server(&self, node_id: usize) -> &Server {
self.servers
.get(node_id)
.unwrap_or_else(|| panic!("No server found for node ID: {}", node_id))
}
pub fn account_id(&self, login: &str) -> u32 {
self.account_ids
.get(login)
.cloned()
.unwrap_or_else(|| panic!("No account ID found for login: {}", login))
}
}
fn find_account_secret(login: &str) -> &str {
TEST_USERS
.iter()
.find(|(account, _, _, _)| account == &login)
.map(|(_, secret, _, _)| secret)
.unwrap_or_else(|| panic!("No account found for login: {}", login))
}
async fn build_server(mut config: Config, stores: Stores) -> (Server, watch::Sender<bool>) {
// Parse servers
let mut servers = Listeners::parse(&mut config);
// Bind ports and drop privileges
servers.bind_and_drop_priv(&mut config);
// Parse core
let config_manager = ConfigManager {
cfg_local: Default::default(),
cfg_local_path: PathBuf::new(),
cfg_local_patterns: Patterns::parse(&mut config).into(),
cfg_store: config
.value("storage.data")
.and_then(|id| stores.stores.get(id))
.cloned()
.unwrap_or_default(),
};
let tracers = Telemetry::parse(&mut config, &stores);
let core = Core::parse(&mut config, stores, config_manager)
.await
.enable_enterprise();
let data = Data::parse(&mut config);
let cache = Caches::parse(&mut config);
let (ipc, mut ipc_rxs) = build_ipc(true);
let inner = Arc::new(Inner {
shared_core: core.into_shared(),
data,
ipc,
cache,
});
// Parse acceptors
servers.parse_tcp_acceptors(&mut config, inner.clone());
// Enable tracing
tracers.enable(true);
// Start services
config.assert_no_errors();
ipc_rxs.spawn_queue_manager(inner.clone());
ipc_rxs.spawn_services(inner.clone());
// Spawn servers
let (shutdown_tx, shutdown_rx) = servers.spawn(|server, acceptor, shutdown_rx| {
match &server.protocol {
ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn(
SmtpSessionManager::new(inner.clone()),
inner.clone(),
acceptor,
shutdown_rx,
),
ServerProtocol::Http => server.spawn(
HttpSessionManager::new(inner.clone()),
inner.clone(),
acceptor,
shutdown_rx,
),
ServerProtocol::Imap => server.spawn(
ImapSessionManager::new(inner.clone()),
inner.clone(),
acceptor,
shutdown_rx,
),
ServerProtocol::Pop3 => server.spawn(
Pop3SessionManager::new(inner.clone()),
inner.clone(),
acceptor,
shutdown_rx,
),
ServerProtocol::ManageSieve => server.spawn(
ManageSieveSessionManager::new(inner.clone()),
inner.clone(),
acceptor,
shutdown_rx,
),
};
});
// Start broadcast subscriber
spawn_broadcast_subscriber(inner.clone(), shutdown_rx);
(inner.build_server(), shutdown_tx)
}
const SERVER: &str = r#"
[server]
hostname = "'server{NODE_ID}.example.org'"
[http]
url = "'https://127.0.0.1:800{NODE_ID}'"
[cluster]
node-id = {NODE_ID}
coordinator = "{PUBSUB}"
[server.listener.http]
bind = ["127.0.0.1:1800{NODE_ID}"]
protocol = "http"
max-connections = 81920
tls.implicit = true
[server.listener.imap]
bind = ["127.0.0.1:1900{NODE_ID}"]
protocol = "imap"
max-connections = 81920
[server.listener.lmtp]
bind = ['127.0.0.1:1700{NODE_ID}']
protocol = 'lmtp'
tls.implicit = false
[server.socket]
reuse-addr = true
[server.tls]
enable = true
implicit = false
certificate = "default"
[session.ehlo]
reject-non-fqdn = false
[session.rcpt]
relay = [ { if = "!is_empty(authenticated_as)", then = true },
{ else = false } ]
directory = "'{STORE}'"
[session.rcpt.errors]
total = 5
wait = "1ms"
[session.auth]
mechanisms = "[plain, login, oauthbearer]"
directory = "'{STORE}'"
[resolver]
type = "system"
[queue.strategy]
route = [ { if = "rcpt_domain == 'example.com'", then = "'local'" },
{ else = "'mx'" } ]
[store."foundationdb"]
type = "foundationdb"
[store."postgresql"]
type = "postgresql"
host = "localhost"
port = 5432
database = "stalwart"
user = "postgres"
password = "mysecretpassword"
[store."mysql"]
type = "mysql"
host = "localhost"
port = 3307
database = "stalwart"
user = "root"
password = "password"
[certificate.default]
cert = "%{file:{CERT}}%"
private-key = "%{file:{PK}}%"
[storage]
data = "{STORE}"
fts = "{STORE}"
blob = "{STORE}"
lookup = "{STORE}"
directory = "{STORE}"
[directory."{STORE}"]
type = "internal"
store = "{STORE}"
[imap.auth]
allow-plain-text = true
[oauth]
key = "parerga_und_paralipomena"
[spam-filter]
enable = false
[tracer.console]
type = "console"
level = "{LEVEL}"
multiline = false
ansi = true
disabled-events = ["network.*", "telemetry.webhook-error", "http.request-body",
"eval.result", "store.*", "dkim.*", "queue.*", "delivery.*",
"*.raw-input", "*.raw-output" ]
"#;
const SERVER_NATS: &str = r#"
[store."nats"]
type = "nats"
urls = "127.0.0.1:4444"
"#;
const SERVER_REDIS: &str = r#"
[store."redis"]
type = "redis"
urls = "redis://127.0.0.1"
redis-type = "single"
"#;

51
tests/src/cluster/stress.rs
View File

@@ -4,8 +4,7 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::jmap::{assert_is_empty, mail::mailbox::destroy_all_mailboxes_no_wait, wait_for_tasks};
use common::Server;
use crate::utils::server::{DestroyAllMailboxes, TestServer, TestServerBuilder};
use email::{
cache::{MessageCacheFetch, email::MessageCacheAccess},
message::metadata::MessageData,
@@ -16,6 +15,7 @@ use jmap_client::{
core::set::{SetErrorType, SetObject},
mailbox::{self, Mailbox, Role},
};
use registry::schema::prelude::ObjectType;
use std::{str::FromStr, sync::Arc, time::Duration};
use store::{
ValueKey,
@@ -28,22 +28,33 @@ use types::{collection::Collection, id::Id};
const TEST_USER_ID: u32 = 1;
const NUM_PASSES: usize = 1;
pub async fn test(server: Server, mut client: Client) {
println!("Running cluster concurrency stress tests...");
server
.core
.storage
.data
.get_or_create_principal_id("john", directory::Type::Individual)
#[tokio::test(flavor = "multi_thread")]
pub async fn stress_tests() {
println!("Running concurrency stress tests...");
let mut test = TestServerBuilder::new("stress_tests")
.await
.unwrap();
client.set_default_account_id(Id::from(TEST_USER_ID).to_string());
let client = Arc::new(client);
email_tests(server.clone(), client.clone()).await;
mailbox_tests(server.clone(), client.clone()).await;
.with_default_listeners()
.await
.build()
.await;
let admin = test.create_admin_account("admin@example.com").await;
admin
.registry_destroy_all(ObjectType::MtaConnectionStrategy)
.await;
admin
.registry_destroy_all(ObjectType::MtaInboundThrottle)
.await;
test.insert_account(admin);
email_tests(&test).await;
mailbox_tests(&test).await;
}
async fn email_tests(server: Server, client: Arc<Client>) {
async fn email_tests(test: &TestServer) {
let server = &test.server;
let client = Arc::new(test.account("admin@example.com").jmap_client().await);
for pass in 0..NUM_PASSES {
println!(
"----------------- EMAIL STRESS TEST {} -----------------",
@@ -265,12 +276,14 @@ async fn email_tests(server: Server, client: Arc<Client>) {
}
test.wait_for_tasks().await;
destroy_all_mailboxes_no_wait(&client).await;
assert_is_empty(&server).await;
client.destroy_all_mailboxes().await;
test.assert_is_empty().await;
}
}
async fn mailbox_tests(server: Server, client: Arc<Client>) {
async fn mailbox_tests(test: &TestServer) {
let client = Arc::new(test.account("admin@example.com").jmap_client().await);
let mailboxes = Arc::new(vec![
"test/test1/test2/test3".to_string(),
"test1/test2/test3".to_string(),
@@ -361,7 +374,7 @@ async fn mailbox_tests(server: Server, client: Arc<Client>) {
{
let _ = client.mailbox_destroy(&mailbox_id, true).await;
}
assert_is_empty(&server).await;
test.assert_is_empty().await;
}
async fn create_mailbox(client: &Client, mailbox: &str) -> Vec<String> {

197
tests/src/directory/integration.rs Normal file
View File

@@ -0,0 +1,197 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{
directory::ldap::ldap_test_directory,
utils::{server::TestServerBuilder, smtp::SmtpConnection},
};
use ahash::AHashMap;
use email::cache::MessageCacheFetch;
use registry::schema::structs::{Account, AccountSettings, Directory};
use types::id::Id;
pub async fn test() {
let test = TestServerBuilder::new("directory_integration_test")
.await
.with_default_listeners()
.await
.with_object(Directory::Ldap(ldap_test_directory()))
.await
.build()
.await;
let admin = test.account("admin");
admin.mta_no_auth().await;
admin.mta_disable_spam_filter().await;
admin.reload_settings().await;
// Test account creation by login
let account = crate::utils::account::Account::new(
"john.doe@example.org",
"this is John's LDAP password",
&[],
"",
Id::from(u32::MAX),
);
assert_eq!(
account
.registry_get::<AccountSettings>(Id::singleton())
.await
.description
.as_deref(),
Some("John Doe")
);
// Test account creation by rcpt
let mut lmtp = SmtpConnection::connect().await;
for rcpt in [
"corporate@example.org",
"jane.smith@example.org",
"john@example.org",
"bill@example.org",
"sales@example.org",
] {
lmtp.ingest(
"bill@remote.org",
&[rcpt],
&TEST_EMAIL.replace("$RCPT", rcpt),
)
.await;
}
// Fetch all accounts
let mut accounts = admin
.registry_get_all::<Account>()
.await
.into_iter()
.map(|(id, account)| {
(
match &account {
Account::User(user_account) => user_account.name.clone(),
Account::Group(group_account) => group_account.name.clone(),
},
(account, id),
)
})
.collect::<AHashMap<_, _>>();
assert_eq!(accounts.len(), 5, "Got: {accounts:#?}");
// Validate accounts
for (name, description, secret, groups, aliases) in [
(
"john.doe",
"John Doe",
"$app$8958830913002348890$",
&["sales"][..],
&["john"][..],
),
(
"jane.smith",
"Jane Smith",
"$app$4096614298472586996$",
&["sales", "corporate"][..],
&[][..],
),
(
"bill.foobar",
"Bill Foobar",
"",
&["corporate"][..],
&["bill"][..],
),
] {
let (account, id) = accounts
.remove(name)
.map(|(account, id)| (account.into_user().unwrap(), id))
.unwrap();
assert_eq!(account.description.as_deref(), Some(description));
if !secret.is_empty() {
assert_eq!(
test.server
.registry()
.object::<Account>(id)
.await
.unwrap()
.unwrap()
.into_user()
.unwrap()
.credentials
.values()
.next()
.and_then(|v| v.as_main_credential())
.map(|v| v.secret.as_str()),
Some(secret)
);
}
for group in groups {
let id = accounts.get(*group).unwrap().1;
assert!(
account
.member_group_ids
.iter()
.any(|group_id| group_id == &id),
"Account {name} is not a member of group {group}"
);
}
for alias in aliases {
assert!(
account
.aliases
.iter()
.any(|account_alias| account_alias.name == *alias),
"Account {name} does not have alias {alias}"
);
}
assert_eq!(
test.server
.get_cached_messages(id.document_id())
.await
.unwrap()
.emails
.index
.len(),
1
);
}
// Validate groups
for (name, description, aliases) in [
("sales", "sales", &[][..]),
("corporate", "corporate", &["everyone"][..]),
] {
let (account, id) = accounts
.remove(name)
.map(|(account, id)| (account.into_group().unwrap(), id))
.unwrap();
assert_eq!(account.description.as_deref(), Some(description));
for alias in aliases {
assert!(
account
.aliases
.iter()
.any(|account_alias| account_alias.name == *alias),
"Group {name} does not have alias {alias}"
);
}
assert_eq!(
test.server
.get_cached_messages(id.document_id())
.await
.unwrap()
.emails
.index
.len(),
1
);
}
}
const TEST_EMAIL: &str = r#"From: bill@remote.org
To: $RCPT
Subject: TPS Report for $RCPT
I'm going to need those TPS reports ASAP. So, if you could do that, that'd be great.
"#;

59
tests/src/directory/ldap.rs
View File

@@ -10,34 +10,8 @@ use registry::{
types::map::Map,
};
#[tokio::test]
async fn ldap_directory() {
let mut config = structs::LdapDirectory {
url: "ldap://localhost".into(),
use_tls: false,
attr_class: Map::new(vec!["objectClass".to_string()]),
attr_description: Map::new(vec!["cn".to_string()]),
attr_email: Map::new(vec!["mail".to_string()]),
attr_email_alias: Map::new(vec!["mailAlias".to_string()]),
attr_member_of: Map::new(vec!["memberOf".to_string()]),
attr_secret: Map::new(vec![]),
attr_secret_changed: Map::new(vec!["shadowLastChange".to_string()]),
base_dn: "dc=stalwart,dc=test".into(),
bind_dn: "cn=admin,dc=stalwart,dc=test".to_string().into(),
bind_secret: SecretKeyOptional::Value(SecretKeyValue {
secret: "admin".into(),
}),
filter_member_of: "(&(objectClass=groupOfNames)(member=?))".to_string().into(),
filter_login: "(&(objectClass=inetOrgPerson)(mail=?))".into(),
filter_mailbox: concat!(
"(|(&(objectClass=inetOrgPerson)(|(mail=?)(mailAlias=?)))",
"(&(objectClass=groupOfNames)(|(mail=?)(mailAlias=?))))"
)
.into(),
group_class: "groupOfNames".into(),
bind_authentication: true,
..Default::default()
};
pub async fn test() {
let mut config = ldap_test_directory();
// Test bind authentication
let ldap = LdapDirectory::open(config.clone()).await.unwrap();
@@ -156,3 +130,32 @@ async fn ldap_directory() {
Recipient::Invalid
);
}
pub fn ldap_test_directory() -> structs::LdapDirectory {
structs::LdapDirectory {
url: "ldap://localhost".into(),
use_tls: false,
attr_class: Map::new(vec!["objectClass".to_string()]),
attr_description: Map::new(vec!["cn".to_string()]),
attr_email: Map::new(vec!["mail".to_string()]),
attr_email_alias: Map::new(vec!["mailAlias".to_string()]),
attr_member_of: Map::new(vec!["memberOf".to_string()]),
attr_secret: Map::new(vec![]),
attr_secret_changed: Map::new(vec!["shadowLastChange".to_string()]),
base_dn: "dc=stalwart,dc=test".into(),
bind_dn: "cn=admin,dc=stalwart,dc=test".to_string().into(),
bind_secret: SecretKeyOptional::Value(SecretKeyValue {
secret: "admin".into(),
}),
filter_member_of: "(&(objectClass=groupOfNames)(member=?))".to_string().into(),
filter_login: "(&(objectClass=inetOrgPerson)(mail=?))".into(),
filter_mailbox: concat!(
"(|(&(objectClass=inetOrgPerson)(|(mail=?)(mailAlias=?)))",
"(&(objectClass=groupOfNames)(|(mail=?)(mailAlias=?))))"
)
.into(),
group_class: "groupOfNames".into(),
bind_authentication: true,
..Default::default()
}
}

15
tests/src/directory/mod.rs
View File

@@ -4,6 +4,19 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
pub mod integration;
pub mod ldap;
pub mod oidc;
//pub mod sql;
#[cfg(feature = "sqlite")]
pub mod sql;
pub mod synchronization;
#[tokio::test(flavor = "multi_thread")]
pub async fn directory_tests() {
ldap::test().await;
oidc::test().await;
#[cfg(feature = "sqlite")]
sql::test().await;
synchronization::test().await;
integration::test().await;
}

3
tests/src/directory/oidc.rs
View File

@@ -11,8 +11,7 @@
use directory::{Account, Credentials, Directory, backend::oidc::OpenIdDirectory};
use registry::{schema::structs, types::map::Map};
#[tokio::test]
async fn oidc_directory() {
pub async fn test() {
let config = structs::OidcDirectory {
issuer_url: "http://localhost:9080/realms/stalwart".to_string(),
claim_username: "preferred_username".to_string(),

698
tests/src/directory/sql.rs
View File

@@ -4,573 +4,141 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use mail_send::Credentials;
use directory::{Account, Credentials, Group, Recipient, backend::sql::SqlDirectory};
use registry::schema::structs::{self, SqlAuthStore};
use store::{Store, backend::sqlite::SqliteStore};
#[allow(unused_imports)]
use store::{InMemoryStore, Store};
pub async fn test() {
let sql_store = Store::SQLite(SqliteStore::open_memory().unwrap().into());
use crate::{
directory::{DirectoryTest, IntoTestPrincipal, TestPrincipal, map_account_id, map_account_ids},
store::cleanup::store_destroy,
};
use super::DirectoryStore;
#[tokio::test]
async fn sql_directory() {
// Enable logging
/*tracing::subscriber::set_global_default(
tracing_subscriber::FmtSubscriber::builder()
.with_max_level(tracing::Level::TRACE)
.finish(),
)
.unwrap();*/
// Obtain directory handle
for directory_id in ["sqlite", "postgresql", "mysql"] {
// Parse config
let mut config = DirectoryTest::new(directory_id.into()).await;
println!("Testing SQL directory {:?}", directory_id);
let handle = config.directories.directories.remove(directory_id).unwrap();
let store = DirectoryStore {
store: config.stores.stores.remove(directory_id).unwrap(),
};
let base_store = &store.store;
let core = config.server;
// Create tables
store_destroy(base_store).await;
store.create_test_directory().await;
// Create test users
store
.create_test_user("admin", "very_secret", "Administrator")
.await;
store.create_test_user("john", "12345", "John Doe").await;
store.create_test_user("jane", "abcde", "Jane Doe").await;
store
.create_test_user(
"bill",
"$2y$05$bvIG6Nmid91Mu9RcmmWZfO5HJIMCT8riNW0hEp8f6/FuA2/mHZFpe",
"Bill Foobar",
)
.await;
store.set_test_quota("bill", 500000).await;
// Create test groups
store.create_test_group("sales", "Sales Team").await;
store.create_test_group("support", "Support Team").await;
// Link users to groups
store.add_to_group("john", "sales").await;
store.add_to_group("jane", "sales").await;
store.add_to_group("jane", "support").await;
// Add email addresses
store
.link_test_address("john", "john@example.org", "primary")
.await;
store
.link_test_address("jane", "jane@example.org", "primary")
.await;
store
.link_test_address("bill", "bill@example.org", "primary")
.await;
// Add aliases and lists
store
.link_test_address("john", "john.doe@example.org", "alias")
.await;
store
.link_test_address("john", "jdoe@example.org", "alias")
.await;
store
.link_test_address("john", "info@example.org", "list")
.await;
store
.link_test_address("jane", "info@example.org", "list")
.await;
store
.link_test_address("bill", "info@example.org", "list")
.await;
// Add catch-all user
store
.create_test_user("robert", "abcde", "Robert Foobar")
.await;
store
.link_test_address("robert", "robert@catchall.org", "primary")
.await;
store
.link_test_address("robert", "@catchall.org", "alias")
.await;
// Test authentication
assert_eq!(
handle
.query(
QueryParams::credentials(&Credentials::Plain {
username: "john".into(),
secret: "12345".into()
})
.with_return_member_of(true)
)
.await
.unwrap()
.unwrap()
.into_test(),
TestPrincipal {
id: base_store.get_principal_id("john").await.unwrap().unwrap(),
name: "john".into(),
description: Some("John Doe".into()),
secrets: vec!["12345".into()],
typ: Type::Individual,
member_of: map_account_ids(base_store, vec!["sales"])
.await
.into_iter()
.map(|v| v.to_string())
.collect(),
emails: vec![
"john@example.org".into(),
"jdoe@example.org".into(),
"john.doe@example.org".into()
],
roles: vec![ROLE_USER.to_string()],
..Default::default()
}
);
assert_eq!(
handle
.query(
QueryParams::credentials(&Credentials::Plain {
username: "bill".into(),
secret: "password".into()
})
.with_return_member_of(true)
)
.await
.unwrap()
.unwrap()
.into_test(),
TestPrincipal {
id: base_store.get_principal_id("bill").await.unwrap().unwrap(),
name: "bill".into(),
description: Some("Bill Foobar".into()),
secrets: vec![
"$2y$05$bvIG6Nmid91Mu9RcmmWZfO5HJIMCT8riNW0hEp8f6/FuA2/mHZFpe".into()
],
typ: Type::Individual,
quota: 500000,
emails: vec!["bill@example.org".into(),],
roles: vec![ROLE_USER.to_string()],
..Default::default()
}
);
assert_eq!(
handle
.query(
QueryParams::credentials(&Credentials::Plain {
username: "admin".into(),
secret: "very_secret".into()
})
.with_return_member_of(true)
)
.await
.unwrap()
.unwrap()
.into_test(),
TestPrincipal {
id: base_store.get_principal_id("admin").await.unwrap().unwrap(),
name: "admin".into(),
description: Some("Administrator".into()),
secrets: vec!["very_secret".into()],
typ: Type::Individual,
roles: vec![ROLE_ADMIN.to_string()],
..Default::default()
}
);
assert!(
handle
.query(
QueryParams::credentials(&Credentials::Plain {
username: "bill".into(),
secret: "invalid".into()
})
.with_return_member_of(true)
)
.await
.unwrap()
.is_none()
);
// Get user by name
let mut p = handle
.query(QueryParams::name("jane").with_return_member_of(true))
// Create test directory
for query in [
concat!(
"CREATE TABLE accounts (name TEXT PRIMARY KEY, secret TEXT, description TEXT,",
" type TEXT NOT NULL, active BOOLEAN DEFAULT TRUE)"
),
concat!(
"CREATE TABLE group_members (name TEXT NOT NULL, member_of ",
"TEXT NOT NULL, PRIMARY KEY (name, member_of))"
),
concat!(
"CREATE TABLE emails (name TEXT NOT NULL, address TEXT NOT",
" NULL, PRIMARY KEY (name, address))"
),
concat!(
"INSERT INTO accounts (name, secret, description, type) ",
"VALUES ('john@example.org', 'john secret', 'John Doe', 'individual')"
),
concat!(
"INSERT INTO accounts (name, secret, description, type) ",
"VALUES ('jane@example.org', 'jane secret', 'Jane Doe', 'individual')"
),
concat!(
"INSERT INTO accounts (name, secret, description, type) ",
"VALUES ('sales@example.org', NULL, 'Sales Team', 'group')"
),
concat!(
"INSERT INTO group_members (name, member_of) VALUES ",
"('john@example.org', 'sales@example.org')"
),
concat!(
"INSERT INTO group_members (name, member_of) VALUES ",
"('jane@example.org', 'sales@example.org')"
),
concat!(
"INSERT INTO emails (name, address) VALUES ",
"('john@example.org', 'john.doe@example.org')"
),
] {
sql_store
.sql_query::<usize>(query, vec![])
.await
.unwrap()
.unwrap()
.into_test();
p.member_of.sort();
assert_eq!(
p,
TestPrincipal {
id: base_store.get_principal_id("jane").await.unwrap().unwrap(),
name: "jane".into(),
description: Some("Jane Doe".into()),
typ: Type::Individual,
secrets: vec!["abcde".into()],
member_of: map_account_ids(base_store, vec!["sales", "support"])
.await
.into_iter()
.map(|v| v.to_string())
.collect(),
emails: vec!["jane@example.org".into(),],
roles: vec![ROLE_USER.to_string()],
..Default::default()
}
);
// Get group by name
assert_eq!(
handle
.query(QueryParams::name("sales").with_return_member_of(true))
.await
.unwrap()
.unwrap()
.into_test(),
TestPrincipal {
id: base_store.get_principal_id("sales").await.unwrap().unwrap(),
name: "sales".into(),
description: Some("Sales Team".into()),
typ: Type::Group,
roles: vec![ROLE_USER.to_string()],
..Default::default()
}
);
// Ids by email
assert_eq!(
core.email_to_id(&handle, "jane@example.org", 0)
.await
.unwrap(),
Some(map_account_id(base_store, "jane").await)
);
assert_eq!(
core.email_to_id(&handle, "jane+alias@example.org", 0)
.await
.unwrap(),
Some(map_account_id(base_store, "jane").await)
);
assert_eq!(
core.email_to_id(&handle, "unknown@example.org", 0)
.await
.unwrap(),
None
);
assert_eq!(
core.email_to_id(&handle, "anything@catchall.org", 0)
.await
.unwrap(),
Some(map_account_id(base_store, "robert").await)
);
// Domain validation
assert!(handle.is_local_domain("example.org").await.unwrap());
assert!(!handle.is_local_domain("other.org").await.unwrap());
// RCPT TO
assert_eq!(
core.rcpt(&handle, "jane@example.org", 0).await.unwrap(),
RcptType::Mailbox
);
assert_eq!(
core.rcpt(&handle, "info@example.org", 0).await.unwrap(),
RcptType::Mailbox
);
assert_eq!(
core.rcpt(&handle, "jane+alias@example.org", 0)
.await
.unwrap(),
RcptType::Mailbox
);
assert_eq!(
core.rcpt(&handle, "info+alias@example.org", 0)
.await
.unwrap(),
RcptType::Mailbox
);
assert_eq!(
core.rcpt(&handle, "random_user@catchall.org", 0)
.await
.unwrap(),
RcptType::Mailbox
);
assert_eq!(
core.rcpt(&handle, "invalid@example.org", 0).await.unwrap(),
RcptType::Invalid
);
// VRFY
assert_eq!(
core.vrfy(&handle, "jane", 0).await.unwrap(),
vec!["jane@example.org".to_string()]
);
assert_eq!(
core.vrfy(&handle, "john", 0).await.unwrap(),
vec![
"john.doe@example.org".to_string(),
"john@example.org".to_string(),
]
);
assert_eq!(
core.vrfy(&handle, "jane+alias@example", 0).await.unwrap(),
vec!["jane@example.org".to_string()]
);
assert_eq!(
core.vrfy(&handle, "info", 0).await.unwrap(),
Vec::<String>::new()
);
assert_eq!(
core.vrfy(&handle, "invalid", 0).await.unwrap(),
Vec::<String>::new()
);
// EXPN (now handled by the internal store)
/*assert_eq!(
core.expn(&handle, "info@example.org", 0).await.unwrap(),
vec![
"bill@example.org".into(),
"jane@example.org".into(),
"john@example.org".into()
]
);
assert_eq!(
core.expn(&handle, "john@example.org", 0).await.unwrap(),
Vec::<String>::new()
);*/
}
}
impl DirectoryStore {
pub async fn create_test_directory(&self) {
// Create tables
for table in ["accounts", "group_members", "emails"] {
self.store
.sql_query::<usize>(&format!("DROP TABLE IF EXISTS {table}"), vec![])
.await
.unwrap();
}
for query in [
concat!(
"CREATE TABLE accounts (name TEXT PRIMARY KEY, secret TEXT, description TEXT,",
" type TEXT NOT NULL, quota INTEGER ",
"DEFAULT 0, active BOOLEAN DEFAULT TRUE)"
),
concat!(
"CREATE TABLE group_members (name TEXT NOT NULL, member_of ",
"TEXT NOT NULL, PRIMARY KEY (name, member_of))"
),
concat!(
"CREATE TABLE emails (name TEXT NOT NULL, address TEXT NOT",
" NULL, type TEXT, PRIMARY KEY (name, address))"
),
"INSERT INTO accounts (name, secret, type) VALUES ('admin', 'secret', 'admin')",
] {
let query = if self.is_mysql() {
query.replace("TEXT", "VARCHAR(255)")
} else {
query.into()
};
self.store
.sql_query::<usize>(&query, vec![])
.await
.unwrap_or_else(|_| panic!("failed for {query}"));
}
}
pub async fn create_test_user(&self, login: &str, secret: &str, name: &str) {
let account_type = if login == "admin" {
"admin"
} else {
"individual"
};
self.store
.sql_query::<usize>(
if self.is_postgresql() {
concat!(
"INSERT INTO accounts (name, secret, description, ",
"type, active) VALUES ($1, $2, $3, $4, true) ",
"ON CONFLICT (name) ",
"DO UPDATE SET secret = $2, description = $3, type = $4, active = true"
)
} else if self.is_mysql() {
concat!(
"INSERT INTO accounts (name, secret, description, ",
"type, active) VALUES (?, ?, ?, ?, true) ",
"ON DUPLICATE KEY UPDATE ",
"secret = VALUES(secret), description = VALUES(description), ",
"type = VALUES(type), active = true"
)
} else {
concat!(
"INSERT INTO accounts (name, secret, description, ",
"type, active) VALUES (?, ?, ?, ?, true) ",
"ON CONFLICT(name) DO UPDATE SET ",
"secret = excluded.secret, description = excluded.description, ",
"type = excluded.type, active = true"
)
},
vec![
login.into(),
secret.into(),
name.into(),
account_type.into(),
],
)
.await
.unwrap();
}
pub async fn create_test_user_with_email(&self, login: &str, secret: &str, name: &str) {
self.create_test_user(login, secret, name).await;
self.link_test_address(login, login, "primary").await;
}
pub async fn create_test_group(&self, login: &str, name: &str) {
self.store
.sql_query::<usize>(
if self.is_postgresql() {
concat!(
"INSERT INTO accounts (name, description, ",
"type, active) VALUES ($1, $2, $3, $4) ON CONFLICT (name) DO NOTHING"
)
} else if self.is_mysql() {
concat!(
"INSERT IGNORE INTO accounts (name, description, ",
"type, active) VALUES (?, ?, ?, ?)"
)
} else {
concat!(
"INSERT OR IGNORE INTO accounts (name, description, ",
"type, active) VALUES (?, ?, ?, ?)"
)
},
vec![login.into(), name.into(), "group".into(), true.into()],
)
.await
.unwrap();
}
pub async fn create_test_group_with_email(&self, login: &str, name: &str) {
self.create_test_group(login, name).await;
self.link_test_address(login, login, "primary").await;
}
pub async fn link_test_address(&self, login: &str, address: &str, typ: &str) {
self.store
.sql_query::<usize>(
if self.is_postgresql() {
"INSERT INTO emails (name, address, type) VALUES ($1, $2, $3) ON CONFLICT (name, address) DO NOTHING"
} else if self.is_mysql() {
"INSERT IGNORE INTO emails (name, address, type) VALUES (?, ?, ?)"
} else {
"INSERT OR IGNORE INTO emails (name, address, type) VALUES (?, ?, ?)"
},
vec![login.into(), address.into(), typ.into()],
)
.await
.unwrap();
}
pub async fn set_test_quota(&self, login: &str, quota: u32) {
self.store
.sql_query::<usize>(
if self.is_postgresql() {
"UPDATE accounts SET quota = $1 where name = $2"
} else {
"UPDATE accounts SET quota = ? where name = ?"
},
vec![quota.into(), login.into()],
)
.await
.unwrap();
}
pub async fn add_to_group(&self, login: &str, group: &str) {
self.store
.sql_query::<usize>(
if self.is_postgresql() {
"INSERT INTO group_members (name, member_of) VALUES ($1, $2)"
} else {
"INSERT INTO group_members (name, member_of) VALUES (?, ?)"
},
vec![login.into(), group.into()],
)
.await
.unwrap();
}
pub async fn remove_from_group(&self, login: &str, group: &str) {
self.store
.sql_query::<usize>(
if self.is_postgresql() {
"DELETE FROM group_members WHERE name = $1 AND member_of = $2"
} else {
"DELETE FROM group_members WHERE name = ? AND member_of = ?"
},
vec![login.into(), group.into()],
)
.await
.unwrap();
}
pub async fn remove_test_alias(&self, login: &str, alias: &str) {
self.store
.sql_query::<usize>(
if self.is_postgresql() {
"DELETE FROM emails WHERE name = $1 AND address = $2"
} else {
"DELETE FROM emails WHERE name = ? AND address = ?"
},
vec![login.into(), alias.into()],
)
.await
.unwrap();
}
fn is_mysql(&self) -> bool {
#[cfg(feature = "mysql")]
{
matches!(self.store, Store::MySQL(_))
}
#[cfg(not(feature = "mysql"))]
{
false
}
}
fn is_postgresql(&self) -> bool {
#[cfg(feature = "postgres")]
{
matches!(self.store, Store::PostgreSQL(_))
}
#[cfg(not(feature = "postgres"))]
{
false
}
}
#[allow(dead_code)]
fn is_sqlite(&self) -> bool {
#[cfg(feature = "sqlite")]
{
matches!(self.store, Store::SQLite(_))
}
#[cfg(not(feature = "sqlite"))]
{
false
}
.unwrap_or_else(|_| panic!("failed for {query}"));
}
let config = structs::SqlDirectory {
query_login: concat!(
"SELECT name, secret, description, type FROM accounts ",
"WHERE name = $1 AND active = true"
)
.into(),
query_recipient: concat!(
"SELECT name, secret, description, type FROM accounts ",
"WHERE name = $1 AND active = true"
)
.into(),
query_email_aliases: concat!("SELECT address FROM emails ", "WHERE name = $1")
.to_string()
.into(),
query_member_of: concat!("SELECT member_of FROM group_members ", "WHERE name = $1")
.to_string()
.into(),
column_class: "type".to_string().into(),
column_description: "description".to_string().into(),
column_email: "name".into(),
column_secret: "secret".into(),
store: SqlAuthStore::Default,
};
// Test authentication
let sql = SqlDirectory::open(config, &sql_store).await.unwrap();
assert_eq!(
sql.authenticate(&Credentials::Basic {
username: "john@example.org".to_string(),
secret: "john secret".to_string(),
mfa_token: None,
})
.await
.unwrap(),
Account {
email: "john@example.org".to_string(),
email_aliases: vec!["john.doe@example.org".to_string(),],
secret: Some("john secret".to_string()),
groups: vec!["sales@example.org".to_string()],
description: Some("John Doe".to_string()),
}
);
assert!(
sql.authenticate(&Credentials::Basic {
username: "john@example.org".to_string(),
secret: "wrong secret".to_string(),
mfa_token: None,
})
.await
.is_err()
);
// Test recipient lookup
assert_eq!(
sql.recipient("john@example.org").await.unwrap(),
Recipient::Account(Account {
email: "john@example.org".to_string(),
email_aliases: vec!["john.doe@example.org".to_string()],
secret: Some("john secret".to_string()),
groups: vec!["sales@example.org".to_string()],
description: Some("John Doe".to_string()),
})
);
assert_eq!(
sql.recipient("jane@example.org").await.unwrap(),
Recipient::Account(Account {
email: "jane@example.org".to_string(),
email_aliases: vec![],
secret: Some("jane secret".to_string()),
groups: vec!["sales@example.org".to_string()],
description: Some("Jane Doe".to_string()),
})
);
assert_eq!(
sql.recipient("sales@example.org").await.unwrap(),
Recipient::Group(Group {
email: "sales@example.org".to_string(),
email_aliases: vec![],
description: Some("Sales Team".to_string())
})
);
assert_eq!(
sql.recipient("unknown@example.org").await.unwrap(),
Recipient::Invalid
);
}

271
tests/src/directory/synchronization.rs Normal file
View File

@@ -0,0 +1,271 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::utils::server::TestServerBuilder;
use registry::schema::{
prelude::ObjectType,
structs::{Account, Domain, EmailAlias},
};
use types::id::Id;
pub async fn test() {
let test = TestServerBuilder::new("directory_synchronization_test")
.await
.with_default_listeners()
.await
.disable_services()
.build()
.await;
let admin = test.account("admin");
// Synchronizing an account with an unknown domain should fail
assert!(
test.server
.synchronize_account(directory::Account {
email: "john@unknown.org".to_string(),
email_aliases: vec![],
secret: "supersecret".to_string().into(),
groups: vec![],
description: "John Doe".to_string().into(),
})
.await
.is_err()
);
// Initial account synchronization
let mut account_in = directory::Account {
email: "john@example.org".to_string(),
email_aliases: vec![
"john.doe@example.org".to_string(),
"j.doe@example.org".to_string(),
],
secret: "supersecret".to_string().into(),
groups: vec![
"corporate@example.org".to_string(),
"sales@example.org".to_string(),
],
description: "John Doe".to_string().into(),
};
let result = test
.server
.synchronize_account(account_in.clone())
.await
.unwrap();
let account_id = Id::from(result.id);
let account_out = test
.server
.registry()
.object::<Account>(account_id)
.await
.unwrap()
.unwrap()
.into_user()
.unwrap();
let domain_id = account_out.domain_id;
assert_eq!(
admin.registry_get::<Domain>(domain_id).await.name,
"example.org"
);
assert_eq!(account_out.name, "john");
assert_eq!(account_out.description.as_deref(), Some("John Doe"));
assert_eq!(
account_out
.credentials
.values()
.next()
.and_then(|v| v.as_main_credential())
.map(|c| c.secret.as_str()),
Some("supersecret")
);
assert_eq!(account_out.aliases.len(), 2);
let aliases = account_out.aliases.iter().collect::<Vec<_>>();
assert_eq!(
aliases[0],
&EmailAlias {
description: None,
domain_id,
enabled: true,
name: "john.doe".to_string(),
}
);
assert_eq!(
aliases[1],
&EmailAlias {
description: None,
domain_id,
enabled: true,
name: "j.doe".to_string(),
}
);
assert_eq!(account_out.member_group_ids.len(), 2);
for (idx, group_id) in account_out.member_group_ids.iter().enumerate() {
let group = admin
.registry_get::<Account>(*group_id)
.await
.into_group()
.unwrap();
assert_eq!(group.name, if idx == 0 { "corporate" } else { "sales" });
assert_eq!(group.domain_id, domain_id);
}
assert_eq!(
test.server
.registry()
.count_object(ObjectType::Account)
.await
.unwrap(),
3
);
assert_eq!(
test.server
.registry()
.count_object(ObjectType::Domain)
.await
.unwrap(),
1
);
// No changes should not cause any updates
assert_eq!(
test.server
.synchronize_account(account_in.clone())
.await
.unwrap()
.id,
account_id.document_id()
);
assert_eq!(
test.server
.registry()
.object::<Account>(account_id)
.await
.unwrap()
.unwrap()
.into_user()
.unwrap(),
account_out
);
assert_eq!(
test.server
.registry()
.count_object(ObjectType::Account)
.await
.unwrap(),
3
);
// Make some changes and synchronize again
account_in.description = "Johnathan Doe".to_string().into();
account_in
.email_aliases
.push("johnny@example.org".to_string());
account_in.groups.pop();
account_in.groups.push("support@example.org".to_string());
account_in.secret = "evenmoresecret".to_string().into();
assert_eq!(
test.server
.synchronize_account(account_in.clone())
.await
.unwrap()
.id,
account_id.document_id()
);
let account_out = test
.server
.registry()
.object::<Account>(account_id)
.await
.unwrap()
.unwrap()
.into_user()
.unwrap();
assert_eq!(
account_out
.credentials
.values()
.next()
.and_then(|v| v.as_main_credential())
.map(|c| c.secret.as_str()),
Some("evenmoresecret")
);
assert_eq!(account_out.description.as_deref(), Some("Johnathan Doe"));
assert_eq!(account_out.aliases.len(), 3);
let aliases = account_out.aliases.iter().collect::<Vec<_>>();
assert_eq!(
aliases[2],
&EmailAlias {
description: None,
domain_id,
enabled: true,
name: "johnny".to_string(),
}
);
assert_eq!(account_out.member_group_ids.len(), 2);
let account_groups = account_out
.member_group_ids
.iter()
.copied()
.collect::<Vec<_>>();
for (idx, group_id) in account_groups.iter().enumerate() {
let group = admin
.registry_get::<Account>(*group_id)
.await
.into_group()
.unwrap();
assert_eq!(group.name, if idx == 0 { "corporate" } else { "support" });
assert_eq!(group.domain_id, domain_id);
}
assert_eq!(
test.server
.registry()
.count_object(ObjectType::Account)
.await
.unwrap(),
4
);
// Synchronize a group
assert_eq!(
test.server
.synchronize_group(directory::Group {
email: "corporate@example.org".to_string(),
email_aliases: vec!["everyone@example.org".to_string()],
description: "Corporate Group".to_string().into(),
})
.await
.unwrap(),
account_groups[0].document_id()
);
let group_out = test
.server
.registry()
.object::<Account>(account_groups[0])
.await
.unwrap()
.unwrap()
.into_group()
.unwrap();
assert_eq!(group_out.name, "corporate");
assert_eq!(group_out.description.as_deref(), Some("Corporate Group"));
assert_eq!(group_out.aliases.len(), 1);
let aliases = group_out.aliases.iter().collect::<Vec<_>>();
assert_eq!(
aliases[0],
&EmailAlias {
description: None,
domain_id,
enabled: true,
name: "everyone".to_string(),
}
);
assert_eq!(
test.server
.registry()
.count_object(ObjectType::Account)
.await
.unwrap(),
4
);
}

2
tests/src/imap/idle.rs
View File

@@ -163,7 +163,7 @@ pub async fn test(
.assert_contains("* 0 EXISTS");
// Test SMTP delivery notifications
let mut lmtp = SmtpConnection::connect_port(if is_cluster_test { 17000 } else { 11200 }).await;
let mut lmtp = SmtpConnection::connect().await;
lmtp.ingest(
"bill@example.com",
&["jdoe@example.com"],

2
tests/src/imap/mod.rs
View File

@@ -41,7 +41,7 @@ use serde_json::json;
use std::{path::PathBuf, time::Instant};
use utils::map::vec_map::VecMap;
#[tokio::test]
#[tokio::test(flavor = "multi_thread")]
pub async fn imap_tests() {
let mut test = TestServerBuilder::new("imap_tests")
.await

3
tests/src/lib.rs
View File

@@ -13,10 +13,8 @@ use jemallocator::Jemalloc;
#[global_allocator]
static GLOBAL: Jemalloc = Jemalloc;
/*
#[cfg(test)]
pub mod cluster;
*/
#[cfg(test)]
pub mod directory;
#[cfg(test)]
@@ -45,7 +43,6 @@ pub trait AssertConfig {
#[cfg(test)]
impl AssertConfig for Bootstrap {
fn assert_no_errors(self) -> Self {
let todo = "cluster tests";
if !self.errors.is_empty() {
panic!("Errors: {:#?}", self.errors);
}

5
tests/src/utils/cleanup.rs
View File

@@ -101,10 +101,9 @@ async fn store_destroy_sql_indexes(store: &Store) {
#[cfg(feature = "mysql")]
let table = index.mysql_table();
store
let _ = store
.sql_query::<usize>(&format!("TRUNCATE TABLE {table}"), vec![])
.await
.unwrap();
.await;
}
}
}

42
tests/src/utils/jmap.rs
View File

@@ -44,19 +44,35 @@ impl Account {
.into_iter()
.map(|id| Value::String(id.to_string()))
.collect::<Vec<Value>>();
self.jmap_method_calls(json!([[
format!("{object}/get"),
{
"accountId": account.id_string(),
"properties": properties
.into_iter()
.map(|p| Value::String(p.to_string()))
.collect::<Vec<_>>(),
"ids": if !ids.is_empty() { Some(ids) } else { None }
},
"0"
]]))
.await
if account.id().document_id() != u32::MAX {
self.jmap_method_calls(json!([[
format!("{object}/get"),
{
"accountId": account.id_string(),
"properties": properties
.into_iter()
.map(|p| Value::String(p.to_string()))
.collect::<Vec<_>>(),
"ids": if !ids.is_empty() { Some(ids) } else { None }
},
"0"
]]))
.await
} else {
self.jmap_method_calls(json!([[
format!("{object}/get"),
{
"properties": properties
.into_iter()
.map(|p| Value::String(p.to_string()))
.collect::<Vec<_>>(),
"ids": if !ids.is_empty() { Some(ids) } else { None }
},
"0"
]]))
.await
}
}
pub async fn jmap_query(

137
tests/src/utils/server.rs
View File

@@ -40,10 +40,10 @@ use pop3::Pop3SessionManager;
use registry::{
schema::{
enums::{DataStoreType, EventPolicy, NetworkListenerProtocol, TracingLevel},
prelude::{Object, SocketAddr},
prelude::{Object, ObjectType, SocketAddr},
structs::{
Certificate, Expression, Http, NetworkListener, PublicText, SecretKeyFile, SecretText,
Tracer, TracerStdout,
Authentication, Certificate, Expression, Http, NetworkListener, PublicText,
SecretKeyFile, SecretText, Tracer, TracerStdout,
},
},
types::{EnumImpl, map::Map},
@@ -61,7 +61,7 @@ use smtp::{
use std::{path::PathBuf, str::FromStr, sync::Arc};
use store::{
RegistryStore, Store, ValueKey,
registry::{bootstrap::Bootstrap, write::RegistryWrite},
registry::{RegistryQuery, bootstrap::Bootstrap, write::RegistryWrite},
write::{AlignedBytes, Archive},
};
use tokio::sync::{mpsc, watch};
@@ -92,6 +92,16 @@ pub struct TestServerBuilder {
impl TestServerBuilder {
pub async fn new(test_name: &str) -> Self {
let reset = std::env::var("NO_INSERT").is_err();
Self::new_with_role(test_name, "mail.example.org".to_string(), None, reset).await
}
pub async fn new_with_role(
test_name: &str,
hostname: String,
node_role: Option<String>,
reset: bool,
) -> Self {
let temp_dir = TempDir::new(test_name, reset);
let path = temp_dir.path.to_string_lossy().to_string();
let data_store = build_data_store(
@@ -114,7 +124,7 @@ impl TestServerBuilder {
Self {
bootstrap: Bootstrap::new(
RegistryStore::new(&path, store, "mail.example.org".to_string(), 1, None).await,
RegistryStore::new(&path, store, hostname, 1, node_role).await,
)
.await,
http_listener_port: 8899,
@@ -149,8 +159,7 @@ impl TestServerBuilder {
.await
}
pub async fn with_http_listener(mut self, port: u16) -> Self {
self.http_listener_port = port;
pub async fn with_http_listener(self, port: u16) -> Self {
self.with_listener(NetworkListenerProtocol::Http, "jmap", port, true)
.await
.with_object(Http {
@@ -169,6 +178,11 @@ impl TestServerBuilder {
.await
}
pub async fn with_imap_listener(self, port: u16) -> Self {
self.with_listener(NetworkListenerProtocol::Imap, "imap", port, false)
.await
}
pub async fn with_dummy_tls_cert(self) -> Self {
let mut cert_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
cert_path.push("resources");
@@ -191,12 +205,15 @@ impl TestServerBuilder {
}
pub async fn with_listener(
self,
mut self,
protocol: NetworkListenerProtocol,
name: &str,
port: u16,
tls_implicit: bool,
) -> Self {
if protocol == NetworkListenerProtocol::Http {
self.http_listener_port = port;
}
self.insert_object(NetworkListener {
bind: Map::new(vec![
SocketAddr::from_str(&format!("127.0.0.1:{port}")).unwrap(),
@@ -245,59 +262,91 @@ impl TestServerBuilder {
.unwrap_id(trc::location!())
}
pub async fn build(mut self) -> TestServer {
// Register stores from environment
self.bootstrap.registry.insert_stores_from_env().await;
pub async fn build(self) -> TestServer {
self.build_with_opts(true).await
}
// Enable logging if requested
let level = std::env::var("LOG")
.map(|log| TracingLevel::parse(&log).expect("Invalid log level"))
.ok();
pub async fn build_with_opts(mut self, init_store: bool) -> TestServer {
if init_store {
// Register stores from environment
self.bootstrap.registry.insert_stores_from_env().await;
self.insert_object(Tracer::Stdout(TracerStdout {
enable: level.is_some() || self.logging_enabled,
level: level.unwrap_or(TracingLevel::Info),
ansi: true,
multiline: false,
events: Map::new(
EventType::variants()
.iter()
.filter(|ev| {
let ev = ev.as_str();
ev.starts_with("network.")
|| ev.starts_with("http.connection-")
|| ev == "telemetry.webhook-error"
|| ev == "http.request-body"
|| ev == "http.request-url"
|| ev == "tls.no-certificates-available"
|| ev == "store.cache-hit"
})
.copied()
.collect(),
),
events_policy: EventPolicy::Exclude,
..Default::default()
}))
.await;
// Enable logging if requested
let level = std::env::var("LOG")
.map(|log| TracingLevel::parse(&log).expect("Invalid log level"))
.ok();
self.insert_object(Tracer::Stdout(TracerStdout {
enable: level.is_some() || self.logging_enabled,
level: level.unwrap_or(TracingLevel::Info),
ansi: true,
multiline: false,
events: Map::new(
EventType::variants()
.iter()
.filter(|ev| {
let ev = ev.as_str();
ev.starts_with("network.")
|| ev.starts_with("http.connection-")
|| ev == "telemetry.webhook-error"
|| ev == "http.request-body"
|| ev == "http.request-url"
|| ev == "tls.no-certificates-available"
|| ev == "store.cache-hit"
})
.copied()
.collect(),
),
events_policy: EventPolicy::Exclude,
..Default::default()
}))
.await;
}
// Start listeners
let mut servers = Listeners::parse(&mut self.bootstrap).await;
servers.bind_and_drop_priv(&mut self.bootstrap);
if init_store {
// Add safe defaults if missing
self.bootstrap.insert_safe_defaults().await;
// Add directory
if let Some(directory_id) = self
.bootstrap
.registry
.query::<Vec<Id>>(RegistryQuery::new(ObjectType::Directory))
.await
.unwrap()
.first()
{
let mut auth = self
.bootstrap
.registry
.object::<Authentication>(Id::singleton())
.await
.unwrap()
.unwrap();
auth.directory_id = Some(*directory_id);
self.bootstrap
.registry
.write(RegistryWrite::insert(&auth.into()))
.await
.unwrap();
}
}
// Parse storage
let storage = Storage::parse(&mut self.bootstrap).await;
// Reset search store
if self.reset {
if init_store && self.reset {
search_store_destroy(&storage.search).await;
}
// Parse telemetry
let telemetry = Telemetry::parse(&mut self.bootstrap, &storage).await;
// Add safe defaults if missing
self.bootstrap.insert_safe_defaults().await;
// Parse components
let core = Box::pin(Core::parse(&mut self.bootstrap, storage)).await;
let data = Data::parse(&mut self.bootstrap).await;

4
tests/src/utils/storage.rs
View File

@@ -73,9 +73,9 @@ pub fn build_data_store(typ: DataStoreType, path: &str) -> DataStore {
DataStoreType::PostgreSql => DataStore::PostgreSql(PostgreSqlStore {
host: "localhost".into(),
port: 5432,
auth_username: "postgres".to_string().into(),
auth_username: "stalwart".to_string().into(),
auth_secret: SecretKeyOptional::Value(SecretKeyValue {
secret: "mysecretpassword".into(),
secret: "stalwart".into(),
}),
database: "stalwart".into(),
use_tls: false,