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 err, 59 Error::ErrAgentClosed, 60 "closed agent should return <{}>, but got <{}>", 61 Error::ErrAgentClosed, 62 err, 63 ); 64 } else { 65 panic!("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 err, 82 Error::ErrTransactionExists, 83 "duplicate start should return <{}>, got <{}>", 84 Error::ErrTransactionExists, 85 err, 86 ); 87 } else { 88 panic!("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 err, 97 Error::ErrAgentClosed, 98 "start on closed agent should return <{}>, got <{}>", 99 Error::ErrAgentClosed, 100 err, 101 ); 102 } else { 103 panic!("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 err, 110 Error::ErrAgentClosed, 111 "SetHandler on closed agent should return <{}>, got <{}>", 112 Error::ErrAgentClosed, 113 err, 114 ); 115 } else { 116 panic!("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 err, 131 Error::ErrTransactionNotExists, 132 "unexpected error: {}, should be {}", 133 Error::ErrTransactionNotExists, 134 err, 135 ); 136 } else { 137 panic!("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!( 152 err, 153 Error::ErrTransactionStopped, 154 "unexpected error: {}, should be {}", 155 err, 156 Error::ErrTransactionStopped 157 ); 158 }else{ 159 panic!("expected error, got ok"); 160 } 161 } 162 _ = timeout.as_mut() => panic!("timed out"), 163 } 164 165 a.close()?; 166 167 let result = a.close(); 168 if let Err(err) = result { 169 assert_eq!( 170 err, 171 Error::ErrAgentClosed, 172 "a.Close returned {} instead of {}", 173 Error::ErrAgentClosed, 174 err, 175 ); 176 } else { 177 panic!("expected error, but got ok"); 178 } 179 180 let result = a.stop(TransactionId::default()); 181 if let Err(err) = result { 182 assert_eq!( 183 err, 184 Error::ErrAgentClosed, 185 "unexpected error: {}, should be {}", 186 Error::ErrAgentClosed, 187 err, 188 ); 189 } else { 190 panic!("expected error, but got ok"); 191 } 192 193 Ok(()) 194 } 195