1 use super::*;
2 use crate::error::*;
3
4 use std::ops::Add;
5 use tokio::time::Duration;
6
7 #[tokio::test]
test_agent_process_in_transaction() -> Result<()>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]
test_agent_process() -> Result<()>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]
test_agent_start() -> Result<()>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]
test_agent_stop() -> Result<()>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