taler-rust

GNU Taler code in Rust. Largely core banking integrations.
Log | Files | Refs | Submodules | README | LICENSE

cyclos-harness.rs (19352B)


      1 /*
      2   This file is part of TALER
      3   Copyright (C) 2025, 2026 Taler Systems SA
      4 
      5   TALER is free software; you can redistribute it and/or modify it under the
      6   terms of the GNU Affero General Public License as published by the Free Software
      7   Foundation; either version 3, or (at your option) any later version.
      8 
      9   TALER is distributed in the hope that it will be useful, but WITHOUT ANY
     10   WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
     11   A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more details.
     12 
     13   You should have received a copy of the GNU Affero General Public License along with
     14   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
     15 */
     16 
     17 use std::time::Duration;
     18 
     19 use clap::Parser as _;
     20 use compact_str::CompactString;
     21 use failure_injection::{InjectedErr, set_failure_scenario};
     22 use jiff::Timestamp;
     23 use owo_colors::OwoColorize as _;
     24 use sqlx::{PgPool, Row as _, postgres::PgRow};
     25 use taler_api::notification::dummy_listen;
     26 use taler_build::long_version;
     27 use taler_common::{
     28     CommonArgs,
     29     api::{
     30         EddsaPublicKey, HashCode, ShortHashCode,
     31         params::{History, Page, Pooling},
     32         wire::{IncomingBankTransaction, TransferState},
     33     },
     34     config::Config,
     35     taler_main,
     36     types::{
     37         amount::{Currency, Decimal, decimal},
     38         url,
     39     },
     40 };
     41 use taler_cyclos::{
     42     config::{AccountType, HarnessCfg},
     43     constants::CONFIG_SOURCE,
     44     cyclos_api::{
     45         api::CyclosAuth,
     46         client::Client,
     47         types::{HistoryItem, OrderBy},
     48     },
     49     db::{self, TransferResult, dbinit},
     50     payto::FullCyclosPayto,
     51     setup,
     52     worker::{Worker, WorkerError, WorkerResult, run_worker},
     53 };
     54 
     55 /// Cyclos Adapter harness test suite
     56 #[derive(clap::Parser, Debug)]
     57 #[command(long_version = long_version(), about, long_about = None)]
     58 struct Args {
     59     #[clap(flatten)]
     60     common: CommonArgs,
     61 
     62     #[command(subcommand)]
     63     cmd: Command,
     64 }
     65 
     66 #[derive(clap::Subcommand, Debug)]
     67 enum Command {
     68     /// Run logic tests
     69     Logic {
     70         #[arg(short, long)]
     71         reset: bool,
     72     },
     73     /// Run online tests
     74     Online {
     75         #[arg(short, long)]
     76         reset: bool,
     77     },
     78 }
     79 
     80 fn step(step: &str) {
     81     println!("{}", step.green());
     82 }
     83 
     84 struct Harness<'a> {
     85     pool: &'a PgPool,
     86     client: Client<'a>,
     87     wire: Client<'a>,
     88     client_payto: FullCyclosPayto,
     89     wire_payto: FullCyclosPayto,
     90     payment_type_id: i64,
     91     account_type_id: i64,
     92     currency: Currency,
     93     root: CompactString,
     94 }
     95 
     96 impl<'a> Harness<'a> {
     97     async fn balance(&self) -> (Decimal, Decimal) {
     98         let (exchange, client) =
     99             tokio::try_join!(self.wire.accounts(), self.client.accounts()).unwrap();
    100         (
    101             exchange[0]
    102                 .status
    103                 .available_balance
    104                 .unwrap_or(exchange[0].status.balance),
    105             client[0]
    106                 .status
    107                 .available_balance
    108                 .unwrap_or(client[0].status.balance),
    109         )
    110     }
    111 
    112     /// Send transaction from client to exchange
    113     async fn client_send(&self, subject: &str, amount: Decimal) -> i64 {
    114         *self
    115             .client
    116             .direct_payment(*self.wire_payto.id, self.payment_type_id, amount, subject)
    117             .await
    118             .unwrap()
    119             .id
    120     }
    121 
    122     /// Send transaction from exchange to client
    123     async fn exchange_send(&self, subject: &str, amount: Decimal) -> i64 {
    124         *self
    125             .wire
    126             .direct_payment(*self.client_payto.id, self.payment_type_id, amount, subject)
    127             .await
    128             .unwrap()
    129             .id
    130     }
    131 
    132     /// Chargeback a transfer
    133     async fn chargeback(&self, id: i64) -> i64 {
    134         self.client.chargeback(id).await.unwrap()
    135     }
    136 
    137     /// Fetch last transfer related to client
    138     async fn client_last_transfer(&self) -> HistoryItem {
    139         self.client
    140             .history(*self.client_payto.id, OrderBy::DateDesc, 0, None)
    141             .await
    142             .unwrap()
    143             .page
    144             .remove(0)
    145     }
    146 
    147     /// Run the worker once
    148     async fn worker(&'a self) -> WorkerResult {
    149         let db = &mut self.pool.acquire().await.unwrap().detach();
    150         Worker {
    151             db,
    152             currency: self.currency,
    153             client: &self.wire,
    154             account_type: AccountType::Exchange,
    155             account_type_id: self.account_type_id,
    156             payment_type_id: self.payment_type_id,
    157         }
    158         .run()
    159         .await
    160     }
    161 
    162     async fn expect_incoming(&self, key: EddsaPublicKey) {
    163         let transfer = db::incoming_history(
    164             self.pool,
    165             &History {
    166                 page: Page {
    167                     limit: -1,
    168                     offset: None,
    169                 },
    170                 pooling: Pooling { timeout_ms: None },
    171             },
    172             &self.currency,
    173             &self.root,
    174             dummy_listen,
    175         )
    176         .await
    177         .unwrap();
    178         // TODO revert to assert_matches! once rust-lang#82775 is stable
    179         assert!(matches!(
    180             transfer.first().unwrap(),
    181             IncomingBankTransaction::Reserve { reserve_pub, .. } if *reserve_pub == key,
    182         ));
    183     }
    184 
    185     async fn custom_transfer(&self, amount: Decimal, creditor_id: i64, creditor_name: &str) -> u64 {
    186         let res = db::make_transfer(
    187             self.pool,
    188             &db::Transfer {
    189                 request_uid: HashCode::rand(),
    190                 amount,
    191                 exchange_base_url: url("https://test.com"),
    192                 metadata: None,
    193                 wtid: ShortHashCode::rand(),
    194                 creditor_id,
    195                 creditor_name: CompactString::new(creditor_name),
    196             },
    197             &Timestamp::now(),
    198         )
    199         .await
    200         .unwrap();
    201         match res {
    202             TransferResult::Success { id, .. } => id,
    203             TransferResult::RequestUidReuse | TransferResult::WtidReuse => unreachable!(),
    204         }
    205     }
    206 
    207     async fn transfer(&self, amount: Decimal) -> u64 {
    208         self.custom_transfer(amount, *self.client_payto.id, &self.client_payto.name)
    209             .await
    210     }
    211 
    212     async fn transfer_id(&self, transfer_id: u64) -> i64 {
    213         sqlx::query(
    214             "SELECT transfer_id
    215                 FROM transfer
    216                     JOIN initiated USING (initiated_id)
    217                     JOIN tx_out USING (tx_out_id)
    218                 WHERE initiated_id=$1",
    219         )
    220         .bind(transfer_id as i64)
    221         .try_map(|r: PgRow| r.try_get(0))
    222         .fetch_one(self.pool)
    223         .await
    224         .unwrap()
    225     }
    226 
    227     async fn expect_transfer_status(&self, id: u64, status: TransferState, msg: Option<&str>) {
    228         let mut attempts = 0;
    229         loop {
    230             let transfer = db::transfer_by_id(self.pool, id, &self.currency, &self.root)
    231                 .await
    232                 .unwrap()
    233                 .unwrap();
    234             if (transfer.status, transfer.status_msg.as_deref()) == (status, msg) {
    235                 return;
    236             }
    237             if attempts > 40 {
    238                 assert_eq!(
    239                     (transfer.status, transfer.status_msg.as_deref()),
    240                     (status, msg)
    241                 );
    242             }
    243             attempts += 1;
    244             tokio::time::sleep(Duration::from_millis(200)).await;
    245         }
    246     }
    247 }
    248 
    249 struct Balances<'a> {
    250     client: &'a Harness<'a>,
    251     exchange_balance: Decimal,
    252     client_balance: Decimal,
    253 }
    254 
    255 impl<'a> Balances<'a> {
    256     pub async fn new(client: &'a Harness<'a>) -> Self {
    257         let (exchange_balance, client_balance) = client.balance().await;
    258         Self {
    259             client,
    260             exchange_balance,
    261             client_balance,
    262         }
    263     }
    264 
    265     async fn expect_add(&mut self, diff: Decimal) {
    266         self.exchange_balance = self.exchange_balance.try_add(&diff).unwrap();
    267         self.client_balance = self.client_balance.try_sub(&diff).unwrap();
    268         let mut attempts = 0;
    269         loop {
    270             let current = self.client.balance().await;
    271             if current == (self.exchange_balance, self.client_balance) {
    272                 return;
    273             }
    274             if attempts > 40 {
    275                 assert_eq!(
    276                     current,
    277                     (self.exchange_balance, self.client_balance),
    278                     "({} {}) +{diff}",
    279                     current.0,
    280                     current.1
    281                 );
    282             }
    283             attempts += 1;
    284             tokio::time::sleep(Duration::from_millis(200)).await;
    285         }
    286     }
    287 
    288     async fn expect_sub(&mut self, diff: Decimal) {
    289         self.exchange_balance = self.exchange_balance.try_sub(&diff).unwrap();
    290         self.client_balance = self.client_balance.try_add(&diff).unwrap();
    291 
    292         let mut attempts = 0;
    293         loop {
    294             let current = self.client.balance().await;
    295             if current == (self.exchange_balance, self.client_balance) {
    296                 return;
    297             }
    298             if attempts > 40 {
    299                 assert_eq!(
    300                     current,
    301                     (self.exchange_balance, self.client_balance),
    302                     "({} {}) -{diff}",
    303                     current.0,
    304                     current.1
    305                 );
    306             }
    307             attempts += 1;
    308             tokio::time::sleep(Duration::from_millis(200)).await;
    309         }
    310     }
    311 }
    312 
    313 /// Run logic tests against local Cyclos backend
    314 async fn logic_harness(cfg: &Config, reset: bool) -> anyhow::Result<()> {
    315     step("Run Cyclos logic harness tests");
    316 
    317     step("Prepare db");
    318     let pool = dbinit(cfg, reset).await?;
    319 
    320     let client = http_client::client();
    321     setup::setup(cfg, reset, &client).await?;
    322     let cfg = HarnessCfg::parse(cfg)?;
    323     let wire = Client {
    324         client: &client,
    325         api_url: &cfg.worker.host.api_url,
    326         auth: &CyclosAuth::Basic {
    327             username: cfg.worker.host.username,
    328             password: cfg.worker.host.password,
    329         },
    330     };
    331     let client = Client {
    332         client: &client,
    333         api_url: &cfg.worker.host.api_url,
    334         auth: &CyclosAuth::Basic {
    335             username: cfg.username,
    336             password: cfg.password,
    337         },
    338     };
    339     let harness = Harness {
    340         pool: &pool,
    341         client_payto: client
    342             .whoami()
    343             .await
    344             .unwrap()
    345             .payto(cfg.worker.root.clone()),
    346         wire_payto: wire.whoami().await.unwrap().payto(cfg.worker.root.clone()),
    347         client,
    348         wire,
    349         currency: cfg.worker.currency,
    350         root: cfg.worker.root,
    351         payment_type_id: *cfg.worker.payment_type_id,
    352         account_type_id: *cfg.worker.account_type_id,
    353     };
    354 
    355     step("Warmup");
    356     harness.worker().await.unwrap();
    357     let now = Timestamp::now();
    358     let balance = &mut Balances::new(&harness).await;
    359 
    360     step("Test incoming talerable transaction");
    361     // Send talerable transaction
    362     let reserve_pub = EddsaPublicKey::rand();
    363     let amount = decimal("3.3");
    364     harness
    365         .client_send(&format!("Taler {reserve_pub}"), amount)
    366         .await;
    367     // Sync and register
    368     harness.worker().await?;
    369     harness.expect_incoming(reserve_pub).await;
    370     balance.expect_add(amount).await;
    371 
    372     step("Test incoming malformed transaction");
    373     // Send malformed transaction
    374     let amount = decimal("3.4");
    375     harness
    376         .client_send(&format!("Malformed test {now}"), amount)
    377         .await;
    378     balance.expect_add(amount).await;
    379     // Sync and bounce
    380     harness.worker().await?;
    381     balance.expect_sub(amount).await;
    382 
    383     step("Test transfer transactions");
    384     let amount = decimal("3.5");
    385     // Init a transfer to client
    386     let transfer_id = harness.transfer(amount).await;
    387     // Check transfer pending
    388     harness
    389         .expect_transfer_status(transfer_id, TransferState::pending, None)
    390         .await;
    391     // Should send
    392     harness.worker().await?;
    393     // Wait for transaction to finalize
    394     balance.expect_sub(amount).await;
    395     // Should register
    396     harness.worker().await?;
    397     // Check transfer is now successful
    398     harness
    399         .expect_transfer_status(transfer_id, TransferState::success, None)
    400         .await;
    401 
    402     step("Test transfer to self");
    403     // Init a transfer to self
    404     let transfer_id = harness
    405         .custom_transfer(
    406             decimal("10.1"),
    407             *harness.wire_payto.id,
    408             &harness.wire_payto.name,
    409         )
    410         .await;
    411     // Should failed
    412     harness.worker().await?;
    413     // Check transfer failed
    414     harness
    415         .expect_transfer_status(
    416             transfer_id,
    417             TransferState::permanent_failure,
    418             Some("permissionDenied - The operation was denied because a required permission was not granted"),
    419         )
    420         .await;
    421 
    422     step("Test transfer to unknown account");
    423     // Init a transfer to unknown
    424     let transfer_id = harness
    425         .custom_transfer(decimal("10.1"), 42, "Unknown")
    426         .await;
    427     // Should failed
    428     harness.worker().await?;
    429     // Check transfer failed
    430     harness
    431         .expect_transfer_status(
    432             transfer_id,
    433             TransferState::permanent_failure,
    434             Some("unknown BasicUser 42"),
    435         )
    436         .await;
    437 
    438     step("Test unexpected outgoing");
    439     // Manual tx from the exchange
    440     let amount = decimal("4");
    441     harness
    442         .exchange_send(&format!("What is this ? {now}"), amount)
    443         .await;
    444     harness.worker().await?;
    445     // Wait for transaction to finalize
    446     balance.expect_sub(amount).await;
    447     harness.worker().await?;
    448 
    449     step("Test transfer chargeback");
    450     let amount = decimal("10.1");
    451     // Init a transfer to client
    452     let transfer_id = harness.transfer(amount).await;
    453     harness
    454         .expect_transfer_status(transfer_id, TransferState::pending, None)
    455         .await;
    456     // Send
    457     harness.worker().await?;
    458     balance.expect_sub(amount).await;
    459     harness
    460         .expect_transfer_status(transfer_id, TransferState::pending, None)
    461         .await;
    462     // Sync
    463     harness.worker().await?;
    464     harness
    465         .expect_transfer_status(transfer_id, TransferState::success, None)
    466         .await;
    467     // Chargeback
    468     harness
    469         .chargeback(harness.transfer_id(transfer_id).await)
    470         .await;
    471     balance.expect_add(amount).await;
    472     harness.worker().await?;
    473     harness
    474         .expect_transfer_status(
    475             transfer_id,
    476             TransferState::late_failure,
    477             Some("charged back"),
    478         )
    479         .await;
    480 
    481     step("Test recover unexpected chargeback");
    482     let amount = decimal("10.2");
    483     // Manual tx from the exchange
    484     harness
    485         .exchange_send(&format!("What is this chargebacked ? {now}"), amount)
    486         .await;
    487     balance.expect_sub(amount).await;
    488     // Chargeback
    489     harness
    490         .chargeback(*harness.client_last_transfer().await.id)
    491         .await;
    492     balance.expect_add(amount).await;
    493     // Sync
    494     harness.worker().await?;
    495 
    496     step("Test direct-payment failure");
    497     let amount = decimal("10.3");
    498     harness.transfer(amount).await;
    499     set_failure_scenario(&["direct-payment"]);
    500     // TODO revert to assert_matches! once rust-lang#82775 is stable
    501     assert!(matches!(
    502         harness.worker().await.unwrap_err(),
    503         WorkerError::Injected(InjectedErr("direct-payment"))
    504     ));
    505     harness.worker().await?;
    506     balance.expect_sub(amount).await;
    507     harness.worker().await?;
    508 
    509     step("Test chargeback failure");
    510     // Send malformed transaction
    511     let amount = decimal("10.4");
    512     harness
    513         .client_send(&format!("Malformed test {now} with failure"), amount)
    514         .await;
    515     balance.expect_add(amount).await;
    516     // Sync and bounce
    517     set_failure_scenario(&["chargeback"]);
    518     // TODO revert to assert_matches! once rust-lang#82775 is stable
    519     assert!(matches!(
    520         harness.worker().await.unwrap_err(),
    521         WorkerError::Injected(InjectedErr("chargeback"))
    522     ));
    523     balance.expect_sub(amount).await;
    524     // Sync recover
    525     harness.worker().await?;
    526 
    527     step("Finish");
    528     harness.worker().await?;
    529     balance.expect_add(Decimal::ZERO).await;
    530     Ok(())
    531 }
    532 
    533 /// Run online tests against real Cyclos backend
    534 async fn online_harness(config: &Config, reset: bool) -> anyhow::Result<()> {
    535     step("Run Cyclos online harness tests");
    536 
    537     step("Prepare db");
    538     let pool = dbinit(config, reset).await?;
    539     let http_client = http_client::client();
    540     setup::setup(config, reset, &http_client).await?;
    541     let cfg = HarnessCfg::parse(config)?;
    542     let wire = Client {
    543         client: &http_client,
    544         api_url: &cfg.worker.host.api_url,
    545         auth: &CyclosAuth::Basic {
    546             username: cfg.worker.host.username,
    547             password: cfg.worker.host.password,
    548         },
    549     };
    550     let client = Client {
    551         client: &http_client,
    552         api_url: &cfg.worker.host.api_url,
    553         auth: &CyclosAuth::Basic {
    554             username: cfg.username,
    555             password: cfg.password,
    556         },
    557     };
    558 
    559     let harness = Harness {
    560         pool: &pool,
    561         client_payto: client
    562             .whoami()
    563             .await
    564             .unwrap()
    565             .payto(cfg.worker.root.clone()),
    566         wire_payto: wire.whoami().await.unwrap().payto(cfg.worker.root.clone()),
    567         client,
    568         wire,
    569         currency: cfg.worker.currency,
    570         root: cfg.worker.root,
    571         payment_type_id: *cfg.worker.payment_type_id,
    572         account_type_id: *cfg.worker.account_type_id,
    573     };
    574 
    575     step("Warmup worker");
    576     let _worker_task = {
    577         let client = http_client.clone();
    578         let pool = pool.clone();
    579         let config = config.clone();
    580         tokio::spawn(async move { run_worker(&config, &pool, &client, false).await })
    581     };
    582     tokio::time::sleep(Duration::from_secs(5)).await;
    583     let now = Timestamp::now();
    584     let balance = &mut Balances::new(&harness).await;
    585 
    586     step("Test incoming transactions");
    587     let taler_amount = decimal("3");
    588     let malformed_amount = decimal("4");
    589     let reserve_pub = EddsaPublicKey::rand();
    590     harness
    591         .client_send(&format!("Taler {reserve_pub}"), taler_amount)
    592         .await;
    593     harness
    594         .client_send(&format!("Malformed test {now}"), malformed_amount)
    595         .await;
    596     balance.expect_add(taler_amount).await;
    597     harness.expect_incoming(reserve_pub).await;
    598 
    599     step("Test outgoing transactions");
    600     let self_amount = decimal("1");
    601     let taler_amount = decimal("2");
    602 
    603     let transfer_self = harness
    604         .custom_transfer(
    605             self_amount,
    606             *harness.wire_payto.id,
    607             &harness.wire_payto.name,
    608         )
    609         .await;
    610     let transfer_id = harness.transfer(taler_amount).await;
    611     balance.expect_sub(taler_amount).await;
    612     harness
    613         .expect_transfer_status(
    614             transfer_self,
    615             TransferState::permanent_failure,
    616             Some("permissionDenied - The operation was denied because a required permission was not granted"),
    617         )
    618         .await;
    619     harness
    620         .expect_transfer_status(transfer_id, TransferState::success, None)
    621         .await;
    622 
    623     step("Finish");
    624     tokio::time::sleep(Duration::from_secs(5)).await;
    625     balance.expect_add(Decimal::ZERO).await;
    626 
    627     Ok(())
    628 }
    629 
    630 fn main() {
    631     let args = Args::parse();
    632     taler_main(CONFIG_SOURCE, args.common, async |cfg| match args.cmd {
    633         Command::Logic { reset } => logic_harness(cfg, reset).await,
    634         Command::Online { reset } => online_harness(cfg, reset).await,
    635     });
    636 }