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
|
use axum::{
extract::{Json, Path, State},
http::StatusCode,
};
use futures::stream::StreamExt;
use crate::{
channel::routes,
events::app,
repo::channel,
test::fixtures::{self, future::Immediately as _},
};
#[tokio::test]
async fn channel_exists() {
// Set up the environment
let app = fixtures::scratch_app().await;
let sender = fixtures::login::create(&app).await;
let channel = fixtures::channel::create(&app).await;
// Call the endpoint
let sent_at = fixtures::now();
let request = routes::SendRequest {
message: fixtures::message::propose(),
};
let status = routes::on_send(
State(app.clone()),
Path(channel.id.clone()),
sent_at.clone(),
sender.clone(),
Json(request.clone()),
)
.await
.expect("sending to a valid channel");
// Verify the structure of the response
assert_eq!(StatusCode::ACCEPTED, status);
// Verify the semantics
let subscribed_at = fixtures::now();
let mut events = app
.events()
.subscribe(&channel.id, &subscribed_at, None)
.await
.expect("subscribing to a valid channel");
let event = events
.next()
.immediately()
.await
.expect("event received by subscribers");
assert_eq!(request.message, event.body);
assert_eq!(sender, event.sender);
assert_eq!(*sent_at, event.sent_at);
}
#[tokio::test]
async fn messages_in_order() {
// Set up the environment
let app = fixtures::scratch_app().await;
let sender = fixtures::login::create(&app).await;
let channel = fixtures::channel::create(&app).await;
// Call the endpoint (twice)
let requests = vec![
(
fixtures::now(),
routes::SendRequest {
message: fixtures::message::propose(),
},
),
(
fixtures::now(),
routes::SendRequest {
message: fixtures::message::propose(),
},
),
];
for (sent_at, request) in &requests {
routes::on_send(
State(app.clone()),
Path(channel.id.clone()),
sent_at.clone(),
sender.clone(),
Json(request.clone()),
)
.await
.expect("sending to a valid channel");
}
// Verify the semantics
let subscribed_at = fixtures::now();
let events = app
.events()
.subscribe(&channel.id, &subscribed_at, None)
.await
.expect("subscribing to a valid channel")
.take(requests.len());
let events = events.collect::<Vec<_>>().immediately().await;
for ((sent_at, request), event) in requests.into_iter().zip(events) {
assert_eq!(request.message, event.body);
assert_eq!(sender, event.sender);
assert_eq!(*sent_at, event.sent_at);
}
}
#[tokio::test]
async fn nonexistent_channel() {
// Set up the environment
let app = fixtures::scratch_app().await;
let login = fixtures::login::create(&app).await;
// Call the endpoint
let sent_at = fixtures::now();
let channel = channel::Id::generate();
let request = routes::SendRequest {
message: fixtures::message::propose(),
};
let routes::ErrorResponse(error) = routes::on_send(
State(app),
Path(channel.clone()),
sent_at,
login,
Json(request),
)
.await
.expect_err("sending to a nonexistent channel");
// Verify the structure of the response
assert!(matches!(
error,
app::EventsError::ChannelNotFound(error_channel) if channel == error_channel
));
}
|
