1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
|
//! Example: tests for the timing features in tor-rtcompat.
use tor_rtcompat::test_with_all_runtimes;
use tor_rtcompat::{SleepProvider, SleepProviderExt, Timeout, TimeoutError};
use tor_rtmock::time::MockSleepProvider;
use futures::FutureExt;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, SystemTime};
use tor_async_utils::oneshot;
#[test]
fn timeouts() {
fn setup() -> (
MockSleepProvider,
oneshot::Sender<()>,
Timeout<oneshot::Receiver<()>, tor_rtmock::time::Sleeping>,
) {
let start = SystemTime::now();
let (send, recv) = oneshot::channel::<()>();
let mock_sp = MockSleepProvider::new(start);
let ten_min = Duration::new(10 * 60, 0);
let timeout_future = mock_sp.timeout(ten_min, recv);
(mock_sp, send, timeout_future)
}
// The timeout occurs.
test_with_all_runtimes!(|_| async {
let (mock_sp, _send, timeout_future) = setup();
mock_sp.advance(Duration::new(3600, 0)).await;
assert!(matches!(timeout_future.await, Err(TimeoutError)));
});
// The data is ready immediately.
test_with_all_runtimes!(|_| async {
let (_, send, timeout_future) = setup();
send.send(()).unwrap();
assert_eq!(timeout_future.await, Ok(Ok(())));
});
// The data is ready after a little while
test_with_all_runtimes!(|_| async {
let (mock_sp, send, timeout_future) = setup();
mock_sp.advance(Duration::new(10, 0)).await;
send.send(()).unwrap();
assert_eq!(timeout_future.await, Ok(Ok(())));
});
// The data is ready _and_ the timeout occurs.
test_with_all_runtimes!(|_| async {
let (mock_sp, send, timeout_future) = setup();
send.send(()).unwrap();
mock_sp.advance(Duration::new(3600, 0)).await;
assert_eq!(timeout_future.await, Ok(Ok(())));
});
// Make sure that nothing happens too early.
test_with_all_runtimes!(|_| async {
let (mock_sp, _send, timeout_future) = setup();
mock_sp.advance(Duration::new(300, 0)).await;
assert_eq!(timeout_future.now_or_never(), None);
});
}
fn start() -> SystemTime {
use humantime::parse_rfc3339;
parse_rfc3339("2009-04-13T09:13:00Z").unwrap()
}
const ONE_DAY: Duration = Duration::from_secs(86400);
#[test]
fn wallclock_simple() {
// Simple case: time goes by.
test_with_all_runtimes!(|_| async {
let mock_sp = MockSleepProvider::new(start());
let b = AtomicBool::new(false);
futures::join!(
async {
mock_sp.sleep_until_wallclock(start() + ONE_DAY).await;
b.store(true, Ordering::SeqCst);
},
async {
while mock_sp.wallclock() < start() + ONE_DAY {
assert!(!b.load(Ordering::SeqCst));
mock_sp.advance(Duration::new(413, 0)).await;
}
}
);
assert!(b.load(Ordering::SeqCst));
});
}
#[test]
fn wallclock_early() {
// Simple case 2: time goes by, but not enough of it.
test_with_all_runtimes!(|_| async {
let mock_sp = MockSleepProvider::new(start());
let b = AtomicBool::new(false);
let (send, mut recv) = oneshot::channel();
futures::join!(
async {
let mut sleep = mock_sp.sleep_until_wallclock(start() + ONE_DAY).fuse();
futures::select! {
_ = sleep => b.store(true, Ordering::SeqCst),
_ = recv => (),
};
},
async {
while mock_sp.wallclock() < start() + (ONE_DAY / 2) {
assert!(!b.load(Ordering::SeqCst));
mock_sp.advance(Duration::new(413, 0)).await;
}
send.send(()).unwrap();
}
);
assert!(!b.load(Ordering::SeqCst));
});
}
#[test]
fn wallclock_jump_forward() {
// Clock jumps forward, so event triggers.
test_with_all_runtimes!(|_| async {
let mock_sp = MockSleepProvider::new(start());
let b = AtomicBool::new(false);
let i1 = mock_sp.now();
futures::join!(
async {
mock_sp.sleep_until_wallclock(start() + ONE_DAY).await;
b.store(true, Ordering::SeqCst);
},
async {
mock_sp.jump_to(start() + ONE_DAY);
mock_sp.advance(Duration::new(1000, 0)).await; // have to rest some.
}
);
assert!(b.load(Ordering::SeqCst));
let i2 = mock_sp.now();
assert!(i2 - i1 < ONE_DAY);
});
}
#[test]
fn wallclock_jump_backwards() {
// Clock jumps backward, so event does not trigger early.
test_with_all_runtimes!(|_| async {
let mock_sp = MockSleepProvider::new(start());
let b = AtomicBool::new(false);
let (send, mut recv) = oneshot::channel();
let i1 = mock_sp.now();
futures::join!(
async {
let mut sleep = mock_sp.sleep_until_wallclock(start() + ONE_DAY).fuse();
futures::select! {
_ = sleep => b.store(true, Ordering::SeqCst),
_ = recv => (),
};
},
async {
mock_sp.jump_to(start() - ONE_DAY);
let mut elapsed = Duration::new(0, 0);
while elapsed < (3 * ONE_DAY) / 2 {
assert!(!b.load(Ordering::SeqCst));
mock_sp.advance(Duration::new(413, 0)).await;
elapsed += Duration::new(413, 0);
}
send.send(()).unwrap();
}
);
assert!(!b.load(Ordering::SeqCst));
let i2 = mock_sp.now();
assert!(i2 - i1 > ONE_DAY);
assert!(mock_sp.wallclock() < start() + ONE_DAY);
});
}
|
