commit 7bfaf2c5e3f7ee89c29cc9175da77ee7cb582b18
parent 3f6a0fd79c5c9f294536810a2edb2f5419acf326
Author: Antoine A <>
Date: Thu, 1 Oct 2026 12:07:34 +0200
common: improve test db logic with macros
Diffstat:
20 files changed, 506 insertions(+), 296 deletions(-)
diff --git a/adapters/taler-cyclos/Cargo.toml b/adapters/taler-cyclos/Cargo.toml
@@ -19,6 +19,7 @@ jiff.workspace = true
taler-common.workspace = true
taler-api.workspace = true
taler-build.workspace = true
+taler-macros.workspace = true
http-client.workspace = true
clap.workspace = true
serde.workspace = true
diff --git a/adapters/taler-cyclos/src/api.rs b/adapters/taler-cyclos/src/api.rs
@@ -399,9 +399,9 @@ mod test {
payto::{PaytoURI, payto},
},
};
+ use taler_macros::db_test;
use taler_test_utils::{
Router,
- db::db_test_setup,
routine::{
Status, admin_add_incoming_routine, in_history_routine, out_history_routine,
registration_routine, revenue_routine, transfer_routine,
@@ -412,7 +412,6 @@ mod test {
use crate::{
api::CyclosApi,
- constants::CONFIG_SOURCE,
db::{self, TxIn, TxOutKind},
payto::{FullCyclosPayto, cyclos_payto},
};
@@ -425,27 +424,24 @@ mod test {
payto("payto://cyclos/localhost/7762070814178012478?receiver-name=Unknown")
});
- async fn setup() -> (Router, PgPool) {
- let (_, pool) = db_test_setup(CONFIG_SOURCE).await;
+ async fn setup(pool: &PgPool) -> Router {
let api = Arc::new(CyclosApi::start(
pool.clone(),
CompactString::const_new("localhost"),
PAYTO.clone(),
Currency::TEST,
));
- let server = Router::new()
+ Router::new()
.wire_gateway(api.clone(), AuthMethod::None)
.prepared_transfer(api.clone())
.revenue(api.clone(), AuthMethod::None)
.observability(api, AuthMethod::None)
- .finalize();
-
- (server, pool)
+ .finalize()
}
- #[tokio::test]
- async fn config() {
- let (server, _) = setup().await;
+ #[db_test]
+ async fn config(pool: PgPool) {
+ let server = setup(&pool).await;
server
.get("/taler-wire-gateway/config")
.await
@@ -465,9 +461,9 @@ mod test {
server.get("/taler-observability/metrics").await.assert_ok();
}
- #[tokio::test]
- async fn transfer() {
- let (server, _) = setup().await;
+ #[db_test]
+ async fn transfer(pool: PgPool) {
+ let server = setup(&pool).await;
transfer_routine(
&server.prefix("/taler-wire-gateway"),
TransferState::pending,
@@ -539,9 +535,10 @@ mod test {
out(db, &TxOutKind::Simple).await
}
- #[tokio::test]
- async fn outgoing_history() {
- let (server, db) = &setup().await;
+ #[db_test]
+ async fn outgoing_history(db: PgPool) {
+ let db = &db;
+ let server = setup(db).await;
out_history_routine(
&server.prefix("/taler-wire-gateway"),
@@ -556,9 +553,9 @@ mod test {
.await;
}
- #[tokio::test]
- async fn admin_add_incoming() {
- let (server, _) = setup().await;
+ #[db_test]
+ async fn admin_add_incoming(db: PgPool) {
+ let server = setup(&db).await;
admin_add_incoming_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-prepared-transfer"),
@@ -568,9 +565,10 @@ mod test {
.await;
}
- #[tokio::test]
- async fn in_history() {
- let (server, db) = &setup().await;
+ #[db_test]
+ async fn in_history(db: PgPool) {
+ let db = &db;
+ let server = setup(db).await;
in_history_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-prepared-transfer"),
@@ -587,9 +585,10 @@ mod test {
.await;
}
- #[tokio::test]
- async fn revenue() {
- let (server, db) = &setup().await;
+ #[db_test]
+ async fn revenue(db: PgPool) {
+ let db = &db;
+ let server = setup(db).await;
revenue_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-revenue"),
@@ -634,9 +633,9 @@ mod test {
.unwrap()
}
- #[tokio::test]
- async fn registration() {
- let (server, pool) = setup().await;
+ #[db_test]
+ async fn registration(pool: PgPool) {
+ let server = setup(&pool).await;
registration_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-prepared-transfer"),
diff --git a/adapters/taler-cyclos/src/db.rs b/adapters/taler-cyclos/src/db.rs
@@ -916,7 +916,7 @@ mod test {
use compact_str::CompactString;
use jiff::{Span, Timestamp};
use serde_json::json;
- use sqlx::{PgPool, Postgres, Row as _, pool::PoolConnection, postgres::PgRow};
+ use sqlx::{PgConnection, PgPool, Row as _, postgres::PgRow};
use taler_api::{
db::TypeHelper,
notification::dummy_listen,
@@ -934,26 +934,18 @@ mod test {
utils::now_sql_stable_ts,
},
};
+ use taler_macros::db_test;
- use crate::{
- constants::CONFIG_SOURCE,
- db::{
- self, AddIncomingResult, AddOutgoingResult, BounceResult, ChargebackFailureResult,
- Transfer, TransferResult, TxIn, TxInAdmin, TxOut, TxOutKind,
- },
+ use crate::db::{
+ self, AddIncomingResult, AddOutgoingResult, BounceResult, ChargebackFailureResult,
+ Transfer, TransferResult, TxIn, TxInAdmin, TxOut, TxOutKind,
};
pub const CURR: Currency = Currency::TEST;
pub const ROOT: CompactString = CompactString::const_new("localhost");
- async fn setup() -> (PoolConnection<Postgres>, PgPool) {
- taler_test_utils::db::db_test_setup(CONFIG_SOURCE).await
- }
-
- #[tokio::test]
- async fn kv() {
- let (mut db, _) = setup().await;
-
+ #[db_test]
+ async fn kv(mut db: PgConnection) {
let value = json!({
"name": "Mr Smith",
"no way": 32
@@ -975,14 +967,12 @@ mod test {
);
}
- #[tokio::test]
- async fn tx_in() {
- let (mut db, pool) = setup().await;
-
+ #[db_test]
+ async fn tx_in(mut db: PgConnection, pool: PgPool) {
let mut routine = async |first: &Option<IncomingKey>, second: &Option<IncomingKey>| {
let id = sqlx::query("SELECT count(*) + 1 FROM tx_in")
.try_map(|r: PgRow| r.try_get_u64(0))
- .fetch_one(&mut *db)
+ .fetch_one(&mut db)
.await
.unwrap();
let now = now_sql_stable_ts();
@@ -1102,10 +1092,8 @@ mod test {
);
}
- #[tokio::test]
- async fn tx_in_admin() {
- let (_, pool) = setup().await;
-
+ #[db_test]
+ async fn tx_in_admin(pool: PgPool) {
// Empty db
assert_eq!(
db::incoming_history(&pool, &History::default(), &CURR, &ROOT, dummy_listen)
@@ -1166,14 +1154,12 @@ mod test {
);
}
- #[tokio::test]
- async fn tx_out() {
- let (mut db, pool) = setup().await;
-
+ #[db_test]
+ async fn tx_out(mut db: PgConnection, pool: PgPool) {
let mut routine = async |first: &TxOutKind, second: &TxOutKind| {
let transfer_id = sqlx::query("SELECT count(*) + 1 FROM tx_out")
.try_map(|r: PgRow| r.try_get(0))
- .fetch_one(&mut *db)
+ .fetch_one(&mut db)
.await
.unwrap();
let now = now_sql_stable_ts();
@@ -1293,10 +1279,8 @@ mod test {
// TODO tx out failure
- #[tokio::test]
- async fn transfer() {
- let (_, pool) = setup().await;
-
+ #[db_test]
+ async fn transfer(pool: PgPool) {
// Empty db
assert_eq!(
db::transfer_by_id(&pool, 0, &CURR, &ROOT).await.unwrap(),
@@ -1415,10 +1399,8 @@ mod test {
);
}
- #[tokio::test]
- async fn bounce() {
- let (mut db, _) = setup().await;
-
+ #[db_test]
+ async fn bounce(mut db: PgConnection) {
let amount = decimal("10");
let now = now_sql_stable_ts();
@@ -1497,10 +1479,8 @@ mod test {
);
}
- #[tokio::test]
- async fn status() {
- let (mut db, pool) = setup().await;
-
+ #[db_test]
+ async fn status(mut db: PgConnection, pool: PgPool) {
let check_status = async |id: u64, status: TransferState, msg: Option<&str>| {
let transfer = db::transfer_by_id(&pool, id, &CURR, &ROOT)
.await
@@ -1597,9 +1577,8 @@ mod test {
);
}
- #[tokio::test]
- async fn batch() {
- let (mut db, pool) = setup().await;
+ #[db_test]
+ async fn batch(mut db: PgConnection, pool: PgPool) {
let start = Timestamp::now();
// Empty db
diff --git a/adapters/taler-cyclos/src/lib.rs b/adapters/taler-cyclos/src/lib.rs
@@ -32,6 +32,7 @@ pub mod notification;
pub mod payto;
pub mod setup;
pub mod worker;
+pub use crate::constants::CONFIG_SOURCE;
pub async fn run_serve(cfg: &Config, pool: PgPool) -> anyhow::Result<()> {
let cfg = ServeCfg::parse(cfg)?;
diff --git a/adapters/taler-magnet-bank/Cargo.toml b/adapters/taler-magnet-bank/Cargo.toml
@@ -23,6 +23,7 @@ jiff.workspace = true
taler-common.workspace = true
taler-api.workspace = true
taler-build.workspace = true
+taler-macros.workspace = true
http-client.workspace = true
clap.workspace = true
serde.workspace = true
diff --git a/adapters/taler-magnet-bank/src/api.rs b/adapters/taler-magnet-bank/src/api.rs
@@ -371,9 +371,9 @@ mod test {
payto::{PaytoURI, payto},
},
};
+ use taler_macros::db_test;
use taler_test_utils::{
Router,
- db::db_test_setup,
routine::{
Status, admin_add_incoming_routine, in_history_routine, out_history_routine,
registration_routine, revenue_routine, transfer_routine,
@@ -385,7 +385,6 @@ mod test {
use crate::{
FullHuPayto,
api::MagnetApi,
- constants::CONFIG_SOURCE,
db::{self, TxIn, TxOutKind},
magnet_api::types::TxStatus,
magnet_payto,
@@ -398,22 +397,19 @@ mod test {
static UNKNOWN: LazyLock<PaytoURI> =
LazyLock::new(|| payto("payto://iban/HU60162006491000639900000000?receiver-name=Unknown"));
- async fn setup() -> (Router, PgPool) {
- let (_, pool) = db_test_setup(CONFIG_SOURCE).await;
+ async fn setup(pool: &PgPool) -> Router {
let api = Arc::new(MagnetApi::start(pool.clone(), PAYTO.clone()).await);
- let server = Router::new()
+ Router::new()
.wire_gateway(api.clone(), AuthMethod::None)
.prepared_transfer(api.clone())
.revenue(api.clone(), AuthMethod::None)
.observability(api, AuthMethod::None)
- .finalize();
-
- (server, pool)
+ .finalize()
}
- #[tokio::test]
- async fn config() {
- let (server, _) = setup().await;
+ #[db_test]
+ async fn config(pool: PgPool) {
+ let server = setup(&pool).await;
server
.get("/taler-wire-gateway/config")
.await
@@ -433,9 +429,9 @@ mod test {
server.get("/taler-observability/metrics").await.assert_ok();
}
- #[tokio::test]
- async fn transfer() {
- let (server, _) = setup().await;
+ #[db_test]
+ async fn transfer(pool: PgPool) {
+ let server = setup(&pool).await;
transfer_routine(
&server.prefix("/taler-wire-gateway"),
TransferState::pending,
@@ -504,9 +500,10 @@ mod test {
out(db, &TxOutKind::Simple).await
}
- #[tokio::test]
- async fn outgoing_history() {
- let (server, db) = &setup().await;
+ #[db_test]
+ async fn outgoing_history(db: PgPool) {
+ let db = &db;
+ let server = setup(db).await;
out_history_routine(
&server.prefix("/taler-wire-gateway"),
tasks!({ out_talerable(db).await }),
@@ -520,9 +517,9 @@ mod test {
.await;
}
- #[tokio::test]
- async fn admin_add_incoming() {
- let (server, _) = setup().await;
+ #[db_test]
+ async fn admin_add_incoming(db: PgPool) {
+ let server = setup(&db).await;
admin_add_incoming_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-prepared-transfer"),
@@ -532,9 +529,10 @@ mod test {
.await;
}
- #[tokio::test]
- async fn in_history() {
- let (server, db) = &setup().await;
+ #[db_test]
+ async fn in_history(db: PgPool) {
+ let db = &db;
+ let server = setup(db).await;
in_history_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-prepared-transfer"),
@@ -551,9 +549,10 @@ mod test {
.await;
}
- #[tokio::test]
- async fn revenue() {
- let (server, db) = &setup().await;
+ #[db_test]
+ async fn revenue(db: PgPool) {
+ let db = &db;
+ let server = setup(db).await;
revenue_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-revenue"),
@@ -598,16 +597,16 @@ mod test {
.unwrap()
}
- #[tokio::test]
- async fn registration() {
- let (server, pool) = setup().await;
+ #[db_test]
+ async fn registration(db: PgPool) {
+ let server = setup(&db).await;
registration_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-prepared-transfer"),
&EXCHANGE,
&EXCHANGE,
&UNKNOWN,
- || check_in(&pool),
+ || check_in(&db),
)
.await;
}
diff --git a/adapters/taler-magnet-bank/src/db.rs b/adapters/taler-magnet-bank/src/db.rs
@@ -881,7 +881,7 @@ pub async fn transfer_unregister(db: &PgPool, req: &Unregistration) -> sqlx::Res
mod test {
use jiff::{Span, Timestamp, Zoned, tz::TimeZone};
use serde_json::json;
- use sqlx::{PgPool, Postgres, pool::PoolConnection, postgres::PgRow};
+ use sqlx::{PgConnection, PgPool, postgres::PgRow};
use taler_api::{
db::TypeHelper,
notification::dummy_listen,
@@ -898,10 +898,10 @@ mod test {
utils::now_sql_stable_ts,
},
};
+ use taler_macros::db_test;
use super::TxInAdmin;
use crate::{
- constants::CONFIG_SOURCE,
db::{
self, AddIncomingResult, AddOutgoingResult, BounceResult, Initiated, OutFailureResult,
TransferResult, TxIn, TxOut, TxOutKind, kv_get, kv_set, make_transfer,
@@ -911,14 +911,8 @@ mod test {
magnet_payto,
};
- async fn setup() -> (PoolConnection<Postgres>, PgPool) {
- taler_test_utils::db::db_test_setup(CONFIG_SOURCE).await
- }
-
- #[tokio::test]
- async fn kv() {
- let (mut db, _) = setup().await;
-
+ #[db_test]
+ async fn kv(mut db: PgConnection) {
let value = json!({
"name": "Mr Smith",
"no way": 32
@@ -936,15 +930,13 @@ mod test {
);
}
- #[tokio::test]
- async fn tx_in() {
- let (mut db, pool) = setup().await;
-
+ #[db_test]
+ async fn tx_in(mut db: PgConnection, pool: PgPool) {
let mut routine = async |first: &Option<IncomingKey>, second: &Option<IncomingKey>| {
let (id, code) =
sqlx::query("SELECT count(*) + 1, COALESCE(max(magnet_code), 0) + 20 FROM tx_in")
.try_map(|r: PgRow| Ok((r.try_get_u64(0)?, r.try_get_u64(1)?)))
- .fetch_one(&mut *db)
+ .fetch_one(&mut db)
.await
.unwrap();
let now = now_sql_stable_ts();
@@ -1063,10 +1055,8 @@ mod test {
);
}
- #[tokio::test]
- async fn tx_in_admin() {
- let (_, pool) = setup().await;
-
+ #[db_test]
+ async fn tx_in_admin(pool: PgPool) {
// Empty db
assert_eq!(
db::incoming_history(&pool, &History::default(), dummy_listen)
@@ -1128,15 +1118,13 @@ mod test {
);
}
- #[tokio::test]
- async fn tx_out() {
- let (mut db, pool) = setup().await;
-
+ #[db_test]
+ async fn tx_out(mut db: PgConnection, pool: PgPool) {
let mut routine = async |first: &TxOutKind, second: &TxOutKind| {
let (id, code) =
sqlx::query("SELECT count(*) + 1, COALESCE(max(magnet_code), 0) + 20 FROM tx_out")
.try_map(|r: PgRow| Ok((r.try_get_u64(0)?, r.try_get_u64(1)?)))
- .fetch_one(&mut *db)
+ .fetch_one(&mut db)
.await
.unwrap();
let now = now_sql_stable_ts();
@@ -1254,10 +1242,8 @@ mod test {
);
}
- #[tokio::test]
- async fn tx_out_failure() {
- let (mut db, pool) = setup().await;
-
+ #[db_test]
+ async fn tx_out_failure(mut db: PgConnection, pool: PgPool) {
let now = now_sql_stable_ts();
// Unknown
@@ -1359,10 +1345,8 @@ mod test {
);
}
- #[tokio::test]
- async fn transfer() {
- let (_, pool) = setup().await;
-
+ #[db_test]
+ async fn transfer(pool: PgPool) {
// Empty db
assert_eq!(db::transfer_by_id(&pool, 0).await.unwrap(), None);
assert_eq!(
@@ -1458,10 +1442,8 @@ mod test {
);
}
- #[tokio::test]
- async fn bounce() {
- let (mut db, _) = setup().await;
-
+ #[db_test]
+ async fn bounce(mut db: PgConnection) {
let amount = amount("HUF:10");
let payto = magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name");
let now = now_sql_stable_ts();
@@ -1613,10 +1595,8 @@ mod test {
);
}
- #[tokio::test]
- async fn status() {
- let (mut db, _) = setup().await;
-
+ #[db_test]
+ async fn status(mut db: PgConnection) {
// Unknown transfer
db::initiated_submit_permanent_failure(&mut db, 1, &Timestamp::now(), "msg")
.await
@@ -1626,9 +1606,8 @@ mod test {
.unwrap();
}
- #[tokio::test]
- async fn batch() {
- let (mut db, pool) = setup().await;
+ #[db_test]
+ async fn batch(mut db: PgConnection, pool: PgPool) {
let start = Timestamp::now();
let magnet_payto =
magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name");
diff --git a/adapters/taler-wise/Cargo.toml b/adapters/taler-wise/Cargo.toml
@@ -17,6 +17,7 @@ sqlx.workspace = true
taler-api.workspace = true
taler-build.workspace = true
taler-common.workspace = true
+taler-macros.workspace = true
url.workspace = true
http-client.workspace = true
serde.workspace = true
diff --git a/adapters/taler-wise/src/api.rs b/adapters/taler-wise/src/api.rs
@@ -364,16 +364,16 @@ mod test {
payto::{BankID, PaytoURI},
},
};
+ use taler_macros::db_test;
use taler_test_utils::{
Router,
- db::db_test_setup,
routine::{admin_add_incoming_routine, in_history_routine, revenue_routine},
server::TestServer,
tasks,
};
use crate::{
- CONFIG_SOURCE, api,
+ api,
config::WiseBalance,
db::{TxIn, register_tx_in},
payto::{FullWisePayto, WiseAccount},
@@ -386,8 +386,7 @@ mod test {
});
static EXCHANGE: LazyLock<PaytoURI> = LazyLock::new(|| PAYTO.as_uri());
- async fn setup() -> (Router, PgPool) {
- let (_, pool) = db_test_setup(CONFIG_SOURCE).await;
+ async fn setup(pool: &PgPool) -> Router {
let balances = vec![
WiseBalance {
id: 42,
@@ -400,7 +399,7 @@ mod test {
payto: PAYTO.clone().into_inner().0,
},
];
- let server = api::start(
+ api::start(
pool.clone(),
"Exchange",
balances,
@@ -415,14 +414,12 @@ mod test {
}),
)
.await
- .finalize();
-
- (server, pool)
+ .finalize()
}
- #[tokio::test]
- async fn config() {
- let (server, _) = setup().await;
+ #[db_test]
+ async fn config(db: PgPool) {
+ let server = setup(&db).await;
for id in [42, 34] {
server
.get(format!("/balances/{id}/taler-wire-gateway/config"))
@@ -477,9 +474,9 @@ mod test {
r#in(db, Some(IncomingKey::reserve(EddsaPublicKey::rand()))).await
}
- #[tokio::test]
- async fn admin_add_incoming() {
- let (server, _) = setup().await;
+ #[db_test]
+ async fn admin_add_incoming(db: PgPool) {
+ let server = setup(&db).await;
admin_add_incoming_routine(
&server.prefix("/balances/42/taler-wire-gateway"),
&server.prefix("/balances/42/taler-prepared-transfer"),
@@ -489,9 +486,10 @@ mod test {
.await;
}
- #[tokio::test]
- async fn in_history() {
- let (server, db) = &setup().await;
+ #[db_test]
+ async fn in_history(db: PgPool) {
+ let server = setup(&db).await;
+ let db = &db;
in_history_routine(
&server.prefix("/balances/42/taler-wire-gateway"),
&server.prefix("/balances/42/taler-prepared-transfer"),
@@ -503,9 +501,10 @@ mod test {
.await;
}
- #[tokio::test]
- async fn revenue() {
- let (server, db) = &setup().await;
+ #[db_test]
+ async fn revenue(db: PgPool) {
+ let server = setup(&db).await;
+ let db = &db;
revenue_routine(
&server.prefix("/balances/42/taler-wire-gateway"),
&server.prefix("/balances/42/taler-revenue"),
diff --git a/adapters/taler-wise/src/constants.rs b/adapters/taler-wise/src/constants.rs
@@ -0,0 +1,19 @@
+/*
+ This file is part of TALER
+ Copyright (C) 2026 Taler Systems SA
+
+ TALER is free software; you can redistribute it and/or modify it under the
+ terms of the GNU Affero General Public License as published by the Free Software
+ Foundation; either version 3, or (at your option) any later version.
+
+ TALER is distributed in the hope that it will be useful, but WITHOUT ANY
+ WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.
+
+ You should have received a copy of the GNU Affero General Public License along with
+ TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+*/
+
+use taler_common::config::parser::ConfigSource;
+
+pub const CONFIG_SOURCE: ConfigSource = ConfigSource::new("taler-wise", "wise", "taler-wise");
diff --git a/adapters/taler-wise/src/db.rs b/adapters/taler-wise/src/db.rs
@@ -331,7 +331,7 @@ mod test {
use compact_str::{CompactString, format_compact};
use jiff::Span;
- use sqlx::{PgPool, Postgres, pool::PoolConnection, postgres::PgRow};
+ use sqlx::{PgConnection, PgPool, postgres::PgRow};
use taler_api::{db::TypeHelper, notification::dummy_listen, subject::IncomingKey};
use taler_common::{
api::{EddsaPublicKey, params::History},
@@ -342,28 +342,22 @@ mod test {
utils::now_sql_stable_ts,
},
};
+ use taler_macros::db_test;
use crate::{
- CONFIG_SOURCE,
db::{AddIncomingResult, TxIn, incoming_history, register_tx_in, revenue_history},
payto::WiseAccount,
};
- async fn setup() -> (PoolConnection<Postgres>, PgPool) {
- taler_test_utils::db::db_test_setup(CONFIG_SOURCE).await
- }
-
const BALANCE_ID: u32 = 42;
const CURR: Currency = Currency::TEST;
- #[tokio::test]
- async fn tx_in() {
- let (mut db, pool) = setup().await;
-
+ #[db_test]
+ async fn tx_in(mut db: PgConnection, pool: PgPool) {
let mut routine = async |first: &Option<IncomingKey>, second: &Option<IncomingKey>| {
let id = sqlx::query("SELECT count(*) + 1 FROM tx_in")
.try_map(|r: PgRow| r.try_get_u64(0))
- .fetch_one(&mut *db)
+ .fetch_one(&mut db)
.await
.unwrap();
let now = now_sql_stable_ts();
diff --git a/adapters/taler-wise/src/lib.rs b/adapters/taler-wise/src/lib.rs
@@ -14,14 +14,11 @@
TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
*/
-use taler_common::config::parser::ConfigSource;
-
pub mod api;
pub mod config;
+pub mod constants;
pub mod db;
pub mod payto;
pub mod setup;
pub mod wise_api;
pub mod worker;
-
-pub const CONFIG_SOURCE: ConfigSource = ConfigSource::new("taler-wise", "wise", "taler-wise");
diff --git a/adapters/taler-wise/src/main.rs b/adapters/taler-wise/src/main.rs
@@ -19,8 +19,9 @@ use taler_api::api::TalerRouter as _;
use taler_build::long_version;
use taler_common::{CommonArgs, cli::ConfigCmd, config::Config, taler_main};
use taler_wise::{
- CONFIG_SOURCE, api,
+ api,
config::ServeCfg,
+ constants::CONFIG_SOURCE,
db::{dbinit, pool},
setup::{self},
worker::run_worker,
diff --git a/common/taler-api/Cargo.toml b/common/taler-api/Cargo.toml
@@ -18,6 +18,7 @@ http-body-util.workspace = true
zlib-rs = "0.6.3"
tower-http = { version = "0.7", features = ["cors"] }
tokio = { workspace = true, features = ["signal"] }
+taler-macros.workspace = true
rand.workspace = true
serde.workspace = true
serde_json.workspace = true
diff --git a/common/taler-api/src/test.rs b/common/taler-api/src/test.rs
@@ -41,8 +41,8 @@ use taler_common::{
url,
},
};
+use taler_macros::db_test;
use taler_test_utils::{
- db::db_test_setup_manual,
routine::{
Status, admin_add_incoming_routine, in_history_routine, out_history_routine,
registration_routine, revenue_routine, transfer_routine,
@@ -94,17 +94,13 @@ fn test_api(pool: PgPool, currency: Currency) -> Router {
.observability(state, AuthMethod::None)
}
-async fn setup() -> (Router, PgPool) {
- let (_, pool) = db_test_setup_manual("db".as_ref(), "taler-api").await;
- (
- test_api(pool.clone(), "EUR".parse().unwrap()).finalize(),
- pool,
- )
+async fn setup(db: &PgPool) -> Router {
+ test_api(db.clone(), "EUR".parse().unwrap()).finalize()
}
-#[tokio::test]
-async fn body_parsing() {
- let (server, _) = setup().await;
+#[db_test(sql_dir = "db", component = "taler-api")]
+async fn body_parsing(db: PgPool) {
+ let server = setup(&db).await;
let normal_body = TransferRequest {
request_uid: Base32::rand(),
amount: Amount::zero(&Currency::EUR),
@@ -185,9 +181,9 @@ async fn body_parsing() {
.assert_error(ErrorCode::GENERIC_UPLOAD_EXCEEDS_LIMIT);
}
-#[tokio::test]
-async fn errors() {
- let (server, _) = setup().await;
+#[db_test(sql_dir = "db", component = "taler-api")]
+async fn errors(db: PgPool) {
+ let server = setup(&db).await;
server
.get("/unknown")
.await
@@ -198,9 +194,9 @@ async fn errors() {
.assert_error(ErrorCode::GENERIC_METHOD_INVALID);
}
-#[tokio::test]
-async fn config() {
- let (server, _) = setup().await;
+#[db_test(sql_dir = "db", component = "taler-api")]
+async fn config(db: PgPool) {
+ let server = setup(&db).await;
server
.get("/taler-wire-gateway/config")
.await
@@ -215,9 +211,9 @@ async fn config() {
.assert_ok_json::<RevenueConfig>();
}
-#[tokio::test]
-async fn transfer() {
- let (server, _) = setup().await;
+#[db_test(sql_dir = "db", component = "taler-api")]
+async fn transfer(db: PgPool) {
+ let server = setup(&db).await;
transfer_routine(
&server.prefix("/taler-wire-gateway"),
TransferState::success,
@@ -226,9 +222,9 @@ async fn transfer() {
.await;
}
-#[tokio::test]
-async fn outgoing_history() {
- let (server, _) = &setup().await;
+#[db_test(sql_dir = "db", component = "taler-api")]
+async fn outgoing_history(db: PgPool) {
+ let server = setup(&db).await;
out_history_routine(
&server.prefix("/taler-wire-gateway"),
tasks!({
@@ -249,9 +245,9 @@ async fn outgoing_history() {
.await;
}
-#[tokio::test]
-async fn admin_add_incoming() {
- let (server, _) = setup().await;
+#[db_test(sql_dir = "db", component = "taler-api")]
+async fn admin_add_incoming(db: PgPool) {
+ let server = setup(&db).await;
admin_add_incoming_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-prepared-transfer"),
@@ -261,9 +257,9 @@ async fn admin_add_incoming() {
.await;
}
-#[tokio::test]
-async fn in_history() {
- let (server, _) = setup().await;
+#[db_test(sql_dir = "db", component = "taler-api")]
+async fn in_history(db: PgPool) {
+ let server = setup(&db).await;
in_history_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-prepared-transfer"),
@@ -275,9 +271,9 @@ async fn in_history() {
.await;
}
-#[tokio::test]
-async fn revenue() {
- let (server, _) = setup().await;
+#[db_test(sql_dir = "db", component = "taler-api")]
+async fn revenue(db: PgPool) {
+ let server = setup(&db).await;
revenue_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-revenue"),
@@ -288,9 +284,9 @@ async fn revenue() {
.await;
}
-#[tokio::test]
-async fn account_check() {
- let (server, _) = setup().await;
+#[db_test(sql_dir = "db", component = "taler-api")]
+async fn account_check(db: PgPool) {
+ let server = setup(&db).await;
server
.get("/taler-wire-gateway/account/check")
.query("account", "payto://test")
@@ -330,16 +326,16 @@ async fn check_in(pool: &PgPool) -> Vec<Status> {
.unwrap()
}
-#[tokio::test]
-async fn registration() {
- let (server, pool) = setup().await;
+#[db_test(sql_dir = "db", component = "taler-api")]
+async fn registration(db: PgPool) {
+ let server = setup(&db).await;
registration_routine(
&server.prefix("/taler-wire-gateway"),
&server.prefix("/taler-prepared-transfer"),
&EXCHANGE,
&EXCHANGE,
&UNKNOWN,
- || check_in(&pool),
+ || check_in(&db),
)
.await;
}
diff --git a/common/taler-macros/src/lib.rs b/common/taler-macros/src/lib.rs
@@ -348,3 +348,186 @@ pub fn api_config(attr: TokenStream, item: TokenStream) -> TokenStream {
TokenStream::from(expanded)
}
+
+/// Run an async database test with a pool initialized from `crate::CONFIG_SOURCE`.
+///
+/// An explicit config expression can override the default. Components without
+/// installed configuration can use `#[db_test(sql_dir = "db", component = "name")]`.
+/// Use `#[db_test(raw)]` to reserve a database without initializing schemas; its
+/// argument is `PgConnectOptions` instead of `PgPool`.
+/// Initialized tests can take a `PgPool`, a `PoolConnection<Postgres>`, a
+/// `PgConnection`, or `&mut PgConnection`. A pool and a connection may be combined
+/// in either order. The function must return `()` or `Result<(), E>`.
+/// Do not combine this attribute with `#[test]` or `#[tokio::test]`.
+#[proc_macro_attribute]
+pub fn db_test(attr: TokenStream, item: TokenStream) -> TokenStream {
+ let input = parse_macro_input!(item as syn::ItemFn);
+ match expand_db_test(attr.into(), input) {
+ Ok(tokens) => tokens.into(),
+ Err(error) => error.to_compile_error().into(),
+ }
+}
+
+fn expand_db_test(
+ attr: proc_macro2::TokenStream,
+ input: syn::ItemFn,
+) -> syn::Result<proc_macro2::TokenStream> {
+ let sig = &input.sig;
+ if sig.asyncness.is_none()
+ || !(1..=2).contains(&sig.inputs.len())
+ || !sig.generics.params.is_empty()
+ || sig.generics.where_clause.is_some()
+ || sig.constness.is_some()
+ || !matches!(sig.safety, syn::Safety::Default)
+ || sig.abi.is_some()
+ || sig.variadic.is_some()
+ || sig
+ .inputs
+ .iter()
+ .any(|arg| matches!(arg, syn::FnArg::Receiver(_)))
+ {
+ return Err(Error::new_spanned(
+ sig,
+ "db_test requires an async function with one or two database arguments and no generics",
+ ));
+ }
+ for attr in &input.attrs {
+ if attr
+ .path()
+ .segments
+ .last()
+ .is_some_and(|s| s.ident == "test")
+ {
+ return Err(Error::new_spanned(
+ attr,
+ "db_test supplies the test attribute; remove the other test attribute",
+ ));
+ }
+ }
+
+ let name = &sig.ident;
+ let callback = || -> syn::Result<proc_macro2::TokenStream> {
+ let mut has_pool = false;
+ let mut has_conn = false;
+ let mut args = Vec::new();
+ for arg in &sig.inputs {
+ let syn::FnArg::Typed(arg) = arg else {
+ unreachable!()
+ };
+ let (ty, borrowed) = match arg.ty.as_ref() {
+ syn::Type::Reference(reference) if reference.mutability.is_some() => {
+ (reference.elem.as_ref(), true)
+ }
+ ty => (ty, false),
+ };
+ let ident = match ty {
+ syn::Type::Path(path) => path.path.segments.last().map(|s| &s.ident),
+ _ => None,
+ };
+ let (slot, value) = match ident {
+ Some(ident) if ident == "PgPool" && !borrowed => {
+ (&mut has_pool, quote! { __taler_pool })
+ }
+ Some(ident) if ident == "PoolConnection" && !borrowed => {
+ (&mut has_conn, quote! { __taler_conn })
+ }
+ Some(ident) if ident == "PgConnection" => (
+ &mut has_conn,
+ if borrowed {
+ quote! { &mut *__taler_conn }
+ } else {
+ quote! { __taler_conn.detach() }
+ },
+ ),
+ _ => {
+ return Err(Error::new_spanned(
+ &arg.ty,
+ "expected PgPool, PoolConnection<Postgres>, PgConnection, or &mut PgConnection",
+ ));
+ }
+ };
+ if std::mem::replace(slot, true) {
+ return Err(Error::new_spanned(
+ &arg.ty,
+ "expected at most one pool and one connection",
+ ));
+ }
+ args.push(value);
+ }
+ let acquire = has_conn.then(|| quote! {
+ let mut __taler_conn = __taler_pool.acquire().await
+ .unwrap_or_else(|e| panic!("{}: acquire test connection: {e}", __taler_test_name));
+ });
+ Ok(quote! {
+ async |__taler_pool: ::sqlx::PgPool| {
+ #acquire
+ #name(#(#args),*).await
+ }
+ })
+ };
+ let runner = (|input: syn::parse::ParseStream<'_>| {
+ if input.peek(syn::Ident) && input.peek2(syn::Token![=]) {
+ let mut sql_dir = None;
+ let mut component = None;
+ while !input.is_empty() {
+ let key: syn::Ident = input.parse()?;
+ input.parse::<syn::Token![=]>()?;
+ let value: Expr = input.parse()?;
+ let slot = match key.to_string().as_str() {
+ "sql_dir" => &mut sql_dir,
+ "component" => &mut component,
+ _ => return Err(Error::new(key.span(), "expected sql_dir or component")),
+ };
+ if slot.replace(value).is_some() {
+ return Err(Error::new(key.span(), "duplicate db_test option"));
+ }
+ if !input.is_empty() {
+ input.parse::<syn::Token![,]>()?;
+ }
+ }
+ let sql_dir = sql_dir.ok_or_else(|| input.error("missing sql_dir"))?;
+ let component = component.ok_or_else(|| input.error("missing component"))?;
+ let callback = callback()?;
+ return Ok(quote! {
+ ::taler_test_utils::db::run_db_test_manual(
+ ::std::path::Path::new(#sql_dir), #component, __taler_test_name, #callback
+ )
+ });
+ }
+ let source: Expr = if input.is_empty() {
+ syn::parse_quote!(crate::constants::CONFIG_SOURCE)
+ } else {
+ input.parse()?
+ };
+ if matches!(&source, Expr::Path(path) if path.path.is_ident("raw")) {
+ if sig.inputs.len() != 1 {
+ return Err(Error::new_spanned(
+ sig,
+ "raw db_test requires one PgConnectOptions argument",
+ ));
+ }
+ Ok(quote! {
+ ::taler_test_utils::db::run_db_test_raw(__taler_test_name, #name)
+ })
+ } else {
+ let callback = callback()?;
+ Ok(quote! {
+ ::taler_test_utils::db::run_db_test(#source, __taler_test_name, #callback)
+ })
+ }
+ })
+ .parse2(attr)?;
+ let output = &sig.output;
+ let attrs = &input.attrs;
+ let vis = &input.vis;
+ let body = &input.block;
+ Ok(quote! {
+ #(#attrs)*
+ #[test]
+ #vis fn #name() #output {
+ #sig #body
+ let __taler_test_name = concat!(module_path!(), "::", stringify!(#name));
+ #runner
+ }
+ })
+}
diff --git a/common/taler-test-utils/src/db.rs b/common/taler-test-utils/src/db.rs
@@ -1,6 +1,6 @@
/*
This file is part of TALER
- Copyright (C) 2025, 2026 Taler Systems SA
+ Copyright (C) 2025-2026 Taler Systems SA
TALER is free software; you can redistribute it and/or modify it under the
terms of the GNU Affero General Public License as published by the Free Software
@@ -16,9 +16,7 @@
use std::{path::Path, str::FromStr as _};
-use sqlx::{
- Connection, PgConnection, PgPool, Postgres, pool::PoolConnection, postgres::PgConnectOptions,
-};
+use sqlx::{Connection, PgConnection, PgPool, postgres::PgConnectOptions};
use taler_api::config::DbCfg;
use taler_common::{
config::{Config, parser::ConfigSource},
@@ -28,76 +26,144 @@ use tracing::info;
use crate::setup_tracing;
-/// Create a reusable test database and run dbinit for a Taler component
-pub async fn db_test_setup(src: ConfigSource) -> (PoolConnection<Postgres>, PgPool) {
- let cfg = Config::load(src, None::<&str>).unwrap();
- let name = format!("{}db-postgres", src.component_name);
- let sect = cfg.section(&name);
- let db_cfg = DbCfg::parse(sect).unwrap();
- db_test_setup_manual(db_cfg.sql_dir.as_ref(), src.component_name).await
+/// An exclusive reservation of a reusable test database.
+///
+/// The administrative connection holds a session advisory lock. Dropping this
+/// reservation closes that connection and releases the slot. Keep it alive
+/// until every task and connection using the database has been dropped.
+pub struct TestDb {
+ _lock: PgConnection,
+ options: PgConnectOptions,
+}
+
+impl TestDb {
+ /// Connection options for the reserved database.
+ pub fn connect_options(&self) -> PgConnectOptions {
+ self.options.clone()
+ }
+
+ /// Reset the component schema and create a pool using its local config.
+ pub async fn setup(&self, src: ConfigSource) -> PgPool {
+ let cfg = Config::load(src, None::<&str>).unwrap();
+ let name = format!("{}db-postgres", src.component_name);
+ let db_cfg = DbCfg::parse(cfg.section(&name)).unwrap();
+ self.setup_manual(db_cfg.sql_dir.as_ref(), src.component_name)
+ .await
+ }
+
+ /// Reset the component schema using an explicit SQL directory.
+ pub async fn setup_manual(&self, sql_dir: &Path, component_name: &str) -> PgPool {
+ let pool = pool(self.connect_options(), &component_name.replace('-', "_"))
+ .await
+ .unwrap();
+ let mut conn = pool.acquire().await.unwrap();
+ dbinit(&mut conn, sql_dir, component_name, true)
+ .await
+ .unwrap();
+ drop(conn);
+ pool
+ }
+}
+
+/// Run a database test using a component's local configuration.
+///
+/// The test gets its own current-thread Tokio runtime. On return or panic the
+/// runtime drops its remaining tasks before the database reservation is released.
+/// Return values must not retain database resources; use `()` or `Result<(), E>`.
+/// Keep all work using the database on this runtime so teardown can stop it.
+pub fn run_db_test(src: ConfigSource, test_name: &str, test: impl AsyncFnOnce(PgPool)) {
+ run(test_name, async |db| test(db.setup(src).await).await)
}
-/// Create a reusable test database and run dbinit for a Taler component
-pub async fn db_test_setup_manual(
+/// Run a database test using an explicit SQL directory and component name.
+pub fn run_db_test_manual(
sql_dir: &Path,
component_name: &str,
-) -> (PoolConnection<Postgres>, PgPool) {
- setup_tracing();
- let cfg = test_db().await;
- let pool = pool(cfg, &component_name.replace("-", "_")).await.unwrap();
- let mut conn = pool.acquire().await.unwrap();
+ test_name: &str,
+ test: impl AsyncFnOnce(PgPool),
+) {
+ run(test_name, async |db| {
+ test(db.setup_manual(sql_dir, component_name).await).await
+ })
+}
- dbinit(&mut conn, sql_dir, component_name, true)
- .await
- .unwrap();
+/// Run a test with reserved connection options and no schema initialization.
+///
+/// Use this for migration tests, custom fixtures, and tests sharing a database
+/// across components. As with `run_db_test`, the reservation outlives the test
+/// runtime, and all database work must remain on that runtime.
+pub fn run_db_test_raw(test_name: &str, test: impl AsyncFnOnce(PgConnectOptions)) {
+ run(test_name, async |db| test(db.connect_options()).await)
+}
- (conn, pool)
+fn run(test_name: &str, test: impl AsyncFnOnce(&TestDb)) {
+ setup_tracing();
+ // Locals drop in reverse declaration order, including during unwinding.
+ // The reservation MUST outlive the runtime: detached tasks may own pools,
+ // checked-out connections, or listeners until runtime shutdown drops them.
+ let mut reservation = None;
+ let runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .expect("create database test runtime");
+ runtime.block_on(async {
+ reservation = Some(
+ reserve_db(test_name)
+ .await
+ .unwrap_or_else(|e| panic!("{test_name}: reserve test database: {e}")),
+ );
+ let db = reservation.as_ref().unwrap();
+ info!(target: "test", test_name, database = ?db.options.get_database(), "Reserved test database");
+ test(db).await
+ })
}
-/// Create a temporary test database that will be reuse by future tests
-pub async fn test_db() -> PgConnectOptions {
- let mut conn = PgConnection::connect("postgres:///taler_rust_check")
- .await
- .unwrap();
-
- // Find a free slot via Advisory Locks
- let row: Option<(String, bool)> = sqlx::query_as(
- "
- SELECT
- v.db_name,
- EXISTS (
- SELECT 1
- FROM pg_catalog.pg_database
- WHERE datname = v.db_name
- ) AS already_exists
- FROM generate_series(0, 1000) AS id
- CROSS JOIN LATERAL (
- SELECT CASE
- WHEN id = 0 THEN 'taler_rust_test'
- ELSE 'taler_rust_test_' || id
- END AS db_name
- ) AS v
- WHERE pg_try_advisory_lock(id)
- LIMIT 1;
- ",
+/// Reserve a reusable test database through a session advisory lock.
+///
+/// Connection defaults are `postgres:///taler_rust_check`. PostgreSQL's usual
+/// `PGHOST`, `PGPORT`, and `PGUSER` settings can select another test server.
+pub async fn test_db() -> sqlx::Result<TestDb> {
+ reserve_db("taler-test-utils").await
+}
+
+async fn reserve_db(test_name: &str) -> sqlx::Result<TestDb> {
+ let admin =
+ PgConnectOptions::from_str("postgres:///taler_rust_check")?.application_name(test_name);
+ let mut conn = PgConnection::connect_with(&admin).await?;
+
+ // Find a free slot via Advisory Locks.
+ let slot: Option<i32> = sqlx::query_scalar(
+ "SELECT id FROM generate_series(0, 1000) AS id
+ WHERE pg_try_advisory_lock(id) LIMIT 1",
)
.fetch_optional(&mut conn)
- .await
- .unwrap();
+ .await?;
- let Some((name, exists)) = row else {
- panic!("Could not find a free database slot after 1000 attempts.")
+ let Some(slot) = slot else {
+ return Err(sqlx::Error::Protocol(
+ "Could not find a free database slot after 1001 attempts.".into(),
+ ));
+ };
+ let name = match slot {
+ 0 => "taler_rust_test".to_owned(),
+ id => format!("taler_rust_test_{id}"),
};
+ // Check in a fresh statement after acquiring the lock. A previous holder
+ // may have created this database after the allocation query's snapshot.
+ let exists: bool = sqlx::query_scalar(
+ "SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_database WHERE datname = $1)",
+ )
+ .bind(&name)
+ .fetch_one(&mut conn)
+ .await?;
if !exists {
sqlx::raw_sql(&format!("CREATE DATABASE {name}"))
.execute(&mut conn)
- .await
- .unwrap();
+ .await?;
}
- // We need this connection to stay open to keep the advisory lock in place
- // Leaking it is OK in tests
- std::mem::forget(conn);
- let db_url = format!("postgresql:///{name}");
- info!(target: "test", "Running on db {db_url}");
- PgConnectOptions::from_str(&db_url).unwrap()
+ let options = admin.database(&name);
+ Ok(TestDb {
+ _lock: conn,
+ options,
+ })
}
diff --git a/taler-apns-relay/src/api.rs b/taler-apns-relay/src/api.rs
@@ -98,17 +98,18 @@ pub fn router(state: Arc<RelayApi>) -> Router {
mod test {
use std::sync::Arc;
+ use sqlx::PgPool;
use taler_api::api::TalerRouter as _;
+ use taler_macros::db_test;
use taler_test_utils::{json, server::TestServer as _};
use crate::{
api::{ApnsRelayConfig, RelayApi, router},
- db::{all_registrations, test::setup},
+ db::all_registrations,
};
- #[tokio::test]
- async fn api() {
- let pool = setup().await;
+ #[db_test]
+ async fn api(pool: PgPool) {
let api = Arc::new(RelayApi::new(pool.clone()));
let server = router(api).finalize();
diff --git a/taler-apns-relay/src/db.rs b/taler-apns-relay/src/db.rs
@@ -85,20 +85,12 @@ pub async fn clear_registration(db: &PgPool) -> sqlx::Result<()> {
pub mod test {
use jiff::{SignedDuration, Timestamp};
use sqlx::PgPool;
- use taler_test_utils::db::db_test_setup;
+ use taler_macros::db_test;
- use crate::{
- constants::CONFIG_SOURCE,
- db::{all_registrations, clear_registration, register, unregister},
- };
+ use crate::db::{all_registrations, clear_registration, register, unregister};
- pub async fn setup() -> PgPool {
- db_test_setup(CONFIG_SOURCE).await.1
- }
-
- #[tokio::test]
- async fn registration() {
- let db = setup().await;
+ #[db_test]
+ async fn registration(db: PgPool) {
let token1 = "device_token1";
let token2 = "device_token2";
let now = Timestamp::now();
diff --git a/taler-apns-relay/src/lib.rs b/taler-apns-relay/src/lib.rs
@@ -30,6 +30,7 @@ pub mod config;
pub mod constants;
pub mod db;
pub mod worker;
+pub use crate::constants::CONFIG_SOURCE;
pub async fn setup(cfg: &Config, pool: &PgPool, reset: bool) -> anyhow::Result<()> {
let apns_cfg = ApnsConfig::parse(cfg)?;
