depolymerization

utils.rs (15559B)

---

 /*
   This file is part of TALER
   Copyright (C) 2022-2025 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 std::{
     fmt::Display,
     io::Write as _,
     net::{Ipv4Addr, SocketAddrV4, TcpListener},
     ops::{Deref, DerefMut},
     path::{Path, PathBuf},
     process::{Child, Command, Stdio},
     str::FromStr,
     sync::Arc,
     time::Duration,
 };
 
 use indicatif::ProgressBar;
 use ini::Ini;
 use taler_common::{
     api_common::{EddsaPublicKey, ShortHashCode},
     api_wire::{IncomingBankTransaction, IncomingHistory, OutgoingHistory, TransferRequest},
     types::{amount::Amount, base32::Base32, payto::PaytoURI},
 };
 use url::Url;
 
 const LOG: &str = "INFO";
 
 #[must_use]
 pub async fn check_incoming(base_url: &str, txs: &[(EddsaPublicKey, Amount)]) -> bool {
     let mut res = ureq::get(&format!("{base_url}history/incoming"))
         .query("delta", format!("-{}", 100))
         .call()
         .unwrap();
     if txs.is_empty() {
         res.status() == 204
     } else {
         if res.status() != 200 {
             return false;
         }
         let history: IncomingHistory = res.body_mut().read_json().unwrap();
 
         history.incoming_transactions.len() == txs.len()
             && txs.iter().all(|(reserve_pub_key, taler_amount)| {
                 history.incoming_transactions.iter().any(|h| {
                     matches!(
                         h,
                         IncomingBankTransaction::Reserve {
                             reserve_pub,
                             amount,
                             ..
                         } if reserve_pub == reserve_pub_key && amount == taler_amount
                     )
                 })
             })
     }
 }
 
 #[must_use]
 pub async fn check_gateway_down(base_url: &str) -> bool {
     matches!(
         ureq::get(&format!("{base_url}history/incoming"))
             .query("delta", "-5")
             .call(),
         Err(ureq::Error::StatusCode(504 | 502))
     )
 }
 
 #[must_use]
 pub async fn check_gateway_up(base_url: &str) -> bool {
     ureq::get(&format!("{base_url}config")).call().is_ok()
 }
 
 pub async fn transfer(base_url: &str, wtid: &[u8; 32], credit_account: PaytoURI, amount: &Amount) {
     loop {
         let res = ureq::post(&format!("{base_url}transfer")).send_json(TransferRequest {
             request_uid: Base32::rand(),
             amount: amount.clone(),
             exchange_base_url: Url::parse("https://exchange.test/").unwrap(),
             wtid: Base32::from(*wtid),
             credit_account: credit_account.clone(),
         });
         if !matches!(res, Err(ureq::Error::StatusCode(502))) {
             res.unwrap();
             break;
         }
     }
 }
 
 #[must_use]
 pub async fn check_outgoing(base_url: &str, txs: &[(ShortHashCode, Amount)]) -> bool {
     let mut res = ureq::get(format!("{base_url}history/outgoing"))
         .query("delta", format!("-{}", txs.len()))
         .call()
         .unwrap();
     if txs.is_empty() {
         res.status() == 204
     } else {
         if res.status() != 200 {
             return false;
         }
         let history: OutgoingHistory = res.body_mut().read_json().unwrap();
 
         history.outgoing_transactions.len() == txs.len()
             && txs.iter().all(|(wtid, amount)| {
                 history
                     .outgoing_transactions
                     .iter()
                     .any(|h| h.wtid == *wtid && &h.amount == amount)
             })
     }
 }
 pub struct ChildGuard(pub Child);
 
 impl Drop for ChildGuard {
     fn drop(&mut self) {
         self.0.kill().ok();
     }
 }
 
 #[track_caller]
 pub fn try_cmd_redirect(
     cmd: &str,
     args: &[&str],
     path: impl AsRef<Path>,
 ) -> std::io::Result<ChildGuard> {
     let log_file = std::fs::OpenOptions::new()
         .create(true)
         .append(true)
         .open(path)?;
 
     let child = Command::new(cmd)
         .args(args)
         .stderr(log_file.try_clone()?)
         .stdout(log_file)
         .stdin(Stdio::null())
         .spawn()?;
     Ok(ChildGuard(child))
 }
 
 #[track_caller]
 pub fn cmd_redirect(cmd: &str, args: &[&str], path: impl AsRef<Path>) -> ChildGuard {
     try_cmd_redirect(cmd, args, path).unwrap()
 }
 
 #[track_caller]
 pub fn cmd_ok(mut child: ChildGuard, name: &str) {
     let result = child.0.wait().unwrap();
     if !result.success() {
         panic!("cmd {name} failed");
     }
 }
 
 #[track_caller]
 pub fn cmd_redirect_ok(cmd: &str, args: &[&str], path: impl AsRef<Path>, name: &str) {
     cmd_ok(cmd_redirect(cmd, args, path), name)
 }
 
 #[macro_export]
 macro_rules! retry_opt {
     ($expr:expr) => {
         async {
             let start = std::time::Instant::now();
             loop {
                 let result = $expr;
                 if result.is_err() && start.elapsed() < std::time::Duration::from_secs(30) {
                     tokio::time::sleep(std::time::Duration::from_millis(500)).await;
                 } else {
                     return result.unwrap();
                 }
             }
         }
         .await
     };
 }
 
 #[macro_export]
 macro_rules! retry {
     ($expr:expr) => {
         $crate::retry_opt! {
             $expr.then_some(()).ok_or("failure")
         }
     };
 }
 #[derive(Clone)]
 pub struct TestCtx {
     pub name: String,
     pub root: PathBuf,
     pub dir: PathBuf,
     pub pb: ProgressBar,
     pub db: Arc<LocalDb>,
 }
 
 impl TestCtx {
     pub fn new(
         root: impl Into<PathBuf>,
         name: impl Into<String>,
         pb: ProgressBar,
         db: Arc<LocalDb>,
     ) -> Self {
         let root = root.into();
         let name = name.into();
         // Create log dir
         let dir = root.join(&name);
         std::fs::create_dir_all(&dir).unwrap();
 
         Self {
             name,
             dir,
             pb,
             db,
             root,
         }
     }
 
     pub fn log(&self, name: &str) -> PathBuf {
         self.dir.join(format!("{name}.log"))
     }
 
     pub fn step(&self, disp: impl Display) {
         self.pb.set_message(format!("{disp}"))
     }
 
     /* ----- Database ----- */
 
     pub fn stop_db(&mut self) {
         self.db.stop_db(&self.name);
     }
 
     pub fn resume_db(&mut self) {
         self.db.resume_db(&self.name);
     }
 }
 
 pub struct TalerCtx {
     pub wire_dir: PathBuf,
     pub wire2_dir: PathBuf,
     pub conf: PathBuf,
     ctx: TestCtx,
     pub wire_bin_path: String,
     stressed: bool,
     gateway: Option<ChildGuard>,
     pub gateway_url: String,
     wire: Option<ChildGuard>,
     wire2: Option<ChildGuard>,
     pub gateway_port: u16,
 }
 
 impl TalerCtx {
     pub fn new(ctx: &TestCtx, wire_name: impl Into<String>, config: &str, stressed: bool) -> Self {
         // Create temporary dir
         let dir = ctx.dir.clone();
         let conf = dir.join("taler.conf");
 
         // Create common dirs
         let wire_dir = dir.join("wire");
         let wire2_dir = dir.join("wire2");
         for dir in [&wire_dir, &wire2_dir] {
             std::fs::create_dir_all(dir).unwrap();
         }
 
         // Find unused port
         let gateway_port = unused_port();
         let gateway_url = format!("http://localhost:{gateway_port}/taler-wire-gateway/");
 
         // Generate taler config from base
         let wire_name = wire_name.into();
         let config = PathBuf::from_str("testbench/conf").unwrap().join(config);
         let mut cfg = ini::Ini::load_from_file(config).unwrap();
         cfg.with_section(Some("exchange-accountcredentials-admin"))
             .set("WIRE_GATEWAY_URL", &gateway_url);
         cfg.with_section(Some(format!("{wire_name}db-postgres")))
             .set("CONFIG", ctx.db.postgres_uri(&ctx.name));
         cfg.with_section(Some(format!("{wire_name}-httpd")))
             .set("PORT", gateway_port.to_string());
 
         cfg.write_to_file(&conf).unwrap();
 
         Self {
             ctx: ctx.clone(),
             gateway_url,
             wire_dir,
             wire2_dir,
             conf,
             wire_bin_path: if stressed {
                 ctx.root.join(format!("bin/{wire_name}-fail"))
             } else {
                 ctx.root.join(format!("bin/{wire_name}"))
             }
             .to_string_lossy()
             .to_string(),
             stressed,
             gateway: None,
             wire: None,
             wire2: None,
             gateway_port,
         }
     }
 
     pub fn dbinit(&self) {
         self.db.wait_running();
         self.db.create_db(&self.ctx.name);
         // Init db
         cmd_redirect_ok(
             &self.wire_bin_path,
             &["-c", self.conf.to_string_lossy().as_ref(), "dbinit"],
             self.log("cmd"),
             "wire dbinit",
         );
     }
 
     pub fn reset_db(&self) {
         // Reset db
         cmd_redirect_ok(
             &self.wire_bin_path,
             &["-c", self.conf.to_string_lossy().as_ref(), "dbinit", "-r"],
             self.log("cmd"),
             "wire dbinit reset",
         );
     }
 
     pub fn setup(&self) {
         // Init db
         cmd_redirect_ok(
             &self.wire_bin_path,
             &["-c", self.conf.to_string_lossy().as_ref(), "setup"],
             self.log("cmd"),
             "wire setup",
         );
     }
 
     pub async fn run(&mut self) {
         // Run gateway
         self.gateway = Some(cmd_redirect(
             &self.wire_bin_path,
             &[
                 "-c",
                 self.conf.to_string_lossy().as_ref(),
                 "-L",
                 LOG,
                 "serve",
             ],
             self.log("gateway"),
         ));
 
         // Start wires
         self.wire = Some(cmd_redirect(
             &self.wire_bin_path,
             &[
                 "-c",
                 self.conf.to_string_lossy().as_ref(),
                 "-L",
                 LOG,
                 "worker",
             ],
             self.log("worker"),
         ));
         self.wire2 = self.stressed.then(|| {
             cmd_redirect(
                 &self.wire_bin_path,
                 &[
                     "-c",
                     self.conf.to_string_lossy().as_ref(),
                     "-L",
                     LOG,
                     "worker",
                 ],
                 self.log("worker+"),
             )
         });
 
         // Wait for gateway to be up
         retry_opt! {
             tokio::net::TcpStream::connect(SocketAddrV4::new(Ipv4Addr::LOCALHOST, self.gateway_port)).await
         };
     }
 
     /* ----- Process ----- */
 
     #[must_use]
     pub fn wire_running(&mut self) -> bool {
         self.wire.as_mut().unwrap().0.try_wait().unwrap().is_none()
     }
 
     #[must_use]
     pub fn gateway_running(&mut self) -> bool {
         self.gateway
             .as_mut()
             .unwrap()
             .0
             .try_wait()
             .unwrap()
             .is_none()
     }
 
     /* ----- Wire Gateway -----*/
 
     pub async fn expect_credits(&self, txs: &[(EddsaPublicKey, Amount)]) {
         retry! { check_incoming(&self.gateway_url, txs).await }
     }
 
     pub async fn expect_debits(&self, txs: &[(ShortHashCode, Amount)]) {
         retry! { check_outgoing(&self.gateway_url, txs).await }
     }
 
     pub async fn expect_gateway_up(&self) {
         retry! { check_gateway_up(&self.gateway_url).await }
     }
 
     pub async fn expect_gateway_down(&self) {
         retry! { check_gateway_down(&self.gateway_url).await }
     }
 }
 
 impl Deref for TalerCtx {
     type Target = TestCtx;
 
     fn deref(&self) -> &Self::Target {
         &self.ctx
     }
 }
 
 impl DerefMut for TalerCtx {
     fn deref_mut(&mut self) -> &mut Self::Target {
         &mut self.ctx
     }
 }
 
 pub fn unused_port() -> u16 {
     TcpListener::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0))
         .unwrap()
         .local_addr()
         .unwrap()
         .port()
 }
 
 pub struct LocalDb {
     cluster_dir: String,
     cmd_log: String,
     cluster: ChildGuard,
 }
 
 impl LocalDb {
     pub fn new(root: &Path) -> Self {
         let cluster_dir = root.join("db");
         let cluster_log = root.join("postgres.log").to_string_lossy().to_string();
         let cmd_log = root.join("postgres-cmd.log").to_string_lossy().to_string();
         if !std::fs::exists(&cluster_dir).unwrap() {
             // Init databases files
             cmd_redirect_ok(
                 "initdb",
                 &[cluster_dir.to_string_lossy().as_ref()],
                 &cluster_log,
                 "init_db",
             );
         }
         // Generate database config
         std::fs::write(
             cluster_dir.join("postgresql.conf"),
             format!(
                 "
                 listen_addresses=''
                 unix_socket_directories='{}'
                 fsync=off
                 synchronous_commit=off
                 full_page_writes=off
                 ",
                 cluster_dir.to_string_lossy().as_ref()
             ),
         )
         .unwrap();
         let cluster = cmd_redirect(
             "postgres",
             &["-D", cluster_dir.to_string_lossy().as_ref()],
             &cmd_log,
         );
         Self {
             cluster_dir: cluster_dir.to_string_lossy().to_string(),
             cmd_log,
             cluster,
         }
     }
 
     pub fn postgres_uri(&self, database: &str) -> String {
         format!("postgres:///{database}?host={}", self.cluster_dir)
     }
 
     pub fn execute_sql(&self, sql: &str) -> bool {
         let mut psql = ChildGuard(
             Command::new("psql")
                 .arg(self.postgres_uri("postgres"))
                 .stderr(Stdio::null())
                 .stdout(
                     std::fs::File::options()
                         .append(true)
                         .create(true)
                         .open(&self.cmd_log)
                         .unwrap(),
                 )
                 .stdin(Stdio::piped())
                 .spawn()
                 .unwrap(),
         );
         psql.0
             .stdin
             .as_mut()
             .unwrap()
             .write_all(sql.as_bytes())
             .unwrap();
         psql.0.wait().unwrap().success()
     }
 
     pub fn create_db(&self, name: &str) {
         self.execute_sql(&format!("CREATE DATABASE {name};"));
     }
 
     pub fn stop_db(&self, name: &str) {
         self.execute_sql(&format!(
             "
             UPDATE pg_database SET datallowconn=false WHERE datname='{name}';
             SELECT pg_terminate_backend(pid)
             FROM pg_stat_activity
             WHERE datname='{name}' AND pid <> pg_backend_pid();
             "
         ));
     }
     pub fn resume_db(&self, name: &str) {
         self.execute_sql(&format!(
             "UPDATE pg_database SET datallowconn=true WHERE datname='{name}';"
         ));
     }
 
     pub fn wait_running(&self) {
         for _ in 0..10 {
             if self.execute_sql("SELECT true") {
                 break;
             }
             std::thread::sleep(Duration::from_millis(500))
         }
     }
 }
 
 pub fn patch_config(from: impl AsRef<Path>, to: impl AsRef<Path>, patch: impl FnOnce(&mut Ini)) {
     let mut cfg = ini::Ini::load_from_file(from).unwrap();
     patch(&mut cfg);
     cfg.write_to_file(to).unwrap();
 }