1 use super::*; 2 use crate::error::*; 3 4 use std::ops::Add; 5 use tokio::time::Duration; 6 7 #[tokio::test] 8 async fn test_agent_process_in_transaction() -> Result<()> { 9 let mut m = Message::new(); 10 let (handler_tx, mut handler_rx) = tokio::sync::mpsc::unbounded_channel(); 11 let mut a = Agent::new(Some(Arc::new(handler_tx))); 12 m.transaction_id = TransactionId([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]); 13 a.start(m.transaction_id, Instant::now())?; 14 a.process(m)?; 15 a.close()?; 16 17 while let Some(e) = handler_rx.recv().await { 18 assert!(e.event_body.is_ok(), "got error: {:?}", e.event_body); 19 20 let tid = TransactionId([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]); 21 assert_eq!( 22 e.event_body.as_ref().unwrap().transaction_id, 23 tid, 24 "{:?} (got) != {:?} (expected)", 25 e.event_body.as_ref().unwrap().transaction_id, 26 tid 27 ); 28 } 29 30 Ok(()) 31 } 32 33 #[tokio::test] 34 async fn test_agent_process() -> Result<()> { 35 let mut m = Message::new(); 36 let (handler_tx, mut handler_rx) = tokio::sync::mpsc::unbounded_channel(); 37 let mut a = Agent::new(Some(Arc::new(handler_tx))); 38 m.transaction_id = TransactionId([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]); 39 a.process(m.clone())?; 40 a.close()?; 41 42 while let Some(e) = handler_rx.recv().await { 43 assert!(e.event_body.is_ok(), "got error: {:?}", e.event_body); 44 45 let tid = TransactionId([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]); 46 assert_eq!( 47 e.event_body.as_ref().unwrap().transaction_id, 48 tid, 49 "{:?} (got) != {:?} (expected)", 50 e.event_body.as_ref().unwrap().transaction_id, 51 tid 52 ); 53 } 54 55 let result = a.process(m); 56 if let Err(err) = result { 57 assert_eq!( 58 Error::ErrAgentClosed, 59 err, 60 "closed agent should return <{}>, but got <{}>", 61 Error::ErrAgentClosed, 62 err, 63 ); 64 } else { 65 assert!(false, "expected error, but got ok"); 66 } 67 68 Ok(()) 69 } 70 71 #[test] 72 fn test_agent_start() -> Result<()> { 73 let mut a = Agent::new(noop_handler()); 74 let id = TransactionId::new(); 75 let deadline = Instant::now().add(Duration::from_secs(3600)); 76 a.start(id, deadline)?; 77 78 let result = a.start(id, deadline); 79 if let Err(err) = result { 80 assert_eq!( 81 Error::ErrTransactionExists, 82 err, 83 "duplicate start should return <{}>, got <{}>", 84 Error::ErrTransactionExists, 85 err, 86 ); 87 } else { 88 assert!(false, "expected error, but got ok"); 89 } 90 a.close()?; 91 92 let id = TransactionId::new(); 93 let result = a.start(id, deadline); 94 if let Err(err) = result { 95 assert_eq!( 96 Error::ErrAgentClosed, 97 err, 98 "start on closed agent should return <{}>, got <{}>", 99 Error::ErrAgentClosed, 100 err, 101 ); 102 } else { 103 assert!(false, "expected error, but got ok"); 104 } 105 106 let result = a.set_handler(noop_handler()); 107 if let Err(err) = result { 108 assert_eq!( 109 Error::ErrAgentClosed, 110 err, 111 "SetHandler on closed agent should return <{}>, got <{}>", 112 Error::ErrAgentClosed, 113 err, 114 ); 115 } else { 116 assert!(false, "expected error, but got ok"); 117 } 118 119 Ok(()) 120 } 121 122 #[tokio::test] 123 async fn test_agent_stop() -> Result<()> { 124 let (handler_tx, mut handler_rx) = tokio::sync::mpsc::unbounded_channel(); 125 let mut a = Agent::new(Some(Arc::new(handler_tx))); 126 127 let result = a.stop(TransactionId::default()); 128 if let Err(err) = result { 129 assert_eq!( 130 Error::ErrTransactionNotExists, 131 err, 132 "unexpected error: {}, should be {}", 133 Error::ErrTransactionNotExists, 134 err, 135 ); 136 } else { 137 assert!(false, "expected error, but got ok"); 138 } 139 140 let id = TransactionId::new(); 141 let deadline = Instant::now().add(Duration::from_millis(200)); 142 a.start(id, deadline)?; 143 a.stop(id)?; 144 145 let timeout = tokio::time::sleep(Duration::from_millis(400)); 146 tokio::pin!(timeout); 147 148 tokio::select! { 149 evt = handler_rx.recv() => { 150 if let Err(err) = evt.unwrap().event_body{ 151 assert_eq!(Error::ErrTransactionStopped,err, 152 "unexpected error: {}, should be {}", 153 err, Error::ErrTransactionStopped); 154 }else{ 155 assert!(false, "expected error, got ok"); 156 } 157 } 158 _ = timeout.as_mut() => assert!(false, "timed out"), 159 } 160 161 a.close()?; 162 163 let result = a.close(); 164 if let Err(err) = result { 165 assert_eq!( 166 Error::ErrAgentClosed, 167 err, 168 "a.Close returned {} instead of {}", 169 Error::ErrAgentClosed, 170 err, 171 ); 172 } else { 173 assert!(false, "expected error, but got ok"); 174 } 175 176 let result = a.stop(TransactionId::default()); 177 if let Err(err) = result { 178 assert_eq!( 179 Error::ErrAgentClosed, 180 err, 181 "unexpected error: {}, should be {}", 182 Error::ErrAgentClosed, 183 err, 184 ); 185 } else { 186 assert!(false, "expected error, but got ok"); 187 } 188 189 Ok(()) 190 } 191