From 3d7b9b0334c23f60f7699b56c7bdbdb5c18ef425 Mon Sep 17 00:00:00 2001 From: Mauro D Date: Sat, 10 Feb 2024 15:39:05 +0100 Subject: [PATCH] Test fixes --- crates/smtp/src/config/mod.rs | 5 - crates/smtp/src/config/queue.rs | 7 - crates/smtp/src/config/report.rs | 6 - crates/smtp/src/queue/mod.rs | 4 +- crates/store/src/backend/sqlite/main.rs | 2 +- crates/store/src/config.rs | 2 +- tests/src/directory/mod.rs | 18 +-- tests/src/smtp/inbound/data.rs | 10 +- tests/src/smtp/inbound/dmarc.rs | 33 +++-- tests/src/smtp/inbound/milter.rs | 31 ++-- tests/src/smtp/inbound/mod.rs | 185 ++++++++++++++++-------- tests/src/smtp/inbound/scripts.rs | 65 +++++---- tests/src/smtp/inbound/sign.rs | 14 +- tests/src/smtp/management/queue.rs | 15 +- tests/src/smtp/management/report.rs | 9 +- tests/src/smtp/mod.rs | 36 +++-- tests/src/smtp/outbound/dane.rs | 51 ++++--- tests/src/smtp/outbound/extensions.rs | 20 +-- tests/src/smtp/outbound/ip_lookup.rs | 6 +- tests/src/smtp/outbound/lmtp.rs | 22 ++- tests/src/smtp/outbound/mta_sts.rs | 36 +++-- tests/src/smtp/outbound/smtp.rs | 24 +-- tests/src/smtp/outbound/throttle.rs | 36 +++-- tests/src/smtp/outbound/tls.rs | 11 +- tests/src/smtp/queue/dsn.rs | 8 +- tests/src/smtp/queue/retry.rs | 24 +-- tests/src/smtp/queue/serialize.rs | 2 +- tests/src/smtp/reporting/analyze.rs | 3 +- tests/src/smtp/reporting/dmarc.rs | 71 ++++----- tests/src/smtp/reporting/scheduler.rs | 179 +++++++++-------------- tests/src/smtp/reporting/tls.rs | 10 +- tests/src/store/blob.rs | 4 +- tests/src/store/lookup.rs | 77 +++------- 33 files changed, 518 insertions(+), 508 deletions(-) diff --git a/crates/smtp/src/config/mod.rs b/crates/smtp/src/config/mod.rs index f2c19033..8ce29c2e 100644 --- a/crates/smtp/src/config/mod.rs +++ b/crates/smtp/src/config/mod.rs @@ -208,9 +208,6 @@ pub struct RelayHost { } pub struct QueueConfig { - pub path: PathBuf, - pub hash: IfBlock, - // Schedule pub retry: IfBlock, pub notify: IfBlock, @@ -240,8 +237,6 @@ pub struct QueueOutboundSourceIp { } pub struct ReportConfig { - pub path: PathBuf, - pub hash: IfBlock, pub submitter: IfBlock, pub analysis: ReportAnalysis, diff --git a/crates/smtp/src/config/queue.rs b/crates/smtp/src/config/queue.rs index 2bb839b0..7b0f2711 100644 --- a/crates/smtp/src/config/queue.rs +++ b/crates/smtp/src/config/queue.rs @@ -74,13 +74,6 @@ impl ConfigQueue for Config { let default_hostname = self.value_require("server.hostname")?; let config = QueueConfig { - path: self.property_require("queue.path")?, - hash: self - .parse_if_block("queue.hash", |name| { - map_expr_token::(name, sender_envelope_keys) - })? - .unwrap_or_else(|| IfBlock::new(32)), - retry: self .parse_if_block("queue.schedule.retry", |name| { map_expr_token::(name, host_envelope_keys) diff --git a/crates/smtp/src/config/report.rs b/crates/smtp/src/config/report.rs index c434a57c..cc858260 100644 --- a/crates/smtp/src/config/report.rs +++ b/crates/smtp/src/config/report.rs @@ -89,17 +89,11 @@ impl ConfigReport for Config { sender_envelope_keys, )?, tls: self.parse_aggregate_report("tls", default_hostname, rcpt_envelope_keys)?, - path: self.property_require("report.path")?, submitter: self .parse_if_block("report.submitter", |name| { map_expr_token::(name, &[V_RECIPIENT_DOMAIN]) })? .unwrap_or_else(|| IfBlock::new(default_hostname.to_string())), - hash: self - .parse_if_block("report.hash", |name| { - map_expr_token::(name, sender_envelope_keys) - })? - .unwrap_or_else(|| IfBlock::new(32)), analysis: ReportAnalysis { addresses, forward: self.property("report.analysis.forward")?.unwrap_or(false), diff --git a/crates/smtp/src/queue/mod.rs b/crates/smtp/src/queue/mod.rs index eb621dc9..6b1b03a4 100644 --- a/crates/smtp/src/queue/mod.rs +++ b/crates/smtp/src/queue/mod.rs @@ -65,7 +65,7 @@ pub struct Schedule { pub inner: T, } -#[derive(Debug, serde::Serialize, serde::Deserialize)] +#[derive(Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct Message { pub id: QueueId, pub created: u64, @@ -85,7 +85,7 @@ pub struct Message { pub quota_keys: Vec, } -#[derive(Debug, serde::Serialize, serde::Deserialize)] +#[derive(Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub enum QuotaKey { Size { key: Vec, id: u64 }, Count { key: Vec, id: u64 }, diff --git a/crates/store/src/backend/sqlite/main.rs b/crates/store/src/backend/sqlite/main.rs index f89ed166..dd95fb50 100644 --- a/crates/store/src/backend/sqlite/main.rs +++ b/crates/store/src/backend/sqlite/main.rs @@ -36,7 +36,7 @@ use crate::{ use super::{pool::SqliteConnectionManager, SqliteStore}; impl SqliteStore { - pub async fn open(config: &Config, prefix: impl AsKey) -> crate::Result { + pub fn open(config: &Config, prefix: impl AsKey) -> crate::Result { let prefix = prefix.as_key(); let db = Self { conn_pool: Pool::builder() diff --git a/crates/store/src/config.rs b/crates/store/src/config.rs index a727f5a4..5b1dd3a6 100644 --- a/crates/store/src/config.rs +++ b/crates/store/src/config.rs @@ -137,7 +137,7 @@ impl ConfigStore for Config { } #[cfg(feature = "sqlite")] "sqlite" => { - let db: Store = SqliteStore::open(self, prefix).await?.into(); + let db: Store = SqliteStore::open(self, prefix)?.into(); config.stores.insert(store_id.clone(), db.clone()); config .fts_stores diff --git a/tests/src/directory/mod.rs b/tests/src/directory/mod.rs index 43039c91..a69dfeb0 100644 --- a/tests/src/directory/mod.rs +++ b/tests/src/directory/mod.rs @@ -36,7 +36,7 @@ use rustls::ServerConfig; use rustls_pemfile::{certs, pkcs8_private_keys}; use rustls_pki_types::PrivateKeyDer; use std::{borrow::Cow, io::BufReader, path::PathBuf, sync::Arc}; -use store::{config::ConfigStore, LookupKey, LookupStore, LookupValue, Store, Stores}; +use store::{config::ConfigStore, LookupStore, Store, Stores}; use tokio_rustls::TlsAcceptor; use utils::config::Servers; @@ -599,15 +599,13 @@ async fn lookup_local() { ("suffix", "coco", false), ] { assert_eq!( - matches!( - lookups - .get(&format!("local/{lookup}")) - .unwrap() - .key_get::(LookupKey::Key(item.as_bytes().to_vec())) - .await - .unwrap(), - LookupValue::Value { .. } - ), + lookups + .get(&format!("local/{lookup}")) + .unwrap() + .key_get::(item.as_bytes().to_vec()) + .await + .unwrap() + .is_some(), expect, "failed for {lookup}, item {item}" ); diff --git a/tests/src/smtp/inbound/data.rs b/tests/src/smtp/inbound/data.rs index 9149745f..05f2262d 100644 --- a/tests/src/smtp/inbound/data.rs +++ b/tests/src/smtp/inbound/data.rs @@ -150,8 +150,9 @@ async fn data() { session .send_message("john@doe.org", &["bill@foobar.org"], "test:no_msgid", "250") .await; + qr.read_event().await.assert_reload(); assert_eq!( - qr.read_event().await.unwrap_message().read_message(), + qr.last_queued_message().await.read_message(&core).await, load_test_message("no_msgid", "messages") ); @@ -168,10 +169,11 @@ async fn data() { session .send_message("john@doe.org", &["mike@test.com"], "test:no_msgid", "250") .await; - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("From: ") .assert_contains("To: ") .assert_contains("Subject: ") diff --git a/tests/src/smtp/inbound/dmarc.rs b/tests/src/smtp/inbound/dmarc.rs index d7a54f8c..e5255744 100644 --- a/tests/src/smtp/inbound/dmarc.rs +++ b/tests/src/smtp/inbound/dmarc.rs @@ -180,13 +180,15 @@ async fn dmarc() { session.mail_from("bill@example.com", "550 5.7.23").await; // Expect SPF auth failure report - let message = qr.read_event().await.unwrap_message(); + qr.read_event().await.assert_reload(); + let message = qr.last_queued_message().await; assert_eq!( message.recipients.last().unwrap().address, "spf-failures@example.com" ); message - .read_lines() + .read_lines(&core) + .await .assert_contains("DKIM-Signature: v=1; a=rsa-sha256; s=rsa; d=example.com;") .assert_contains("To: spf-failures@example.com") .assert_contains("Feedback-Type: auth-failure") @@ -194,7 +196,7 @@ async fn dmarc() { // Second DKIM failure report should be rate limited session.mail_from("bill@example.com", "550 5.7.23").await; - qr.assert_empty_queue(); + qr.assert_no_events(); // Invalid DKIM signatures should be rejected session.data.remote_ip_str = "10.0.0.1".to_string(); @@ -210,13 +212,15 @@ async fn dmarc() { .await; // Expect DKIM auth failure report - let message = qr.read_event().await.unwrap_message(); + qr.read_event().await.assert_reload(); + let message = qr.last_queued_message().await; assert_eq!( message.recipients.last().unwrap().address, "dkim-failures@example.com" ); message - .read_lines() + .read_lines(&core) + .await .assert_contains("DKIM-Signature: v=1; a=rsa-sha256; s=rsa; d=example.com;") .assert_contains("To: dkim-failures@example.com") .assert_contains("Feedback-Type: auth-failure") @@ -231,7 +235,7 @@ async fn dmarc() { "550 5.7.20", ) .await; - qr.assert_empty_queue(); + qr.assert_no_events(); // Invalid ARC should be rejected session @@ -242,7 +246,7 @@ async fn dmarc() { "550 5.7.29", ) .await; - qr.assert_empty_queue(); + qr.assert_no_events(); // Unaligned DMARC should be rejected core.resolvers.dns.txt_add( @@ -260,13 +264,15 @@ async fn dmarc() { .await; // Expect DMARC auth failure report - let message = qr.read_event().await.unwrap_message(); + qr.read_event().await.assert_reload(); + let message = qr.last_queued_message().await; assert_eq!( message.recipients.last().unwrap().address, "dmarc-failures@example.com" ); message - .read_lines() + .read_lines(&core) + .await .assert_contains("DKIM-Signature: v=1; a=rsa-sha256; s=rsa; d=example.com;") .assert_contains("To: dmarc-failures@example.com") .assert_contains("Feedback-Type: auth-failure") @@ -289,7 +295,7 @@ async fn dmarc() { "550 5.7.1", ) .await; - qr.assert_empty_queue(); + qr.assert_no_events(); // Messagess passing DMARC should be accepted session @@ -300,10 +306,11 @@ async fn dmarc() { "250", ) .await; - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("dkim=pass") .assert_contains("spf=pass") .assert_contains("dmarc=pass") diff --git a/tests/src/smtp/inbound/milter.rs b/tests/src/smtp/inbound/milter.rs index 89fa2a70..aca8ee33 100644 --- a/tests/src/smtp/inbound/milter.rs +++ b/tests/src/smtp/inbound/milter.rs @@ -97,7 +97,7 @@ async fn milter_session() { "503 5.5.3", ) .await; - qr.assert_empty_queue(); + qr.assert_no_events(); // Test discard session @@ -108,7 +108,7 @@ async fn milter_session() { "250 2.0.0", ) .await; - qr.assert_empty_queue(); + qr.assert_no_events(); // Test temp fail session @@ -119,7 +119,7 @@ async fn milter_session() { "451 4.3.5", ) .await; - qr.assert_empty_queue(); + qr.assert_no_events(); // Test shutdown session @@ -130,7 +130,7 @@ async fn milter_session() { "421 4.3.0", ) .await; - qr.assert_empty_queue(); + qr.assert_no_events(); // Test reply code session @@ -141,7 +141,7 @@ async fn milter_session() { "321", ) .await; - qr.assert_empty_queue(); + qr.assert_no_events(); // Test accept with header addition session @@ -152,10 +152,11 @@ async fn milter_session() { "250 2.0.0", ) .await; - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("X-Hello: World") .assert_contains("Subject: Is dinner ready?") .assert_contains("Are you hungry yet?"); @@ -169,10 +170,11 @@ async fn milter_session() { "250 2.0.0", ) .await; - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("Subject: [SPAM] Saying Hello") .assert_count("References: ", 1) .assert_contains("Are you hungry yet?"); @@ -186,10 +188,11 @@ async fn milter_session() { "250 2.0.0", ) .await; - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("X-Spam: Yes") .assert_contains("123456"); } diff --git a/tests/src/smtp/inbound/mod.rs b/tests/src/smtp/inbound/mod.rs index 91e549d5..af037346 100644 --- a/tests/src/smtp/inbound/mod.rs +++ b/tests/src/smtp/inbound/mod.rs @@ -23,10 +23,15 @@ use std::time::Duration; +use store::{ + write::{key::DeserializeBigEndian, Bincode, QueueClass, QueueEvent, ValueClass}, + Deserialize, IterateParams, ValueKey, U64_LEN, +}; use tokio::sync::mpsc::error::TryRecvError; use smtp::{ - queue::{self, Message, OnHold, Schedule, WorkerResult}, + core::SMTP, + queue::{self, DeliveryAttempt, Message, OnHold, QueueId}, reporting::{self, DmarcEvent, TlsEvent}, }; @@ -64,16 +69,106 @@ impl QueueReceiver { Err(_) => None, } } - pub fn assert_empty_queue(&mut self) { + + pub fn assert_no_events(&mut self) { match self.queue_rx.try_recv() { Err(TryRecvError::Empty) => (), - Ok(queue::Event::Queue(message)) => { - panic!("Unexpected message: {}", message.inner.read_message()); - } Ok(event) => panic!("Expected empty queue but got {event:?}"), Err(err) => panic!("Queue error: {err:?}"), } } + + pub async fn assert_queue_is_empty(&mut self) { + assert_eq!(self.read_queued_messages().await, vec![]); + assert_eq!(self.read_queued_events().await, vec![]); + } + + pub async fn expect_message(&mut self) -> Message { + self.read_event().await.assert_reload(); + self.last_queued_message().await + } + + pub async fn expect_message_then_deliver(&mut self) -> DeliveryAttempt { + let message = self.expect_message().await; + let event = QueueEvent { + due: self.message_due(message.id).await, + queue_id: message.id, + }; + + DeliveryAttempt::new(message, event) + } + + pub async fn read_queued_events(&self) -> Vec { + let mut events = Vec::new(); + + let from_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent(QueueEvent { + due: 0, + queue_id: 0, + }))); + let to_key = ValueKey::from(ValueClass::Queue(QueueClass::MessageEvent(QueueEvent { + due: u64::MAX, + queue_id: u64::MAX, + }))); + + self.store + .iterate( + IterateParams::new(from_key, to_key).ascending().no_values(), + |key, _| { + events.push(QueueEvent { + due: key.deserialize_be_u64(1)?, + queue_id: key.deserialize_be_u64(U64_LEN + 1)?, + }); + Ok(true) + }, + ) + .await + .unwrap(); + + events + } + + pub async fn read_queued_messages(&self) -> Vec { + let from_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(0))); + let to_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(u64::MAX))); + let mut messages = Vec::new(); + + self.store + .iterate( + IterateParams::new(from_key, to_key).ascending(), + |key, value| { + let value = Bincode::::deserialize(value)?; + assert_eq!(key.deserialize_be_u64(1)?, value.inner.id); + messages.push(value.inner); + Ok(true) + }, + ) + .await + .unwrap(); + + messages + } + + pub async fn last_queued_message(&self) -> Message { + self.read_queued_messages() + .await + .into_iter() + .next() + .expect("No messages found in queue") + } + + pub async fn message_due(&self, queue_id: QueueId) -> u64 { + self.read_queued_events() + .await + .iter() + .find_map(|event| { + if event.queue_id == queue_id { + Some(event.due) + } else { + None + } + }) + .expect("No event found in queue for message") + } } impl ReportReceiver { @@ -102,65 +197,21 @@ impl ReportReceiver { } pub trait TestQueueEvent { - fn unwrap_message(self) -> Box; - fn unwrap_schedule(self) -> Schedule>; - fn unwrap_result(self) -> WorkerResult; - fn unwrap_done(self); - fn unwrap_on_hold(self) -> OnHold>; - fn unwrap_retry(self) -> Schedule>; + fn assert_reload(self); + fn unwrap_on_hold(self) -> OnHold; } impl TestQueueEvent for queue::Event { - fn unwrap_message(self) -> Box { + fn assert_reload(self) { match self { - queue::Event::Queue(message) => message.inner, + queue::Event::Reload => (), e => panic!("Unexpected event: {e:?}"), } } - fn unwrap_schedule(self) -> Schedule> { + fn unwrap_on_hold(self) -> OnHold { match self { - queue::Event::Queue(message) => message, - e => panic!("Unexpected event: {e:?}"), - } - } - - fn unwrap_result(self) -> WorkerResult { - match self { - queue::Event::Done(result) => result, - queue::Event::Queue(message) => { - panic!("Unexpected message: {}", message.inner.read_message()); - } - e => panic!("Unexpected event: {e:?}"), - } - } - - fn unwrap_done(self) { - match self { - queue::Event::Done(WorkerResult::Done) => (), - queue::Event::Queue(message) => { - panic!("Unexpected message: {}", message.inner.read_message()); - } - e => panic!("Unexpected event: {e:?}"), - } - } - - fn unwrap_on_hold(self) -> OnHold> { - match self { - queue::Event::Done(WorkerResult::OnHold(value)) => value, - queue::Event::Queue(message) => { - panic!("Unexpected message: {}", message.inner.read_message()); - } - e => panic!("Unexpected event: {e:?}"), - } - } - - fn unwrap_retry(self) -> Schedule> { - match self { - queue::Event::Done(WorkerResult::Retry(value)) => value, - queue::Event::Queue(message) => { - panic!("Unexpected message: {}", message.inner.read_message()); - } + queue::Event::OnHold(value) => value, e => panic!("Unexpected event: {e:?}"), } } @@ -188,20 +239,26 @@ impl TestReportingEvent for reporting::Event { } pub trait TestMessage { - fn read_message(&self) -> String; - fn read_lines(&self) -> Vec; + async fn read_message(&self, core: &SMTP) -> String; + async fn read_lines(&self, core: &SMTP) -> Vec; } impl TestMessage for Message { - fn read_message(&self) -> String { - let mut buf = vec![0u8; self.size]; - let mut file = std::fs::File::open(&self.path).unwrap(); - std::io::Read::read_exact(&mut file, &mut buf).unwrap(); - String::from_utf8(buf).unwrap() + async fn read_message(&self, core: &SMTP) -> String { + String::from_utf8( + core.shared + .default_blob_store + .get_blob(self.blob_hash.as_slice(), 0..u32::MAX) + .await + .unwrap() + .expect("Message blob not found"), + ) + .unwrap() } - fn read_lines(&self) -> Vec { - self.read_message() + async fn read_lines(&self, core: &SMTP) -> Vec { + self.read_message(core) + .await .split('\n') .map(|l| l.to_string()) .collect() diff --git a/tests/src/smtp/inbound/scripts.rs b/tests/src/smtp/inbound/scripts.rs index e579d6fc..ba6fe104 100644 --- a/tests/src/smtp/inbound/scripts.rs +++ b/tests/src/smtp/inbound/scripts.rs @@ -228,13 +228,14 @@ async fn sieve_scripts() { // Expect a modified message session.data("test:multipart", "250").await; - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("X-Part-Number: 5") .assert_contains("THIS IS A PIECE OF HTML TEXT"); - qr.assert_empty_queue(); + qr.assert_no_events(); // Expect rejection for bill@foobar.net session @@ -245,7 +246,7 @@ async fn sieve_scripts() { "503 5.5.3 Bill cannot receive messages", ) .await; - qr.assert_empty_queue(); + qr.assert_no_events(); // Expect message delivery plus a notification session @@ -256,7 +257,8 @@ async fn sieve_scripts() { "250", ) .await; - let notification = qr.read_event().await.unwrap_message(); + qr.read_event().await.assert_reload(); + let notification = qr.last_queued_message().await; assert_eq!(notification.return_path, ""); assert_eq!(notification.recipients.len(), 2); assert_eq!( @@ -268,22 +270,24 @@ async fn sieve_scripts() { "jane@example.org" ); notification - .read_lines() + .read_lines(&core) + .await .assert_contains("DKIM-Signature: v=1; a=rsa-sha256; s=rsa; d=example.com;") .assert_contains("From: \"Sieve Daemon\" ") .assert_contains("To: ") .assert_contains("Cc: ") .assert_contains("Subject: You have got mail") .assert_contains("One Two Three Four"); - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("One Two Three Four") .assert_contains("multi-part message in MIME format") .assert_not_contains("X-Part-Number: 5") .assert_not_contains("THIS IS A PIECE OF HTML TEXT"); - qr.assert_empty_queue(); + qr.assert_no_events(); // Expect a modified message delivery plus a notification session @@ -295,10 +299,11 @@ async fn sieve_scripts() { ) .await; - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("DKIM-Signature: v=1; a=rsa-sha256; s=rsa; d=example.com;") .assert_contains("From: \"Sieve Daemon\" ") .assert_contains("To: ") @@ -306,10 +311,11 @@ async fn sieve_scripts() { .assert_contains("Subject: You have got mail") .assert_contains("One Two Three Four"); - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("X-Part-Number: 5") .assert_contains("THIS IS A PIECE OF HTML TEXT") .assert_not_contains("X-My-Header: true"); @@ -323,7 +329,8 @@ async fn sieve_scripts() { "250", ) .await; - let redirect = qr.read_event().await.unwrap_message(); + qr.read_event().await.assert_reload(); + let redirect = qr.last_queued_message().await; assert_eq!(redirect.return_path, ""); assert_eq!(redirect.recipients.len(), 1); assert_eq!( @@ -331,13 +338,14 @@ async fn sieve_scripts() { "redirect@here.email" ); redirect - .read_lines() + .read_lines(&core) + .await .assert_contains("From: no-reply@my.domain") .assert_contains("To: Suzie Q ") .assert_contains("Subject: Is dinner ready?") .assert_contains("Message-ID: <20030712040037.46341.5F8J@football.example.com>") .assert_not_contains("From: Joe SixPack "); - qr.assert_empty_queue(); + qr.assert_no_events(); // Expect an intact redirected message session @@ -348,7 +356,8 @@ async fn sieve_scripts() { "250", ) .await; - let redirect = qr.read_event().await.unwrap_message(); + qr.read_event().await.assert_reload(); + let redirect = qr.last_queued_message().await; assert_eq!(redirect.return_path, ""); assert_eq!(redirect.recipients.len(), 1); assert_eq!( @@ -356,13 +365,14 @@ async fn sieve_scripts() { "redirect@somewhere.email" ); redirect - .read_lines() + .read_lines(&core) + .await .assert_not_contains("From: no-reply@my.domain") .assert_contains("To: Suzie Q ") .assert_contains("Subject: Is dinner ready?") .assert_contains("Message-ID: <20030712040037.46341.5F8J@football.example.com>") .assert_contains("From: Joe SixPack "); - qr.assert_empty_queue(); + qr.assert_no_events(); // Test pipes session.data.remote_ip_str = "10.0.0.123".parse().unwrap(); @@ -375,11 +385,12 @@ async fn sieve_scripts() { "250", ) .await; - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("X-My-Header: true") .assert_contains("Authentication-Results"); - qr.assert_empty_queue(); + qr.assert_no_events(); } diff --git a/tests/src/smtp/inbound/sign.rs b/tests/src/smtp/inbound/sign.rs index 909bdee8..c8e4323d 100644 --- a/tests/src/smtp/inbound/sign.rs +++ b/tests/src/smtp/inbound/sign.rs @@ -192,10 +192,11 @@ async fn sign_and_seal() { "250", ) .await; - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains( "DKIM-Signature: v=1; a=rsa-sha256; s=rsa; d=example.com; c=simple/relaxed;", ); @@ -204,10 +205,11 @@ async fn sign_and_seal() { session .send_message("bill@foobar.org", &["jdoe@example.com"], "test:arc", "250") .await; - qr.read_event() + qr.read_event().await.assert_reload(); + qr.last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("ARC-Seal: i=3; a=ed25519-sha256; s=ed; d=example.com; cv=pass;") .assert_contains( "ARC-Message-Signature: i=3; a=ed25519-sha256; s=ed; d=example.com; c=relaxed/simple;", diff --git a/tests/src/smtp/management/queue.rs b/tests/src/smtp/management/queue.rs index c757beb0..8c46ff6c 100644 --- a/tests/src/smtp/management/queue.rs +++ b/tests/src/smtp/management/queue.rs @@ -40,10 +40,7 @@ use crate::smtp::{ }; use smtp::{ core::{management::Message, Session, SMTP}, - queue::{ - manager::{Queue, SpawnQueue}, - QueueId, Status, - }, + queue::{manager::SpawnQueue, QueueId, Status}, }; const DIRECTORY: &str = r#" @@ -108,7 +105,7 @@ async fn manage_queue() { core.queue.config.expire = IfBlock::new(Duration::from_secs(3000)); let local_qr = core.init_test_queue("smtp_manage_queue_local"); let core = Arc::new(core); - local_qr.queue_rx.spawn(core.clone(), Queue::default()); + local_qr.queue_rx.spawn(core.clone()); let _rx_manage = start_test_server(core.clone(), &[ServerProtocol::Http]); // Send test messages @@ -172,11 +169,11 @@ async fn manage_queue() { // Expect delivery to success@foobar.org tokio::time::sleep(Duration::from_millis(100)).await; + remote_qr.read_event().await.assert_reload(); assert_eq!( remote_qr - .read_event() + .last_queued_message() .await - .unwrap_message() .recipients .into_iter() .map(|r| r.address) @@ -323,11 +320,11 @@ async fn manage_queue() { // Expect delivery to john@foobar.org tokio::time::sleep(Duration::from_millis(100)).await; + remote_qr.read_event().await.assert_reload(); assert_eq!( remote_qr - .read_event() + .last_queued_message() .await - .unwrap_message() .recipients .into_iter() .map(|r| r.address) diff --git a/tests/src/smtp/management/report.rs b/tests/src/smtp/management/report.rs index 7871be40..c6e1f2a0 100644 --- a/tests/src/smtp/management/report.rs +++ b/tests/src/smtp/management/report.rs @@ -44,10 +44,7 @@ use crate::smtp::{ use smtp::{ config::AggregateFrequency, core::{management::Report, SMTP}, - reporting::{ - scheduler::{Scheduler, SpawnReport}, - DmarcEvent, TlsEvent, - }, + reporting::{scheduler::SpawnReport, DmarcEvent, TlsEvent}, }; const DIRECTORY: &str = r#" @@ -77,8 +74,6 @@ async fn manage_reports() { let mut core = SMTP::test(); let temp_dir = make_temp_dir("smtp_report_management_test", true); let config = &mut core.report.config; - config.path = temp_dir.temp_dir.clone(); - config.hash = IfBlock::new(16); config.dmarc_aggregate.max_size = IfBlock::new(1024); config.tls.max_size = IfBlock::new(1024); let directory = Config::new(DIRECTORY) @@ -90,7 +85,7 @@ async fn manage_reports() { let (report_tx, report_rx) = mpsc::channel(1024); core.report.tx = report_tx; let core = Arc::new(core); - report_rx.spawn(core.clone(), Scheduler::default()); + report_rx.spawn(core.clone()); let _rx_manage = start_test_server(core.clone(), &[ServerProtocol::Http]); // Send test reporting events diff --git a/tests/src/smtp/mod.rs b/tests/src/smtp/mod.rs index 3f58bc89..98363689 100644 --- a/tests/src/smtp/mod.rs +++ b/tests/src/smtp/mod.rs @@ -33,7 +33,7 @@ use mail_auth::{ use mail_send::smtp::tls::build_tls_connector; use sieve::Runtime; use smtp_proto::{AUTH_LOGIN, AUTH_PLAIN}; -use store::{LookupStore, Store}; +use store::{backend::sqlite::SqliteStore, LookupStore, Store}; use tokio::sync::mpsc; use smtp::{ @@ -55,7 +55,10 @@ use smtp::{ }, outbound::dane::DnssecResolver, }; -use utils::config::{if_block::IfBlock, utils::ConstantValue, Config}; +use utils::{ + config::{if_block::IfBlock, utils::ConstantValue, Config}, + snowflake::SnowflakeIdGenerator, +}; pub mod config; pub mod inbound; @@ -201,6 +204,7 @@ impl TestConfig for SMTP { blocked_ips: Arc::new(Default::default()), }), default_lookup_store: LookupStore::Store(store.clone()), + default_blob_store: store::BlobStore::Store(store.clone()), default_data_store: store, }, } @@ -300,13 +304,8 @@ impl TestConfig for QueueCore { ThrottleKeyHasherBuilder::default(), 16, ), - quota: DashMap::with_capacity_and_hasher_and_shard_amount( - 10, - ThrottleKeyHasherBuilder::default(), - 16, - ), tx: mpsc::channel(1024).0, - id_seq: 0.into(), + snowflake_id: SnowflakeIdGenerator::new(), connectors: TlsConnectors { pki_verify: build_tls_connector(false), dummy_verify: build_tls_connector(true), @@ -318,8 +317,6 @@ impl TestConfig for QueueCore { impl TestConfig for QueueConfig { fn test() -> Self { Self { - path: Default::default(), - hash: IfBlock::new(10), retry: IfBlock::new(Duration::from_secs(10)), notify: IfBlock::new(Duration::from_secs(20)), expire: IfBlock::new(Duration::from_secs(10)), @@ -404,8 +401,6 @@ impl TestConfig for ReportCore { impl TestConfig for ReportConfig { fn test() -> Self { Self { - path: Default::default(), - hash: IfBlock::new(10), submitter: IfBlock::new("example.org".to_string()), analysis: ReportAnalysis { addresses: vec![], @@ -502,6 +497,7 @@ pub fn add_test_certs(config: &str) -> String { pub struct QueueReceiver { _temp_dir: TempDir, + store: Store, pub queue_rx: mpsc::Receiver, } @@ -514,17 +510,27 @@ pub trait TestSMTP { fn init_test_report(&mut self) -> ReportReceiver; } +const QUEUE_STORE_CONFIG: &str = r#"[store."sqlite"] +type = "sqlite" +path = "{TMP}/queue.db" +"#; + impl TestSMTP for SMTP { fn init_test_queue(&mut self, test_name: &str) -> QueueReceiver { let _temp_dir = make_temp_dir(test_name, true); - self.queue.config.path = _temp_dir.temp_dir.clone(); - + let config = + Config::new(&QUEUE_STORE_CONFIG.replace("{TMP}", _temp_dir.temp_dir.to_str().unwrap())) + .unwrap(); + let store = Store::SQLite(SqliteStore::open(&config, "store.sqlite").unwrap().into()); + self.shared.default_data_store = store.clone(); + self.shared.default_blob_store = store.clone().into(); let (queue_tx, queue_rx) = mpsc::channel(128); self.queue.tx = queue_tx; QueueReceiver { - _temp_dir, + store, queue_rx, + _temp_dir, } } diff --git a/tests/src/smtp/outbound/dane.rs b/tests/src/smtp/outbound/dane.rs index ac498b3d..4d2b6a32 100644 --- a/tests/src/smtp/outbound/dane.rs +++ b/tests/src/smtp/outbound/dane.rs @@ -114,17 +114,20 @@ async fn dane_verify() { session .send_message("john@test.org", &["bill@foobar.org"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; local_qr - .read_event() + .last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains(" (DANE failed to authenticate") .assert_contains("No TLSA records found"); - local_qr.read_event().await.unwrap_done(); + local_qr.read_event().await.assert_reload(); + local_qr.assert_queue_is_empty().await; // Expect TLS failure report let report = rr.read_report().await.unwrap_tls(); @@ -162,17 +165,20 @@ async fn dane_verify() { session .send_message("john@test.org", &["bill@foobar.org"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; local_qr - .read_event() + .expect_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains(" (DANE failed to authenticate") .assert_contains("No matching certificates found"); - local_qr.read_event().await.unwrap_done(); + local_qr.read_event().await.assert_reload(); + local_qr.assert_queue_is_empty().await; // Expect TLS failure report let report = rr.read_report().await.unwrap_tls(); @@ -181,7 +187,7 @@ async fn dane_verify() { report.failure.as_ref().unwrap().result_type, ResultType::ValidationFailure ); - remote_qr.assert_empty_queue(); + remote_qr.assert_no_events(); // DANE successful delivery let tlsa = Arc::new(Tlsa { @@ -205,15 +211,18 @@ async fn dane_verify() { session .send_message("john@test.org", &["bill@foobar.org"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) - .await; - local_qr.read_event().await.unwrap_done(); - remote_qr - .read_event() + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) + .await; + local_qr.read_event().await.assert_reload(); + local_qr.assert_queue_is_empty().await; + remote_qr + .last_queued_message() + .await + .read_lines(&core) .await - .unwrap_message() - .read_lines() .assert_contains("using TLSv1.3 with cipher"); // Expect TLS success report diff --git a/tests/src/smtp/outbound/extensions.rs b/tests/src/smtp/outbound/extensions.rs index 55d39979..67b886c8 100644 --- a/tests/src/smtp/outbound/extensions.rs +++ b/tests/src/smtp/outbound/extensions.rs @@ -94,15 +94,15 @@ async fn extensions() { "250", ) .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr.expect_message_then_deliver().await + .try_deliver(core.clone()) .await; local_qr .read_event() .await .unwrap_message() - .read_lines() + .read_lines(&core).await .assert_contains(" (delivered to") .assert_contains("Final-Recipient: rfc822;bill@foobar.org") .assert_contains("Action: delivered"); @@ -111,21 +111,21 @@ async fn extensions() { .read_event() .await .unwrap_message() - .read_lines() + .read_lines(&core).await .assert_contains("using TLSv1.3 with cipher"); // Test SIZE extension session .send_message("john@test.org", &["bill@foobar.org"], "test:arc", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr.expect_message_then_deliver().await + .try_deliver(core.clone()) .await; local_qr .read_event() .await .unwrap_message() - .read_lines() + .read_lines(&core).await .assert_contains(" (host 'mx.foobar.org' rejected command 'MAIL FROM:") .assert_contains("Action: failed") .assert_contains("Diagnostic-Code: smtp;552") @@ -142,11 +142,11 @@ async fn extensions() { "250", ) .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr.expect_message_then_deliver().await + .try_deliver(core.clone()) .await; local_qr.read_event().await.unwrap_done(); - let message = remote_qr.read_event().await.unwrap_message(); + let message = remote_qr.expect_message().await(); assert_eq!(message.env_id, Some("abc123".to_string())); assert!((message.flags & MAIL_RET_HDRS) != 0); assert!((message.flags & MAIL_REQUIRETLS) != 0); diff --git a/tests/src/smtp/outbound/ip_lookup.rs b/tests/src/smtp/outbound/ip_lookup.rs index f4a45c2c..e317b1cc 100644 --- a/tests/src/smtp/outbound/ip_lookup.rs +++ b/tests/src/smtp/outbound/ip_lookup.rs @@ -93,12 +93,12 @@ async fn ip_lookup_strategy() { session .send_message("john@test.org", &["bill@foobar.org"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr.expect_message_then_deliver().await + .try_deliver(core.clone()) .await; if matches!(strategy, IpLookupStrategy::Ipv6thenIpv4) { local_qr.read_event().await.unwrap_done(); - remote_qr.read_event().await.unwrap_message(); + remote_qr.expect_message().await(); } else { let status = local_qr.read_event().await.unwrap_retry().inner.domains[0] .status diff --git a/tests/src/smtp/outbound/lmtp.rs b/tests/src/smtp/outbound/lmtp.rs index e198babc..c862527f 100644 --- a/tests/src/smtp/outbound/lmtp.rs +++ b/tests/src/smtp/outbound/lmtp.rs @@ -35,7 +35,7 @@ use crate::smtp::{ use smtp::{ config::shared::ConfigShared, core::{Session, SMTP}, - queue::{manager::Queue, DeliveryAttempt, Event, WorkerResult}, + queue::{manager::Queue, DeliveryAttempt, Event}, }; use utils::config::{if_block::IfBlock, Config, ServerProtocol}; @@ -119,8 +119,10 @@ async fn lmtp_delivery() { "250", ) .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; let mut dsn = Vec::new(); loop { @@ -144,7 +146,7 @@ async fn lmtp_delivery() { if !queue.scheduled.is_empty() { tokio::time::sleep(queue.wake_up_time()).await; DeliveryAttempt::from(queue.next_due().unwrap()) - .try_deliver(core.clone(), &mut queue) + .try_deliver(core.clone()) .await; } } @@ -155,7 +157,8 @@ async fn lmtp_delivery() { dsn.next() .unwrap() - .read_lines() + .read_lines(&core) + .await .assert_contains(" (delivered to") .assert_contains(" (delivered to") .assert_contains(" (delivered to") @@ -164,19 +167,22 @@ async fn lmtp_delivery() { dsn.next() .unwrap() - .read_lines() + .read_lines(&core) + .await .assert_contains(" (host 'lmtp.foobar.org' rejected") .assert_contains("Action: delayed"); dsn.next() .unwrap() - .read_lines() + .read_lines(&core) + .await .assert_contains(" (host 'lmtp.foobar.org' rejected") .assert_contains("Action: delayed"); dsn.next() .unwrap() - .read_lines() + .read_lines(&core) + .await .assert_contains(" (host 'lmtp.foobar.org' rejected") .assert_contains("Action: failed"); diff --git a/tests/src/smtp/outbound/mta_sts.rs b/tests/src/smtp/outbound/mta_sts.rs index 1f66b7fb..08b483aa 100644 --- a/tests/src/smtp/outbound/mta_sts.rs +++ b/tests/src/smtp/outbound/mta_sts.rs @@ -101,14 +101,17 @@ async fn mta_sts_verify() { session .send_message("john@test.org", &["bill@foobar.org"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; local_qr .read_event() .await .unwrap_message() - .read_lines() + .read_lines(&core) + .await .assert_contains(" (MTA-STS failed to authenticate") .assert_contains("Record not found"); local_qr.read_event().await.unwrap_done(); @@ -135,14 +138,17 @@ async fn mta_sts_verify() { session .send_message("john@test.org", &["bill@foobar.org"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; local_qr .read_event() .await .unwrap_message() - .read_lines() + .read_lines(&core) + .await .assert_contains(" (MTA-STS failed to authenticate") .assert_contains("No 'mx' entries found"); local_qr.read_event().await.unwrap_done(); @@ -166,14 +172,17 @@ async fn mta_sts_verify() { session .send_message("john@test.org", &["bill@foobar.org"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; local_qr .read_event() .await .unwrap_message() - .read_lines() + .read_lines(&core) + .await .assert_contains(" (MTA-STS failed to authenticate") .assert_contains("not authorized by policy"); local_qr.read_event().await.unwrap_done(); @@ -213,15 +222,18 @@ async fn mta_sts_verify() { session .send_message("john@test.org", &["bill@foobar.org"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; local_qr.read_event().await.unwrap_done(); remote_qr .read_event() .await .unwrap_message() - .read_lines() + .read_lines(&core) + .await .assert_contains("using TLSv1.3 with cipher"); // Expect TLS success report diff --git a/tests/src/smtp/outbound/smtp.rs b/tests/src/smtp/outbound/smtp.rs index 6ed79bef..5d867e31 100644 --- a/tests/src/smtp/outbound/smtp.rs +++ b/tests/src/smtp/outbound/smtp.rs @@ -37,7 +37,7 @@ use crate::smtp::{ }; use smtp::{ core::{Session, SMTP}, - queue::{manager::Queue, DeliveryAttempt, Event, WorkerResult}, + queue::{manager::Queue, DeliveryAttempt, Event}, }; const SMUGGLER: &str = r#"From: Joe SixPack @@ -139,11 +139,11 @@ async fn smtp_delivery() { "250", ) .await; - let message = local_qr.read_event().await.unwrap_message(); + let message = local_qr.expect_message().await(); let num_domains = message.domains.len(); assert_eq!(num_domains, 3); DeliveryAttempt::from(message) - .try_deliver(core.clone(), &mut queue) + .try_deliver(core.clone()) .await; let mut dsn = Vec::new(); let mut domain_retries = vec![0; num_domains]; @@ -171,7 +171,7 @@ async fn smtp_delivery() { if !queue.scheduled.is_empty() { tokio::time::sleep(queue.wake_up_time()).await; DeliveryAttempt::from(queue.next_due().unwrap()) - .try_deliver(core.clone(), &mut queue) + .try_deliver(core.clone()) .await; } } @@ -190,7 +190,7 @@ async fn smtp_delivery() { dsn.next() .unwrap() - .read_lines() + .read_lines(&core).await .assert_contains(" (delivered to") .assert_contains(" (delivered to") .assert_contains(" (failed to lookup") @@ -199,25 +199,25 @@ async fn smtp_delivery() { dsn.next() .unwrap() - .read_lines() + .read_lines(&core).await .assert_contains(" (host ") .assert_contains(" (host ") .assert_contains("Action: delayed"); dsn.next() .unwrap() - .read_lines() + .read_lines(&core).await .assert_contains(" (host ") .assert_contains("Action: delayed"); dsn.next() .unwrap() - .read_lines() + .read_lines(&core).await .assert_contains(" (host "); dsn.next() .unwrap() - .read_lines() + .read_lines(&core).await .assert_contains(" (host ") .assert_contains("Action: failed"); @@ -261,8 +261,8 @@ async fn smtp_delivery() { session .send_message("john@doe.org", &["bill@foobar.com"], &message, "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr.expect_message_then_deliver().await + .try_deliver(core.clone()) .await; let event = local_qr.read_event().await; @@ -272,7 +272,7 @@ async fn smtp_delivery() { event ); - let message = remote_qr.read_event().await.unwrap_message().read_message(); + let message = remote_qr.expect_message().await().read_message(); assert!( message.contains("This is a smuggled message"), diff --git a/tests/src/smtp/outbound/throttle.rs b/tests/src/smtp/outbound/throttle.rs index 8b383d75..b33d9a67 100644 --- a/tests/src/smtp/outbound/throttle.rs +++ b/tests/src/smtp/outbound/throttle.rs @@ -118,8 +118,10 @@ async fn throttle_outbound() { assert!(!in_flight.is_empty()); // Expect concurrency throttle for sender domain 'foobar.org' - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; local_qr.assert_empty_queue(); in_flight.clear(); @@ -142,8 +144,10 @@ async fn throttle_outbound() { session .send_message("john@foobar.net", &["bill@test.org"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; local_qr.assert_empty_queue(); assert!([1799, 1800].contains( @@ -177,8 +181,10 @@ async fn throttle_outbound() { "250", ) .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; local_qr.read_event().await.unwrap_on_hold(); in_flight.clear(); @@ -203,8 +209,10 @@ async fn throttle_outbound() { "250", ) .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; assert!([2399, 2400].contains( &local_qr @@ -244,8 +252,10 @@ async fn throttle_outbound() { session .send_message("john@test.net", &["jane@test.org"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; local_qr.read_event().await.unwrap_on_hold(); in_flight.clear(); @@ -278,8 +288,10 @@ async fn throttle_outbound() { session .send_message("john@test.net", &["jane@test.net"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; assert!([2999, 3000].contains( &local_qr diff --git a/tests/src/smtp/outbound/tls.rs b/tests/src/smtp/outbound/tls.rs index 6e084800..3008eb74 100644 --- a/tests/src/smtp/outbound/tls.rs +++ b/tests/src/smtp/outbound/tls.rs @@ -88,20 +88,23 @@ async fn starttls_optional() { session .send_message("john@test.org", &["bill@foobar.org"], "test:no_dkim", "250") .await; - DeliveryAttempt::from(local_qr.read_event().await.unwrap_message()) - .try_deliver(core.clone(), &mut queue) + local_qr + .expect_message_then_deliver() + .await + .try_deliver(core.clone()) .await; let mut retry = local_qr.read_event().await.unwrap_retry(); assert!(retry.inner.domains[0].disable_tls); retry.inner.domains[0].retry.due = Instant::now(); DeliveryAttempt::from(retry.inner) - .try_deliver(core.clone(), &mut queue) + .try_deliver(core.clone()) .await; local_qr.read_event().await.unwrap_done(); remote_qr .read_event() .await .unwrap_message() - .read_lines() + .read_lines(&core) + .await .assert_not_contains("using TLSv1.3 with cipher"); } diff --git a/tests/src/smtp/queue/dsn.rs b/tests/src/smtp/queue/dsn.rs index 9590e85f..49ae936c 100644 --- a/tests/src/smtp/queue/dsn.rs +++ b/tests/src/smtp/queue/dsn.rs @@ -121,7 +121,7 @@ async fn generate_dsn() { // Failure DSN attempt.message.recipients[0].flags = flags; core.send_dsn(&mut attempt).await; - compare_dsn(qr.read_event().await.unwrap_message(), "failure.eml").await; + compare_dsn(qr.expect_message().await(), "failure.eml").await; // Success DSN attempt.message.recipients.push(Recipient { @@ -140,7 +140,7 @@ async fn generate_dsn() { orcpt: None, }); core.send_dsn(&mut attempt).await; - compare_dsn(qr.read_event().await.unwrap_message(), "success.eml").await; + compare_dsn(qr.expect_message().await(), "success.eml").await; // Delay DSN attempt.message.recipients.push(Recipient { @@ -152,7 +152,7 @@ async fn generate_dsn() { orcpt: "jdoe@example.org".to_string().into(), }); core.send_dsn(&mut attempt).await; - compare_dsn(qr.read_event().await.unwrap_message(), "delay.eml").await; + compare_dsn(qr.expect_message().await(), "delay.eml").await; // Mixed DSN for rcpt in &mut attempt.message.recipients { @@ -160,7 +160,7 @@ async fn generate_dsn() { } attempt.message.domains[0].notify.due = Instant::now(); core.send_dsn(&mut attempt).await; - compare_dsn(qr.read_event().await.unwrap_message(), "mixed.eml").await; + compare_dsn(qr.expect_message().await(), "mixed.eml").await; // Load queue let queue = core.queue.read_queue().await; diff --git a/tests/src/smtp/queue/retry.rs b/tests/src/smtp/queue/retry.rs index 08932265..b8b26a40 100644 --- a/tests/src/smtp/queue/retry.rs +++ b/tests/src/smtp/queue/retry.rs @@ -33,7 +33,7 @@ use crate::smtp::{ }; use smtp::{ core::{Session, SMTP}, - queue::{manager::Queue, DeliveryAttempt, Event, WorkerResult}, + queue::{manager::Queue, DeliveryAttempt, Event}, }; use utils::config::if_block::IfBlock; @@ -75,17 +75,17 @@ async fn queue_retry() { session .send_message("john@test.org", &["bill@foobar.org"], "test:no_dkim", "250") .await; - let attempt = DeliveryAttempt::from(qr.read_event().await.unwrap_message()); + let attempt = DeliveryAttempt::from(qr.expect_message().await()); // Expect a failed DSN let path = attempt.message.path.clone(); - attempt.try_deliver(core.clone(), &mut queue).await; - let message = qr.read_event().await.unwrap_message(); + attempt.try_deliver(core.clone()).await; + let message = qr.expect_message().await(); assert_eq!(message.return_path, ""); assert_eq!(message.domains.first().unwrap().domain, "test.org"); assert_eq!(message.recipients.first().unwrap().address, "john@test.org"); message - .read_lines() + .read_lines(&core).await .assert_contains("Content-Type: multipart/report") .assert_contains("Final-Recipient: rfc822;bill@foobar.org") .assert_contains("Action: failed"); @@ -102,11 +102,11 @@ async fn queue_retry() { "250", ) .await; - let attempt = DeliveryAttempt::from(qr.read_event().await.unwrap_message()); + let attempt = DeliveryAttempt::from(qr.expect_message().await()); let path = attempt.message.path.clone(); let mut dsn = Vec::new(); let mut num_retries = 0; - attempt.try_deliver(core.clone(), &mut queue).await; + attempt.try_deliver(core.clone()).await; loop { match qr.try_read_event().await { Some(Event::Queue(message)) => { @@ -127,7 +127,7 @@ async fn queue_retry() { if !queue.scheduled.is_empty() { tokio::time::sleep(queue.wake_up_time()).await; DeliveryAttempt::from(queue.next_due().unwrap()) - .try_deliver(core.clone(), &mut queue) + .try_deliver(core.clone()) .await; } } @@ -139,28 +139,28 @@ async fn queue_retry() { dsn.next() .unwrap() - .read_lines() + .read_lines(&core).await .assert_contains(" (failed to lookup 'foobar.org'") .assert_contains("Final-Recipient: rfc822;bill@foobar.org") .assert_contains("Action: failed"); dsn.next() .unwrap() - .read_lines() + .read_lines(&core).await .assert_contains(" (failed to lookup '_dns_error.org'") .assert_contains("Final-Recipient: rfc822;jane@_dns_error.org") .assert_contains("Action: delayed"); dsn.next() .unwrap() - .read_lines() + .read_lines(&core).await .assert_contains(" (failed to lookup '_dns_error.org'") .assert_contains("Final-Recipient: rfc822;jane@_dns_error.org") .assert_contains("Action: delayed"); dsn.next() .unwrap() - .read_lines() + .read_lines(&core).await .assert_contains(" (failed to lookup '_dns_error.org'") .assert_contains("Final-Recipient: rfc822;jane@_dns_error.org") .assert_contains("Action: failed"); diff --git a/tests/src/smtp/queue/serialize.rs b/tests/src/smtp/queue/serialize.rs index 37fb1a24..76b461f0 100644 --- a/tests/src/smtp/queue/serialize.rs +++ b/tests/src/smtp/queue/serialize.rs @@ -110,7 +110,7 @@ async fn queue_serialize() { ) .await ); - let mut message = qr.read_event().await.unwrap_message(); + let mut message = qr.expect_message().await(); // Deserialize assert_msg_eq( diff --git a/tests/src/smtp/reporting/analyze.rs b/tests/src/smtp/reporting/analyze.rs index 9c36896c..fb17d4c0 100644 --- a/tests/src/smtp/reporting/analyze.rs +++ b/tests/src/smtp/reporting/analyze.rs @@ -96,5 +96,6 @@ async fn report_analyze() { session .send_message("john@test.org", &["bill@foobar.org"], "test:no_dkim", "250") .await; - qr.read_event().await.unwrap_message(); + qr.read_event().await.assert_reload(); + qr.last_queued_message().await; } diff --git a/tests/src/smtp/reporting/dmarc.rs b/tests/src/smtp/reporting/dmarc.rs index 29d1009b..a7262b46 100644 --- a/tests/src/smtp/reporting/dmarc.rs +++ b/tests/src/smtp/reporting/dmarc.rs @@ -43,11 +43,7 @@ use crate::smtp::{ use smtp::{ config::{AggregateFrequency, ConfigContext}, core::SMTP, - reporting::{ - dmarc::GenerateDmarcReport, - scheduler::{ReportType, Scheduler}, - DmarcEvent, - }, + reporting::DmarcEvent, }; #[tokio::test] @@ -64,15 +60,12 @@ async fn report_dmarc() { core.shared.signers = ConfigContext::new(&[]).parse_signatures().signers; let temp_dir = make_temp_dir("smtp_report_dmarc_test", true); let config = &mut core.report.config; - config.path = temp_dir.temp_dir.clone(); - config.hash = IfBlock::new(16); config.dmarc_aggregate.sign = "\"['rsa']\"".parse_if(); config.dmarc_aggregate.max_size = IfBlock::new(4096); config.submitter = IfBlock::new("mx.example.org".to_string()); config.dmarc_aggregate.address = IfBlock::new("reports@example.org".to_string()); config.dmarc_aggregate.org_name = IfBlock::new("Foobar, Inc.".to_string()); config.dmarc_aggregate.contact_info = IfBlock::new("https://foobar.org/contact".to_string()); - let mut scheduler = Scheduler::default(); // Authorize external report for foobar.org core.resolvers.dns.txt_add( @@ -94,40 +87,32 @@ async fn report_dmarc() { ); assert_eq!(dmarc_record.rua().len(), 2); for _ in 0..2 { - scheduler - .schedule_dmarc( - Box::new(DmarcEvent { - domain: "foobar.org".to_string(), - report_record: Record::new() - .with_source_ip("192.168.1.2".parse().unwrap()) - .with_action_disposition(ActionDisposition::Pass) - .with_dmarc_dkim_result(DmarcResult::Pass) - .with_dmarc_spf_result(DmarcResult::Fail) - .with_envelope_from("hello@example.org") - .with_envelope_to("other@example.org") - .with_header_from("bye@example.org"), - dmarc_record: dmarc_record.clone(), - interval: AggregateFrequency::Weekly, - }), - &core, - ) - .await; - } - scheduler - .schedule_dmarc( - Box::new(DmarcEvent { - domain: "foobar.org".to_string(), - report_record: Record::new() - .with_source_ip("a:b:c::e:f".parse().unwrap()) - .with_action_disposition(ActionDisposition::Reject) - .with_dmarc_dkim_result(DmarcResult::Fail) - .with_dmarc_spf_result(DmarcResult::Pass), - dmarc_record: dmarc_record.clone(), - interval: AggregateFrequency::Weekly, - }), - &core, - ) + core.schedule_dmarc(Box::new(DmarcEvent { + domain: "foobar.org".to_string(), + report_record: Record::new() + .with_source_ip("192.168.1.2".parse().unwrap()) + .with_action_disposition(ActionDisposition::Pass) + .with_dmarc_dkim_result(DmarcResult::Pass) + .with_dmarc_spf_result(DmarcResult::Fail) + .with_envelope_from("hello@example.org") + .with_envelope_to("other@example.org") + .with_header_from("bye@example.org"), + dmarc_record: dmarc_record.clone(), + interval: AggregateFrequency::Weekly, + })) .await; + } + core.schedule_dmarc(Box::new(DmarcEvent { + domain: "foobar.org".to_string(), + report_record: Record::new() + .with_source_ip("a:b:c::e:f".parse().unwrap()) + .with_action_disposition(ActionDisposition::Reject) + .with_dmarc_dkim_result(DmarcResult::Fail) + .with_dmarc_spf_result(DmarcResult::Pass), + dmarc_record: dmarc_record.clone(), + interval: AggregateFrequency::Weekly, + })) + .await; assert_eq!(scheduler.reports.len(), 1); tokio::time::sleep(Duration::from_millis(200)).await; let report_path; @@ -140,7 +125,7 @@ async fn report_dmarc() { } // Expect report - let message = qr.read_event().await.unwrap_message(); + let message = qr.expect_message().await(); qr.assert_empty_queue(); assert_eq!(message.recipients.len(), 1); assert_eq!( @@ -149,7 +134,7 @@ async fn report_dmarc() { ); assert_eq!(message.return_path, "reports@example.org"); message - .read_lines() + .read_lines(&core).await .assert_contains("DKIM-Signature: v=1; a=rsa-sha256; s=rsa; d=example.com;") .assert_contains("To: ") .assert_contains("Report Domain: foobar.org") diff --git a/tests/src/smtp/reporting/scheduler.rs b/tests/src/smtp/reporting/scheduler.rs index d06971e7..402b08cb 100644 --- a/tests/src/smtp/reporting/scheduler.rs +++ b/tests/src/smtp/reporting/scheduler.rs @@ -36,11 +36,7 @@ use crate::smtp::{make_temp_dir, TestConfig}; use smtp::{ config::AggregateFrequency, core::SMTP, - reporting::{ - dmarc::DmarcFormat, - scheduler::{ReportType, Scheduler}, - DmarcEvent, PolicyType, TlsEvent, - }, + reporting::{dmarc::DmarcFormat, DmarcEvent, PolicyType, TlsEvent}, }; #[tokio::test] @@ -56,123 +52,92 @@ async fn report_scheduler() { let mut core = SMTP::test(); let temp_dir = make_temp_dir("smtp_report_scheduler_test", true); let config = &mut core.report.config; - config.path = temp_dir.temp_dir.clone(); - config.hash = IfBlock::new(16); config.dmarc_aggregate.max_size = IfBlock::new(500); config.tls.max_size = IfBlock::new(550); - let mut scheduler = Scheduler::default(); // Schedule two events with a same policy and another one with a different policy let dmarc_record = Arc::new(Dmarc::parse(b"v=DMARC1; p=quarantine; rua=mailto:dmarc@foobar.org").unwrap()); - scheduler - .schedule_dmarc( - Box::new(DmarcEvent { - domain: "foobar.org".to_string(), - report_record: Record::new() - .with_source_ip("192.168.1.2".parse().unwrap()) - .with_action_disposition(ActionDisposition::Pass) - .with_dmarc_dkim_result(DmarcResult::Pass) - .with_dmarc_spf_result(DmarcResult::Fail) - .with_envelope_from("hello@example.org") - .with_envelope_to("other@example.org") - .with_header_from("bye@example.org"), - dmarc_record: dmarc_record.clone(), - interval: AggregateFrequency::Weekly, - }), - &core, - ) - .await; + core.schedule_dmarc(Box::new(DmarcEvent { + domain: "foobar.org".to_string(), + report_record: Record::new() + .with_source_ip("192.168.1.2".parse().unwrap()) + .with_action_disposition(ActionDisposition::Pass) + .with_dmarc_dkim_result(DmarcResult::Pass) + .with_dmarc_spf_result(DmarcResult::Fail) + .with_envelope_from("hello@example.org") + .with_envelope_to("other@example.org") + .with_header_from("bye@example.org"), + dmarc_record: dmarc_record.clone(), + interval: AggregateFrequency::Weekly, + })) + .await; // No records should be added once the 550 bytes max size is reached for _ in 0..10 { - scheduler - .schedule_dmarc( - Box::new(DmarcEvent { - domain: "foobar.org".to_string(), - report_record: Record::new() - .with_source_ip("192.168.1.2".parse().unwrap()) - .with_action_disposition(ActionDisposition::Pass) - .with_dmarc_dkim_result(DmarcResult::Pass) - .with_dmarc_spf_result(DmarcResult::Fail) - .with_envelope_from("hello@example.org") - .with_envelope_to("other@example.org") - .with_header_from("bye@example.org"), - dmarc_record: dmarc_record.clone(), - interval: AggregateFrequency::Weekly, - }), - &core, - ) - .await; + core.schedule_dmarc(Box::new(DmarcEvent { + domain: "foobar.org".to_string(), + report_record: Record::new() + .with_source_ip("192.168.1.2".parse().unwrap()) + .with_action_disposition(ActionDisposition::Pass) + .with_dmarc_dkim_result(DmarcResult::Pass) + .with_dmarc_spf_result(DmarcResult::Fail) + .with_envelope_from("hello@example.org") + .with_envelope_to("other@example.org") + .with_header_from("bye@example.org"), + dmarc_record: dmarc_record.clone(), + interval: AggregateFrequency::Weekly, + })) + .await; } let dmarc_record = Arc::new(Dmarc::parse(b"v=DMARC1; p=reject; rua=mailto:dmarc@foobar.org").unwrap()); - scheduler - .schedule_dmarc( - Box::new(DmarcEvent { - domain: "foobar.org".to_string(), - report_record: Record::new() - .with_source_ip("a:b:c::e:f".parse().unwrap()) - .with_action_disposition(ActionDisposition::Reject) - .with_dmarc_dkim_result(DmarcResult::Fail) - .with_dmarc_spf_result(DmarcResult::Pass), - dmarc_record: dmarc_record.clone(), - interval: AggregateFrequency::Weekly, - }), - &core, - ) - .await; + core.schedule_dmarc(Box::new(DmarcEvent { + domain: "foobar.org".to_string(), + report_record: Record::new() + .with_source_ip("a:b:c::e:f".parse().unwrap()) + .with_action_disposition(ActionDisposition::Reject) + .with_dmarc_dkim_result(DmarcResult::Fail) + .with_dmarc_spf_result(DmarcResult::Pass), + dmarc_record: dmarc_record.clone(), + interval: AggregateFrequency::Weekly, + })) + .await; // Schedule TLS event let tls_record = Arc::new(TlsRpt::parse(b"v=TLSRPTv1;rua=mailto:reports@foobar.org").unwrap()); - scheduler - .schedule_tls( - Box::new(TlsEvent { - domain: "foobar.org".to_string(), - policy: PolicyType::Tlsa(None), - failure: None, - tls_record: tls_record.clone(), - interval: AggregateFrequency::Daily, - }), - &core, - ) - .await; - scheduler - .schedule_tls( - Box::new(TlsEvent { - domain: "foobar.org".to_string(), - policy: PolicyType::Tlsa(None), - failure: None, - tls_record: tls_record.clone(), - interval: AggregateFrequency::Daily, - }), - &core, - ) - .await; - scheduler - .schedule_tls( - Box::new(TlsEvent { - domain: "foobar.org".to_string(), - policy: PolicyType::Sts(None), - failure: None, - tls_record: tls_record.clone(), - interval: AggregateFrequency::Daily, - }), - &core, - ) - .await; - scheduler - .schedule_tls( - Box::new(TlsEvent { - domain: "foobar.org".to_string(), - policy: PolicyType::None, - failure: None, - tls_record: tls_record.clone(), - interval: AggregateFrequency::Daily, - }), - &core, - ) - .await; + core.schedule_tls(Box::new(TlsEvent { + domain: "foobar.org".to_string(), + policy: PolicyType::Tlsa(None), + failure: None, + tls_record: tls_record.clone(), + interval: AggregateFrequency::Daily, + })) + .await; + core.schedule_tls(Box::new(TlsEvent { + domain: "foobar.org".to_string(), + policy: PolicyType::Tlsa(None), + failure: None, + tls_record: tls_record.clone(), + interval: AggregateFrequency::Daily, + })) + .await; + core.schedule_tls(Box::new(TlsEvent { + domain: "foobar.org".to_string(), + policy: PolicyType::Sts(None), + failure: None, + tls_record: tls_record.clone(), + interval: AggregateFrequency::Daily, + })) + .await; + core.schedule_tls(Box::new(TlsEvent { + domain: "foobar.org".to_string(), + policy: PolicyType::None, + failure: None, + tls_record: tls_record.clone(), + interval: AggregateFrequency::Daily, + })) + .await; // Verify sizes and counts let mut total_tls = 0; diff --git a/tests/src/smtp/reporting/tls.rs b/tests/src/smtp/reporting/tls.rs index eecf5abe..3bd0246f 100644 --- a/tests/src/smtp/reporting/tls.rs +++ b/tests/src/smtp/reporting/tls.rs @@ -40,11 +40,7 @@ use crate::smtp::{ use smtp::{ config::{AggregateFrequency, ConfigContext}, core::SMTP, - reporting::{ - scheduler::{ReportType, Scheduler}, - tls::{GenerateTlsReport, TLS_HTTP_REPORT}, - TlsEvent, - }, + reporting::{tls::TLS_HTTP_REPORT, TlsEvent}, }; #[tokio::test] @@ -142,14 +138,14 @@ async fn report_tls() { } // Expect report - let message = qr.read_event().await.unwrap_message(); + let message = qr.expect_message().await(); assert_eq!( message.recipients.last().unwrap().address, "reports@foobar.org" ); assert_eq!(message.return_path, "reports@example.org"); message - .read_lines() + .read_lines(&core).await .assert_contains("DKIM-Signature: v=1; a=rsa-sha256; s=rsa; d=example.com;") .assert_contains("To: ") .assert_contains("Report Domain: foobar.org") diff --git a/tests/src/store/blob.rs b/tests/src/store/blob.rs index fc666354..a54d9711 100644 --- a/tests/src/store/blob.rs +++ b/tests/src/store/blob.rs @@ -25,9 +25,9 @@ use ahash::AHashMap; use store::{ config::ConfigStore, write::{blob::BlobQuota, now, BatchBuilder, BlobOp}, - BlobClass, BlobHash, BlobStore, Serialize, + BlobClass, BlobStore, Serialize, }; -use utils::config::Config; +use utils::{config::Config, BlobHash}; use crate::store::{TempDir, CONFIG}; diff --git a/tests/src/store/lookup.rs b/tests/src/store/lookup.rs index b37f6b29..e11a04b4 100644 --- a/tests/src/store/lookup.rs +++ b/tests/src/store/lookup.rs @@ -21,7 +21,7 @@ * for more details. */ -use store::{config::ConfigStore, LookupKey, LookupStore, LookupValue}; +use store::{config::ConfigStore, LookupStore}; use utils::config::Config; use crate::store::{TempDir, CONFIG}; @@ -33,6 +33,9 @@ pub async fn lookup_tests() { Config::new(&CONFIG.replace("{TMP}", temp_dir.path.as_path().to_str().unwrap())).unwrap(); let stores = config.parse_stores().await.unwrap(); + let todo = "test expiry counter + ratelimit"; + let todo = "use lookup ratelimit everywhere"; + for (store_id, store) in stores.lookup_stores { println!("Testing lookup store {}...", store_id); if let LookupStore::Store(store) = &store { @@ -40,13 +43,7 @@ pub async fn lookup_tests() { } else { // Reset redis counter store - .key_set( - "abc".as_bytes().to_vec(), - LookupValue::Value { - value: "0".as_bytes().to_vec(), - expires: 0, - }, - ) + .key_set("abc".as_bytes().to_vec(), "0".as_bytes().to_vec(), None) .await .unwrap(); } @@ -54,44 +51,26 @@ pub async fn lookup_tests() { // Test key let key = "xyz".as_bytes().to_vec(); store - .key_set( - key.clone(), - LookupValue::Value { - value: "world".to_string().into_bytes(), - expires: 0, - }, - ) + .key_set(key.clone(), "world".to_string().into_bytes(), None) .await .unwrap(); store.purge_expired().await.unwrap(); - assert!(matches!(store - .key_get::(LookupKey::Key(key.clone())) - .await - .unwrap(), LookupValue::Value { value,.. } if value == "world")); + assert_eq!( + store.key_get::(key.clone()).await.unwrap(), + Some("world".to_string()) + ); // Test value expiry store - .key_set( - key.clone(), - LookupValue::Value { - value: "hello".to_string().into_bytes(), - expires: 1, - }, - ) + .key_set(key.clone(), "hello".to_string().into_bytes(), 1.into()) .await .unwrap(); - assert!(matches!(store - .key_get::(LookupKey::Key(key.clone())) - .await - .unwrap(), LookupValue::Value { value,.. } if value == "hello")); - tokio::time::sleep(tokio::time::Duration::from_secs(2)).await; assert_eq!( - LookupValue::None, - store - .key_get::(LookupKey::Key(key.clone())) - .await - .unwrap() + store.key_get::(key.clone()).await.unwrap(), + Some("hello".to_string()) ); + tokio::time::sleep(tokio::time::Duration::from_secs(2)).await; + assert_eq!(None, store.key_get::(key.clone()).await.unwrap()); store.purge_expired().await.unwrap(); if let LookupStore::Store(store) = &store { @@ -100,27 +79,9 @@ pub async fn lookup_tests() { // Test counter let key = "abc".as_bytes().to_vec(); - store - .key_set(key.clone(), LookupValue::Counter { num: 1 }) - .await - .unwrap(); - assert_eq!( - LookupValue::Counter { num: 1 }, - store - .key_get::(LookupKey::Counter(key.clone())) - .await - .unwrap() - ); - store - .key_set(key.clone(), LookupValue::Counter { num: 2 }) - .await - .unwrap(); - assert_eq!( - LookupValue::Counter { num: 3 }, - store - .key_get::(LookupKey::Counter(key.clone())) - .await - .unwrap() - ); + store.counter_incr(key.clone(), 1, None).await.unwrap(); + assert_eq!(1, store.counter_get(key.clone()).await.unwrap()); + store.counter_incr(key.clone(), 2, None).await.unwrap(); + assert_eq!(3, store.counter_get(key.clone()).await.unwrap()); } }