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 }